Company Research: end-to-end 멀티에이전트 웹 리서치 예제
Company Research: end-to-end 멀티에이전트 웹 리서치 예제
회사 리서치(또는 경쟁 분석)는 모든 비즈니스 전략의 핵심 부분이에요. 이 예제에서는 이 작업을 처리할 에이전트 팀을 만드는 방법을 보여줄게요. 작업을 에이전트 방식으로 구현하는 방법은 다양하지만, 우리는 순차적(sequential) 접근을 살펴볼게요. 리서치 과정의 각 단계에 대응하는 에이전트를 만들고, 각자 작업을 수행할 도구를 부여하죠.
- 검색 에이전트(Search Agent): 회사에 관한 정보를 웹에서 검색해요. 검색 결과를 가져오는 검색 엔진 API 도구를 사용할 수 있어요.
- 주식 분석 에이전트(Stock Analysis Agent): 금융 데이터 API에서 회사의 주식 정보를 가져와 기본 통계(현재가, 52주 최고가, 52주 최저가 등)를 계산하고, 연초 대비 주식 가격 플롯을 생성해 파일로 저장해요. 금융 데이터 API 도구로 주식 정보를 가져올 수 있어요.
- 보고서 에이전트(Report Agent): 검색·주식 분석 에이전트가 수집한 정보를 바탕으로 보고서를 생성해요.
먼저 필요한 모듈을 import 해볼게요.
출처: 공식문서
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_core.tools import FunctionTool
from autogen_ext.models.openai import OpenAIChatCompletionClient
도구 정의하기
다음으로 에이전트가 작업을 수행할 때 쓸 도구를 정의해요. 회사에 관한 정보를 웹에서 검색하는 Google Search API를 쓰는 google_search와, yfinance 라이브러리로 회사 주식 정보를 가져오는 analyze_stock 함수를 만들게요.
마지막으로 이 함수들을 FunctionTool 클래스로 감싸 에이전트에서 도구로 쓸 수 있게 해요.
참고: google_search 함수는 동작하려면 API 키가 필요해요. 이 노트북과 같은 디렉토리에 .env 파일을 만들고 API 키를 추가하세요.
GOOGLE_SEARCH_ENGINE_ID =xxx
GOOGLE_API_KEY=xxx
필요한 라이브러리도 설치하세요.
pip install yfinance matplotlib pytz numpy pandas python-dotenv requests bs4
#!pip install yfinance matplotlib pytz numpy pandas python-dotenv requests bs4
def google_search(query: str, num_results: int = 2, max_chars: int = 500) -> list: # type: ignore[type-arg]
import os
import time
import requests
from bs4 import BeautifulSoup
from dotenv import load_dotenv
load_dotenv()
api_key = os.getenv("GOOGLE_API_KEY")
search_engine_id = os.getenv("GOOGLE_SEARCH_ENGINE_ID")
if not api_key or not search_engine_id:
raise ValueError("API key or Search Engine ID not found in environment variables")
url = "https://customsearch.googleapis.com/customsearch/v1"
params = {"key": str(api_key), "cx": str(search_engine_id), "q": str(query), "num": str(num_results)}
response = requests.get(url, params=params)
if response.status_code != 200:
print(response.json())
raise Exception(f"Error in API request: {response.status_code}")
results = response.json().get("items", [])
def get_page_content(url: str) -> str:
try:
response = requests.get(url, timeout=10)
soup = BeautifulSoup(response.content, "html.parser")
text = soup.get_text(separator=" ", strip=True)
words = text.split()
content = ""
for word in words:
if len(content) + len(word) + 1 > max_chars:
break
content += " " + word
return content.strip()
except Exception as e:
print(f"Error fetching {url}: {str(e)}")
return ""
enriched_results = []
for item in results:
body = get_page_content(item["link"])
enriched_results.append(
{"title": item["title"], "link": item["link"], "snippet": item["snippet"], "body": body}
)
time.sleep(1) # Be respectful to the servers
return enriched_results
def analyze_stock(ticker: str) -> dict: # type: ignore[type-arg]
import os
from datetime import datetime, timedelta
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import yfinance as yf
from pytz import timezone # type: ignore
stock = yf.Ticker(ticker)
# Get historical data (1 year of data to ensure we have enough for 200-day MA)
end_date = datetime.now(timezone("UTC"))
start_date = end_date - timedelta(days=365)
hist = stock.history(start=start_date, end=end_date)
# Ensure we have data
if hist.empty:
return {"error": "No historical data available for the specified ticker."}
# Compute basic statistics and additional metrics
current_price = stock.info.get("currentPrice", hist["Close"].iloc[-1])
year_high = stock.info.get("fiftyTwoWeekHigh", hist["High"].max())
year_low = stock.info.get("fiftyTwoWeekLow", hist["Low"].min())
# Calculate 50-day and 200-day moving averages
ma_50 = hist["Close"].rolling(window=50).mean().iloc[-1]
ma_200 = hist["Close"].rolling(window=200).mean().iloc[-1]
# Calculate YTD price change and percent change
ytd_start = datetime(end_date.year, 1, 1, tzinfo=timezone("UTC"))
ytd_data = hist.loc[ytd_start:] # type: ignore[misc]
if not ytd_data.empty:
price_change = ytd_data["Close"].iloc[-1] - ytd_data["Close"].iloc[0]
percent_change = (price_change / ytd_data["Close"].iloc[0]) * 100
else:
price_change = percent_change = np.nan
# Determine trend
if pd.notna(ma_50) and pd.notna(ma_200):
if ma_50 > ma_200:
trend = "Upward"
elif ma_50 < ma_200:
trend = "Downward"
else:
trend = "Neutral"
else:
trend = "Insufficient data for trend analysis"
# Calculate volatility (standard deviation of daily returns)
daily_returns = hist["Close"].pct_change().dropna()
volatility = daily_returns.std() * np.sqrt(252) # Annualized volatility
# Create result dictionary
result = {
"ticker": ticker,
"current_price": current_price,
"52_week_high": year_high,
"52_week_low": year_low,
"50_day_ma": ma_50,
"200_day_ma": ma_200,
"ytd_price_change": price_change,
"ytd_percent_change": percent_change,
"trend": trend,
"volatility": volatility,
}
# Convert numpy types to Python native types for better JSON serialization
for key, value in result.items():
if isinstance(value, np.generic):
result[key] = value.item()
# Generate plot
plt.figure(figsize=(12, 6))
plt.plot(hist.index, hist["Close"], label="Close Price")
plt.plot(hist.index, hist["Close"].rolling(window=50).mean(), label="50-day MA")
plt.plot(hist.index, hist["Close"].rolling(window=200).mean(), label="200-day MA")
plt.title(f"{ticker} Stock Price (Past Year)")
plt.xlabel("Date")
plt.ylabel("Price ($)")
plt.legend()
plt.grid(True)
# Save plot to file
os.makedirs("coding", exist_ok=True)
plot_file_path = f"coding/{ticker}_stockprice.png"
plt.savefig(plot_file_path)
print(f"Plot saved as {plot_file_path}")
result["plot_file_path"] = plot_file_path
return result
google_search_tool = FunctionTool(
google_search, description="Search Google for information, returns results with a snippet and body content"
)
stock_analysis_tool = FunctionTool(analyze_stock, description="Analyze stock data and generate a plot")
에이전트 정의하기
다음으로 작업을 수행할 에이전트를 정의해요. 회사에 관한 정보를 웹에서 검색하는 search_agent, 회사 주식 정보를 가져오는 stock_analysis_agent, 그리고 다른 에이전트들이 수집한 정보를 바탕으로 보고서를 생성하는 report_agent를 만들게요.
model_client = OpenAIChatCompletionClient(model="gpt-4o")
search_agent = AssistantAgent(
name="Google_Search_Agent",
model_client=model_client,
tools=[google_search_tool],
description="Search Google for information, returns top 2 results with a snippet and body content",
system_message="You are a helpful AI assistant. Solve tasks using your tools.",
)
stock_analysis_agent = AssistantAgent(
name="Stock_Analysis_Agent",
model_client=model_client,
tools=[stock_analysis_tool],
description="Analyze stock data and generate a plot",
system_message="Perform data analysis.",
)
report_agent = AssistantAgent(
name="Report_Agent",
model_client=model_client,
description="Generate a report based the search and results of stock analysis",
system_message="You are a helpful assistant that can generate a comprehensive report on a given topic based on search and stock analysis. When you done with generating the report, reply with TERMINATE.",
)
팀 만들기
마지막으로 세 에이전트로 팀을 만들고 회사 리서치를 하도록 설정할게요.
team = RoundRobinGroupChat([stock_analysis_agent, search_agent, report_agent], max_turns=3)
max_turns=3을 사용해 턴 수를 팀의 에이전트 수와 정확히 같게 제한했어요. 이러면 에이전트들이 사실상 순차적으로 작업하게 됩니다.
stream = team.run_stream(task="Write a financial report on American airlines")
await Console(stream)
await model_client.close()
더 알아보기 (Learn more)
- 웹서치·외부 API 도구를 직접 만드는 원리는 도구(Tools)를 참고하세요.
- 라운드로빈 그룹챗과 팀 구성은 팀 튜토리얼을 확인하세요.
- 여행 플래닝처럼 공동 작업하는 다른 end-to-end 예제는 Travel Planning을 보세요.