- IT (1442)
Notice
Recent Posts
Recent Comments
반응형
오늘도 공부
"Ask-the-Web" Agent (Perplexity 스타일) - 완벽 핸즈온 가이드 #3 본문
반응형
🎯 학습 목표
- 웹 검색 기능을 가진 AI 에이전트 구축
- LangChain Agent 프레임워크 이해
- Tool/Function Calling 마스터
- 멀티스텝 추론 (ReAct 패턴) 구현
- 실시간 웹 정보를 활용한 답변 생성
📋 사전 준비
1. 개발 환경 설정
# 새 프로젝트 디렉토리
mkdir ask-the-web-agent
cd ask-the-web-agent
# 가상환경 생성
python -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
# 필수 패키지 설치
pip install openai anthropic streamlit python-dotenv
pip install langchain langchain-openai langchain-community
pip install duckduckgo-search wikipedia beautifulsoup4
pip install requests lxml playwright
pip install tavily-python # 고급 웹 검색 API
# Playwright 브라우저 설치 (웹 스크래핑용)
playwright install chromium
2. API 키 발급
- Tavily API: https://tavily.com (무료 1000회/월)
- SerpAPI (선택): https://serpapi.com
- OpenAI 또는 Anthropic API (이미 있음)
3. 프로젝트 구조
mkdir -p src tools utils data
touch .env app.py requirements.txt
ask-the-web-agent/
├── src/
│ ├── agent.py # 메인 에이전트 로직
│ ├── react_agent.py # ReAct 패턴 구현
│ ├── chain_agent.py # 체인 기반 에이전트
│ └── orchestrator.py # 멀티 에이전트 조율
├── tools/
│ ├── search_tools.py # 검색 도구들
│ ├── scraper_tools.py # 웹 스크래핑
│ ├── calculator_tools.py # 계산기
│ └── wikipedia_tools.py # 위키피디아
├── utils/
│ ├── prompt_templates.py # 프롬프트 템플릿
│ └── response_formatter.py # 응답 포맷팅
├── app.py # Streamlit UI
└── .env
.env 파일
OPENAI_API_KEY=your_openai_key
ANTHROPIC_API_KEY=your_anthropic_key
TAVILY_API_KEY=your_tavily_key # https://tavily.com
🔧 Step 1: 기본 검색 도구 구축 (25분)
tools/search_tools.py
import os
from typing import List, Dict, Optional
from tavily import TavilyClient
from duckduckgo_search import DDGS
import requests
from bs4 import BeautifulSoup
class SearchTools:
"""다양한 검색 도구 모음"""
def __init__(self):
self.tavily_api_key = os.getenv("TAVILY_API_KEY")
if self.tavily_api_key:
self.tavily_client = TavilyClient(api_key=self.tavily_api_key)
else:
self.tavily_client = None
print("⚠️ Tavily API 키가 없습니다. DuckDuckGo를 사용합니다.")
def search_tavily(self, query: str, max_results: int = 5) -> List[Dict]:
"""Tavily 검색 (고품질, 빠름)"""
if not self.tavily_client:
return []
try:
response = self.tavily_client.search(
query=query,
max_results=max_results,
search_depth="advanced" # basic or advanced
)
results = []
for result in response.get('results', []):
results.append({
'title': result.get('title', ''),
'url': result.get('url', ''),
'content': result.get('content', ''),
'score': result.get('score', 0)
})
return results
except Exception as e:
print(f"Tavily 검색 오류: {e}")
return []
def search_duckduckgo(self, query: str, max_results: int = 5) -> List[Dict]:
"""DuckDuckGo 검색 (무료, API 키 불필요)"""
try:
ddgs = DDGS()
results = []
for result in ddgs.text(query, max_results=max_results):
results.append({
'title': result.get('title', ''),
'url': result.get('href', ''),
'content': result.get('body', ''),
'score': 0.5 # DuckDuckGo는 점수를 제공하지 않음
})
return results
except Exception as e:
print(f"DuckDuckGo 검색 오류: {e}")
return []
def search(self, query: str, max_results: int = 5) -> List[Dict]:
"""자동으로 사용 가능한 검색 엔진 선택"""
if self.tavily_client:
results = self.search_tavily(query, max_results)
if results:
return results
# Tavily 실패 시 DuckDuckGo 사용
return self.search_duckduckgo(query, max_results)
def get_content_from_url(self, url: str) -> str:
"""URL에서 텍스트 콘텐츠 추출"""
try:
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}
response = requests.get(url, headers=headers, timeout=10)
response.raise_for_status()
soup = BeautifulSoup(response.text, 'html.parser')
# 스크립트, 스타일 제거
for script in soup(["script", "style"]):
script.decompose()
# 텍스트 추출
text = soup.get_text()
lines = (line.strip() for line in text.splitlines())
chunks = (phrase.strip() for line in lines for phrase in line.split(" "))
text = ' '.join(chunk for chunk in chunks if chunk)
# 길이 제한 (첫 3000자)
return text[:3000]
except Exception as e:
return f"URL 접근 오류: {str(e)}"
# 테스트
if __name__ == "__main__":
search_tools = SearchTools()
query = "2024년 노벨 물리학상"
print(f"🔍 검색: {query}\n")
results = search_tools.search(query, max_results=3)
for i, result in enumerate(results, 1):
print(f"{i}. {result['title']}")
print(f" URL: {result['url']}")
print(f" 내용: {result['content'][:100]}...")
print(f" 점수: {result['score']}\n")
tools/wikipedia_tools.py
import wikipedia
class WikipediaTools:
"""위키피디아 검색 도구"""
def __init__(self, language: str = "ko"):
"""
Args:
language: 위키피디아 언어 코드 (ko, en, ja, etc.)
"""
wikipedia.set_lang(language)
self.language = language
def search(self, query: str, max_results: int = 3) -> List[str]:
"""위키피디아 제목 검색"""
try:
return wikipedia.search(query, results=max_results)
except Exception as e:
print(f"위키피디아 검색 오류: {e}")
return []
def get_summary(self, title: str, sentences: int = 3) -> str:
"""위키피디아 페이지 요약"""
try:
return wikipedia.summary(title, sentences=sentences)
except wikipedia.exceptions.DisambiguationError as e:
# 동음이의어 페이지인 경우 첫 번째 옵션 사용
return wikipedia.summary(e.options[0], sentences=sentences)
except wikipedia.exceptions.PageError:
return f"'{title}' 페이지를 찾을 수 없습니다."
except Exception as e:
return f"오류: {str(e)}"
def get_page(self, title: str) -> Dict:
"""전체 페이지 정보"""
try:
page = wikipedia.page(title)
return {
'title': page.title,
'url': page.url,
'content': page.content[:2000], # 처음 2000자
'summary': page.summary
}
except Exception as e:
return {'error': str(e)}
# 테스트
if __name__ == "__main__":
wiki = WikipediaTools(language="ko")
query = "인공지능"
print(f"🔍 위키피디아 검색: {query}\n")
# 1. 제목 검색
titles = wiki.search(query)
print("검색 결과:")
for title in titles:
print(f" - {title}")
# 2. 요약 가져오기
if titles:
print(f"\n📄 '{titles[0]}' 요약:")
summary = wiki.get_summary(titles[0])
print(summary)
tools/calculator_tools.py
import math
import re
from typing import Union
class CalculatorTools:
"""계산 도구"""
@staticmethod
def calculate(expression: str) -> Union[float, str]:
"""수식 계산"""
try:
# 보안을 위해 허용된 문자만 사용
allowed_chars = r'[0-9+\-*/().\s]'
if not re.match(f'^{allowed_chars}+$', expression):
return "오류: 허용되지 않은 문자가 포함되어 있습니다."
# eval 대신 안전한 계산
result = eval(expression, {"__builtins__": {}}, {})
return float(result)
except ZeroDivisionError:
return "오류: 0으로 나눌 수 없습니다."
except Exception as e:
return f"계산 오류: {str(e)}"
@staticmethod
def advanced_calculate(expression: str) -> Union[float, str]:
"""고급 수학 함수 지원"""
try:
# math 모듈 함수 허용
safe_dict = {
'sqrt': math.sqrt,
'sin': math.sin,
'cos': math.cos,
'tan': math.tan,
'log': math.log,
'log10': math.log10,
'exp': math.exp,
'pi': math.pi,
'e': math.e
}
result = eval(expression, {"__builtins__": {}}, safe_dict)
return float(result)
except Exception as e:
return f"계산 오류: {str(e)}"
# 테스트
if __name__ == "__main__":
calc = CalculatorTools()
expressions = [
"2 + 2",
"100 * 0.15",
"(500 - 200) / 3",
"sqrt(16)",
"sin(pi/2)"
]
for expr in expressions:
result = calc.advanced_calculate(expr)
print(f"{expr} = {result}")
🤖 Step 2: ReAct 에이전트 구현 (35분)
utils/prompt_templates.py
class AgentPrompts:
"""에이전트 프롬프트 템플릿"""
REACT_SYSTEM = """당신은 ReAct (Reasoning + Acting) 패턴을 따르는 AI 에이전트입니다.
사용 가능한 도구:
{tools}
작업 수행 방식:
1. Thought: 현재 상황을 분석하고 다음에 해야 할 일을 생각합니다
2. Action: 사용할 도구와 입력을 선택합니다
3. Observation: 도구의 실행 결과를 확인합니다
4. ... (필요시 1-3을 반복)
5. Final Answer: 최종 답변을 제공합니다
형식:
Thought: [분석 내용]
Action: [도구 이름]
Action Input: [도구 입력값]
Observation: [도구 결과]
... (필요시 반복)
Thought: 이제 최종 답변을 할 수 있습니다
Final Answer: [최종 답변]
중요:
- 반드시 위 형식을 정확히 따라야 합니다
- 추측하지 말고 도구를 사용하여 정확한 정보를 얻으세요
- 최종 답변은 "Final Answer:" 뒤에 작성하세요"""
REACT_USER = """질문: {question}
Thought:"""
@staticmethod
def get_react_prompt(question: str, tools: List[Dict]) -> str:
"""ReAct 프롬프트 생성"""
tools_desc = "\n".join([
f"- {tool['name']}: {tool['description']}"
for tool in tools
])
system = AgentPrompts.REACT_SYSTEM.format(tools=tools_desc)
user = AgentPrompts.REACT_USER.format(question=question)
return system, user
src/react_agent.py
import os
import re
from typing import List, Dict, Optional, Tuple
from openai import OpenAI
from dotenv import load_dotenv
import sys
sys.path.append('..')
from tools.search_tools import SearchTools
from tools.wikipedia_tools import WikipediaTools
from tools.calculator_tools import CalculatorTools
from utils.prompt_templates import AgentPrompts
load_dotenv()
class ReActAgent:
"""ReAct 패턴 기반 에이전트"""
def __init__(self, model: str = "gpt-4o-mini", max_iterations: int = 5):
self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
self.model = model
self.max_iterations = max_iterations
# 도구 초기화
self.search_tools = SearchTools()
self.wiki_tools = WikipediaTools()
self.calc_tools = CalculatorTools()
# 사용 가능한 도구 정의
self.tools = [
{
'name': 'search',
'description': '웹에서 최신 정보를 검색합니다. 실시간 뉴스, 최신 정보가 필요할 때 사용하세요.',
'function': self.search_tools.search
},
{
'name': 'wikipedia',
'description': '위키피디아에서 백과사전 정보를 검색합니다. 역사, 개념, 인물 등의 정보가 필요할 때 사용하세요.',
'function': self.wiki_tools.get_summary
},
{
'name': 'calculator',
'description': '수학 계산을 수행합니다. 숫자 계산이 필요할 때 사용하세요.',
'function': self.calc_tools.advanced_calculate
},
{
'name': 'get_url_content',
'description': 'URL의 내용을 가져옵니다. 특정 웹페이지의 상세 내용이 필요할 때 사용하세요.',
'function': self.search_tools.get_content_from_url
}
]
def parse_action(self, text: str) -> Optional[Tuple[str, str]]:
"""텍스트에서 Action과 Action Input 파싱"""
# Action 찾기
action_match = re.search(r'Action:\s*(.+?)(?:\n|$)', text)
if not action_match:
return None
action = action_match.group(1).strip()
# Action Input 찾기
input_match = re.search(r'Action Input:\s*(.+?)(?:\n|$)', text, re.DOTALL)
if not input_match:
return None
action_input = input_match.group(1).strip()
return action, action_input
def execute_action(self, action: str, action_input: str) -> str:
"""도구 실행"""
# 도구 찾기
tool = None
for t in self.tools:
if t['name'].lower() == action.lower():
tool = t
break
if not tool:
return f"오류: '{action}' 도구를 찾을 수 없습니다. 사용 가능한 도구: {[t['name'] for t in self.tools]}"
# 도구 실행
try:
if tool['name'] == 'search':
results = tool['function'](action_input, max_results=3)
if not results:
return "검색 결과가 없습니다."
output = ""
for i, result in enumerate(results, 1):
output += f"\n[{i}] {result['title']}\n"
output += f"URL: {result['url']}\n"
output += f"내용: {result['content'][:200]}...\n"
return output
elif tool['name'] == 'wikipedia':
return tool['function'](action_input, sentences=3)
else:
return str(tool['function'](action_input))
except Exception as e:
return f"도구 실행 오류: {str(e)}"
def run(self, question: str, verbose: bool = True) -> Dict:
"""에이전트 실행"""
system_prompt, user_prompt = AgentPrompts.get_react_prompt(question, self.tools)
conversation = system_prompt + "\n\n" + user_prompt
iterations = []
final_answer = None
for iteration in range(self.max_iterations):
if verbose:
print(f"\n{'='*60}")
print(f"Iteration {iteration + 1}")
print(f"{'='*60}")
# LLM 호출
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": conversation}],
temperature=0.0,
max_tokens=1000
)
agent_output = response.choices[0].message.content
if verbose:
print(agent_output)
# Final Answer 체크
if "Final Answer:" in agent_output:
final_answer = agent_output.split("Final Answer:")[-1].strip()
iterations.append({
'iteration': iteration + 1,
'output': agent_output,
'action': None,
'observation': None
})
break
# Action 파싱
parsed = self.parse_action(agent_output)
if not parsed:
# Action이 없으면 다시 시도
conversation += "\n\n" + agent_output + "\n\n(Action과 Action Input을 명시해주세요)"
continue
action, action_input = parsed
# Action 실행
observation = self.execute_action(action, action_input)
iterations.append({
'iteration': iteration + 1,
'output': agent_output,
'action': action,
'action_input': action_input,
'observation': observation
})
# 대화에 Observation 추가
conversation += "\n\n" + agent_output
conversation += f"\nObservation: {observation}\n\nThought:"
if not final_answer:
final_answer = "최대 반복 횟수에 도달했습니다. 답변을 생성할 수 없습니다."
return {
'question': question,
'final_answer': final_answer,
'iterations': iterations,
'num_iterations': len(iterations)
}
# 테스트
if __name__ == "__main__":
agent = ReActAgent(max_iterations=5)
test_questions = [
"2024년 노벨 물리학상 수상자는 누구인가요?",
"서울의 인구는 얼마이고, 부산 인구의 몇 배인가요?",
"양자컴퓨터란 무엇인가요?"
]
for question in test_questions:
print(f"\n\n{'#'*60}")
print(f"질문: {question}")
print(f"{'#'*60}")
result = agent.run(question, verbose=True)
print(f"\n{'='*60}")
print(f"최종 답변:")
print(f"{'='*60}")
print(result['final_answer'])
print(f"\n총 {result['num_iterations']}번의 반복")
🔗 Step 3: LangChain 에이전트 (더 간단한 방법) (25분)
src/agent.py
import os
from typing import List
from langchain.agents import Tool, AgentExecutor, create_react_agent
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_community.utilities import WikipediaAPIWrapper
from dotenv import load_dotenv
import sys
sys.path.append('..')
from tools.calculator_tools import CalculatorTools
load_dotenv()
class LangChainAgent:
"""LangChain 기반 에이전트 (더 간단)"""
def __init__(self, model: str = "gpt-4o-mini", verbose: bool = True):
self.llm = ChatOpenAI(
model=model,
temperature=0,
api_key=os.getenv("OPENAI_API_KEY")
)
self.verbose = verbose
# 도구 정의
self.tools = self._setup_tools()
# 에이전트 생성
self.agent = self._create_agent()
def _setup_tools(self) -> List[Tool]:
"""도구 설정"""
# 검색 도구
search = DuckDuckGoSearchRun()
# 위키피디아 도구
wikipedia = WikipediaAPIWrapper(lang="ko")
# 계산기 도구
calculator = CalculatorTools()
tools = [
Tool(
name="Search",
func=search.run,
description="웹에서 최신 정보를 검색할 때 유용합니다. 뉴스, 현재 이벤트, 최신 데이터 등을 찾을 때 사용하세요."
),
Tool(
name="Wikipedia",
func=wikipedia.run,
description="백과사전 정보를 검색할 때 유용합니다. 역사적 사실, 과학 개념, 유명 인물 등에 대한 정보를 찾을 때 사용하세요."
),
Tool(
name="Calculator",
func=calculator.advanced_calculate,
description="수학 계산이 필요할 때 사용합니다. 산술 연산, 제곱근, 삼각함수 등을 계산할 수 있습니다."
)
]
return tools
def _create_agent(self):
"""ReAct 에이전트 생성"""
# 프롬프트 템플릿
template = """당신은 질문에 답하기 위해 도구를 사용할 수 있는 AI 어시스턴트입니다.
다음 도구들을 사용할 수 있습니다:
{tools}
다음 형식을 사용하세요:
Question: 답해야 할 질문
Thought: 무엇을 해야 할지 생각
Action: 수행할 작업, [{tool_names}] 중 하나여야 합니다
Action Input: 작업에 대한 입력
Observation: 작업의 결과
... (이 Thought/Action/Action Input/Observation을 여러 번 반복할 수 있습니다)
Thought: 이제 최종 답변을 알았습니다
Final Answer: 원래 질문에 대한 최종 답변
시작!
Question: {input}
Thought: {agent_scratchpad}"""
prompt = PromptTemplate.from_template(template)
# 에이전트 생성
agent = create_react_agent(
llm=self.llm,
tools=self.tools,
prompt=prompt
)
# 실행기 생성
agent_executor = AgentExecutor(
agent=agent,
tools=self.tools,
verbose=self.verbose,
max_iterations=5,
handle_parsing_errors=True
)
return agent_executor
def run(self, question: str) -> Dict:
"""에이전트 실행"""
try:
result = self.agent.invoke({"input": question})
return {
'question': question,
'answer': result['output'],
'success': True
}
except Exception as e:
return {
'question': question,
'answer': f"오류 발생: {str(e)}",
'success': False
}
# 테스트
if __name__ == "__main__":
agent = LangChainAgent(verbose=True)
questions = [
"현재 미국 대통령은 누구인가요?",
"파이썬이 언제 만들어졌고, 만든 사람은 누구인가요?",
"100달러를 원화로 환산하면 얼마인가요? (환율 1400원 기준)"
]
for question in questions:
print(f"\n{'='*60}")
print(f"질문: {question}")
print(f"{'='*60}\n")
result = agent.run(question)
print(f"\n답변: {result['answer']}\n")
🎨 Step 4: Streamlit UI - Perplexity 스타일 (30분)
app.py
import streamlit as st
import sys
sys.path.append('src')
sys.path.append('tools')
from agent import LangChainAgent
from react_agent import ReActAgent
import time
# 페이지 설정
st.set_page_config(
page_title="Ask-the-Web Agent",
page_icon="🔍",
layout="wide"
)
# CSS 스타일
st.markdown("""
<style>
.main-header {
font-size: 2.5rem;
font-weight: bold;
text-align: center;
margin-bottom: 1rem;
background: linear-gradient(90deg, #667eea 0%, #764ba2 100%);
-webkit-background-clip: text;
-webkit-text-fill-color: transparent;
}
.search-box {
border: 2px solid #667eea;
border-radius: 10px;
padding: 20px;
margin: 20px 0;
background-color: #f8f9fa;
}
.source-card {
border-left: 4px solid #667eea;
padding: 15px;
margin: 10px 0;
background-color: #f8f9fa;
border-radius: 5px;
}
.answer-box {
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
padding: 25px;
border-radius: 10px;
margin: 20px 0;
}
.thinking-step {
background-color: #fff9e6;
border-left: 4px solid #ffc107;
padding: 15px;
margin: 10px 0;
border-radius: 5px;
}
</style>
""", unsafe_allow_html=True)
# 세션 상태 초기화
if 'agent' not in st.session_state:
st.session_state.agent = None
if 'search_history' not in st.session_state:
st.session_state.search_history = []
# 헤더
st.markdown('<h1 class="main-header">🔍 Ask-the-Web Agent</h1>', unsafe_allow_html=True)
st.markdown('<p style="text-align: center; color: #666;">실시간 웹 검색으로 최신 정보를 제공하는 AI 에이전트</p>', unsafe_allow_html=True)
# 사이드바 설정
with st.sidebar:
st.header("⚙️ 설정")
agent_type = st.selectbox(
"에이전트 타입",
["LangChain (간단)", "Custom ReAct (상세)"],
help="LangChain은 간단하고 빠르며, Custom ReAct는 상세한 추론 과정을 보여줍니다"
)
model_choice = st.selectbox(
"LLM 모델",
["gpt-4o-mini", "gpt-4o", "gpt-3.5-turbo"],
help="gpt-4o-mini는 빠르고 저렴하며, gpt-4o는 더 정확합니다"
)
show_thinking = st.checkbox(
"사고 과정 표시",
value=True,
help="에이전트의 추론 과정을 단계별로 보여줍니다"
)
max_iterations = st.slider(
"최대 반복 횟수",
min_value=3,
max_value=10,
value=5,
help="에이전트가 도구를 사용할 수 있는 최대 횟수"
)
# 에이전트 초기화
if st.button("🔄 에이전트 초기화"):
with st.spinner("에이전트 초기화 중..."):
if "LangChain" in agent_type:
st.session_state.agent = LangChainAgent(
model=model_choice,
verbose=show_thinking
)
else:
st.session_state.agent = ReActAgent(
model=model_choice,
max_iterations=max_iterations
)
st.success("✅ 초기화 완료!")
st.markdown("---")
# 검색 히스토리
st.subheader("📜 검색 기록")
if st.session_state.search_history:
for i, item in enumerate(reversed(st.session_state.search_history[-5:]), 1):
with st.expander(f"{i}. {item['question'][:30]}..."):
st.write(f"**답변:** {item['answer'][:100]}...")
st.caption(f"⏱️ {item['time']}")
else:
st.info("아직 검색 기록이 없습니다")
if st.button("🗑️ 기록 삭제"):
st.session_state.search_history = []
st.rerun()
st.markdown("---")
# 예시 질문
st.subheader("💡 예시 질문")
example_questions = [
"2024년 노벨상 수상자는?",
"현재 비트코인 가격은?",
"최신 AI 뉴스는?",
"양자컴퓨터란 무엇인가요?",
"서울과 부산의 인구 비교"
]
for question in example_questions:
if st.button(question, key=f"ex_{question}", use_container_width=True):
st.session_state.current_question = question
# 메인 영역
col1, col2, col3 = st.columns([1, 3, 1])
with col2:
# 검색 입력
question = st.text_input(
"질문을 입력하세요",
placeholder="예: 2024년 노벨 물리학상 수상자는 누구인가요?",
value=st.session_state.get('current_question', ''),
key="question_input"
)
search_button = st.button("🔍 검색", type="primary", use_container_width=True)
# 검색 실행
if search_button and question:
# 에이전트 초기화 확인
if st.session_state.agent is None:
with st.spinner("에이전트 초기화 중..."):
if "LangChain" in agent_type:
st.session_state.agent = LangChainAgent(
model=model_choice,
verbose=show_thinking
)
else:
st.session_state.agent = ReActAgent(
model=model_choice,
max_iterations=max_iterations
)
# 검색 시작
start_time = time.time()
with st.spinner("🤔 답변 생성 중..."):
# 진행 상황 표시
progress_bar = st.progress(0)
status_text = st.empty()
if "LangChain" in agent_type:
# LangChain 에이전트
status_text.text("도구 선택 중...")
progress_bar.progress(33)
result = st.session_state.agent.run(question)
progress_bar.progress(100)
status_text.empty()
progress_bar.empty()
# 답변 표시
st.markdown(f"""
<div class="answer-box">
<h3>💡 답변</h3>
<p>{result['answer']}</p>
</div>
""", unsafe_allow_html=True)
else:
# Custom ReAct 에이전트
result = st.session_state.agent.run(question, verbose=False)
progress_bar.progress(100)
status_text.empty()
progress_bar.empty()
# 사고 과정 표시
if show_thinking and result['iterations']:
st.subheader("🧠 사고 과정")
for iteration in result['iterations']:
with st.expander(f"Step {iteration['iteration']}", expanded=False):
st.markdown(f"""
<div class="thinking-step">
{iteration['output'].replace(chr(10), '<br>')}
</div>
""", unsafe_allow_html=True)
if iteration.get('action'):
st.info(f"🔧 도구: **{iteration['action']}**")
st.code(iteration['action_input'])
st.success(f"📊 결과:\n{iteration['observation'][:300]}...")
# 최종 답변
st.markdown(f"""
<div class="answer-box">
<h3>💡 최종 답변</h3>
<p>{result['final_answer']}</p>
</div>
""", unsafe_allow_html=True)
# 소요 시간
elapsed_time = time.time() - start_time
st.caption(f"⏱️ 소요 시간: {elapsed_time:.2f}초 | 반복: {result.get('num_iterations', 'N/A')}회")
# 기록 저장
st.session_state.search_history.append({
'question': question,
'answer': result.get('answer', result.get('final_answer', '')),
'time': time.strftime('%Y-%m-%d %H:%M:%S')
})
# current_question 초기화
if 'current_question' in st.session_state:
del st.session_state.current_question
# Footer
st.markdown("---")
st.markdown("""
<div style='text-align: center; color: #666;'>
<p>🤖 Powered by ReAct Agent Pattern</p>
<p style='font-size: 0.9em;'>실시간 웹 검색 • 위키피디아 • 계산기</p>
</div>
""", unsafe_allow_html=True)
실행
streamlit run app.py
🎯 Step 5: 멀티 에이전트 오케스트레이션 (고급, 25분)
src/orchestrator.py
import os
from typing import List, Dict
from openai import OpenAI
from dotenv import load_dotenv
import sys
sys.path.append('..')
from tools.search_tools import SearchTools
from tools.wikipedia_tools import WikipediaTools
load_dotenv()
class MultiAgentOrchestrator:
"""여러 전문 에이전트를 조율하는 오케스트레이터"""
def __init__(self):
self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
# 전문 에이전트들
self.agents = {
'researcher': {
'name': '리서처',
'expertise': '웹 검색을 통한 최신 정보 수집',
'tools': SearchTools()
},
'analyst': {
'name': '분석가',
'expertise': '수집된 정보 분석 및 종합',
'tools': None
},
'fact_checker': {
'name': '팩트체커',
'expertise': '정보의 정확성 검증',
'tools': WikipediaTools()
}
}
def route_query(self, question: str) -> str:
"""질문을 분석하여 적절한 에이전트 선택"""
prompt = f"""다음 질문을 분석하여 어떤 에이전트에게 할당할지 결정하세요.
사용 가능한 에이전트:
- researcher: 최신 뉴스, 실시간 정보, 트렌드 등
- analyst: 복잡한 분석, 비교, 추론이 필요한 작업
- fact_checker: 역사적 사실, 백과사전 정보 확인
질문: {question}
가장 적합한 에이전트 하나만 답하세요 (researcher/analyst/fact_checker):"""
response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0
)
agent_choice = response.choices[0].message.content.strip().lower()
if agent_choice not in self.agents:
agent_choice = 'researcher' # 기본값
return agent_choice
def research(self, question: str) -> Dict:
"""리서처 에이전트"""
search_tools = self.agents['researcher']['tools']
results = search_tools.search(question, max_results=3)
# 검색 결과 요약
context = "\n\n".join([
f"출처 {i+1}: {r['title']}\n{r['content']}"
for i, r in enumerate(results)
])
return {
'agent': 'researcher',
'raw_results': results,
'context': context
}
def analyze(self, question: str, context: str) -> str:
"""분석가 에이전트"""
prompt = f"""당신은 정보 분석 전문가입니다. 수집된 정보를 바탕으로 질문에 답변하세요.
수집된 정보:
{context}
질문: {question}
답변 (정확하고 객관적으로):"""
response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0.3,
max_tokens=500
)
return response.choices[0].message.content
def fact_check(self, statement: str) -> Dict:
"""팩트체커 에이전트"""
wiki_tools = self.agents['fact_checker']['tools']
# 위키피디아 검색
search_results = wiki_tools.search(statement, max_results=3)
if not search_results:
return {
'verified': False,
'confidence': 'low',
'note': '검증 불가'
}
# 첫 번째 결과 요약 가져오기
summary = wiki_tools.get_summary(search_results[0], sentences=2)
# LLM으로 검증
prompt = f"""다음 진술이 사실인지 확인해주세요.
진술: {statement}
참고 정보:
{summary}
검증 결과를 JSON 형식으로 답하세요:
{{"verified": true/false, "confidence": "high/medium/low", "explanation": "..."}}"""
response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0
)
return {
'agent': 'fact_checker',
'result': response.choices[0].message.content,
'wiki_summary': summary
}
def orchestrate(self, question: str) -> Dict:
"""전체 프로세스 오케스트레이션"""
# 1. 라우팅
assigned_agent = self.route_query(question)
print(f"📍 질문이 '{self.agents[assigned_agent]['name']}'에게 할당되었습니다")
# 2. 리서치
print("🔍 정보 수집 중...")
research_result = self.research(question)
# 3. 분석
print("🧠 정보 분석 중...")
analysis = self.analyze(question, research_result['context'])
# 4. (선택) 팩트 체크
# fact_check_result = self.fact_check(analysis)
return {
'question': question,
'assigned_agent': assigned_agent,
'sources': research_result['raw_results'],
'answer': analysis
}
# 테스트
if __name__ == "__main__":
orchestrator = MultiAgentOrchestrator()
questions = [
"2024년 가장 인기 있는 프로그래밍 언어는?",
"인공지능의 역사를 간단히 설명해주세요",
"비트코인과 이더리움의 현재 가격 차이는?"
]
for question in questions:
print(f"\n{'='*60}")
print(f"질문: {question}")
print(f"{'='*60}\n")
result = orchestrator.orchestrate(question)
print(f"\n담당 에이전트: {result['assigned_agent']}")
print(f"\n답변:\n{result['answer']}")
print(f"\n참고 출처:")
for i, source in enumerate(result['sources'], 1):
print(f" {i}. {source['title']}")
print(f" {source['url']}")
📊 Step 6: 성능 평가 및 비교 (20분)
src/evaluator.py
import time
from typing import List, Dict
import sys
sys.path.append('..')
from agent import LangChainAgent
from react_agent import ReActAgent
class AgentEvaluator:
"""에이전트 성능 평가"""
def __init__(self):
self.langchain_agent = LangChainAgent(verbose=False)
self.react_agent = ReActAgent(max_iterations=5)
def evaluate_accuracy(self, test_cases: List[Dict]) -> Dict:
"""정확성 평가"""
results = {
'langchain': {'correct': 0, 'total': 0, 'details': []},
'react': {'correct': 0, 'total': 0, 'details': []}
}
for case in test_cases:
question = case['question']
expected_keywords = case['expected_keywords'] # 답변에 포함되어야 할 키워드
print(f"\n테스트 중: {question}")
# LangChain 평가
lc_result = self.langchain_agent.run(question)
lc_correct = any(kw.lower() in lc_result['answer'].lower() for kw in expected_keywords)
results['langchain']['total'] += 1
if lc_correct:
results['langchain']['correct'] += 1
results['langchain']['details'].append({
'question': question,
'answer': lc_result['answer'],
'correct': lc_correct
})
# ReAct 평가
react_result = self.react_agent.run(question, verbose=False)
react_correct = any(kw.lower() in react_result['final_answer'].lower() for kw in expected_keywords)
results['react']['total'] += 1
if react_correct:
results['react']['correct'] += 1
results['react']['details'].append({
'question': question,
'answer': react_result['final_answer'],
'correct': react_correct
})
# 정확도 계산
results['langchain']['accuracy'] = results['langchain']['correct'] / results['langchain']['total']
results['react']['accuracy'] = results['react']['correct'] / results['react']['total']
return results
def evaluate_speed(self, questions: List[str]) -> Dict:
"""속도 평가"""
results = {
'langchain': [],
'react': []
}
for question in questions:
print(f"\n속도 테스트: {question}")
# LangChain
start = time.time()
self.langchain_agent.run(question)
lc_time = time.time() - start
results['langchain'].append(lc_time)
# ReAct
start = time.time()
self.react_agent.run(question, verbose=False)
react_time = time.time() - start
results['react'].append(react_time)
# 평균 계산
results['langchain_avg'] = sum(results['langchain']) / len(results['langchain'])
results['react_avg'] = sum(results['react']) / len(results['react'])
return results
# 실행
if __name__ == "__main__":
evaluator = AgentEvaluator()
# 정확성 테스트
test_cases = [
{
'question': '파이썬은 누가 만들었나요?',
'expected_keywords': ['Guido', 'van Rossum', '귀도']
},
{
'question': '현재 미국 대통령은?',
'expected_keywords': ['Biden', 'Trump', '바이든', '트럼프']
}
]
print("="*60)
print("정확성 평가")
print("="*60)
accuracy_results = evaluator.evaluate_accuracy(test_cases)
print(f"\nLangChain 정확도: {accuracy_results['langchain']['accuracy']:.2%}")
print(f"ReAct 정확도: {accuracy_results['react']['accuracy']:.2%}")
# 속도 테스트
speed_questions = [
'서울의 인구는?',
'양자컴퓨터란?'
]
print("\n" + "="*60)
print("속도 평가")
print("="*60)
speed_results = evaluator.evaluate_speed(speed_questions)
print(f"\nLangChain 평균 속도: {speed_results['langchain_avg']:.2f}초")
print(f"ReAct 평균 속도: {speed_results['react_avg']:.2f}초")
🎓 학습 과제 및 개선 방향
필수 과제 체크리스트
- [ ] 기본 검색 도구 구현 완료
- [ ] ReAct 에이전트 동작 확인
- [ ] LangChain 에이전트 비교
- [ ] Streamlit UI 실행
- [ ] 5개 이상의 질문 테스트
심화 과제
- 병렬 검색: 여러 소스를 동시에 검색
- 캐싱: 이전 검색 결과 재사용
- 스트리밍 응답: 답변을 실시간으로 표시
- 출처 표시: 각 문장마다 출처 링크
- 멀티모달: 이미지 검색 추가
다음 단계
- Project 4: Deep Research (복잡한 추론, Tree of Thoughts)
- 프로덕션 배포: Docker, API 서버화
- 모니터링: 로깅, 성능 추적
반응형