실전 프로젝트 — 자율 리서치 에이전트 만들기

research_agent/
├── main.py           # 진입점
├── agents/
│   ├── planner.py    # 계획 에이전트
│   ├── researcher.py # 리서치 에이전트
│   ├── analyst.py    # 분석 에이전트
│   └── writer.py     # 작성 에이전트
├── tools/
│   ├── search.py     # 검색 도구
│   └── file_ops.py   # 파일 도구
├── state.py          # 공유 상태 정의
└── graph.py          # LangGraph 구성

from typing import Annotated, Optional
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages

class ResearchState(TypedDict):
    # 입력
    topic: str
    
    # 에이전트 간 공유 데이터
    messages: Annotated[list, add_messages]
    research_plan: Optional[str]
    raw_research: list[str]
    analysis: Optional[str]
    draft_report: Optional[str]
    final_report: Optional[str]
    
    # 흐름 제어
    next_step: str
    review_count: int
    quality_score: float

# tools/search.py
from langchain_core.tools import tool
from langchain_community.tools import TavilySearchResults

tavily_search = TavilySearchResults(max_results=5)

@tool
def web_search(query: str) -> str:
    """웹에서 최신 정보를 검색합니다."""
    results = tavily_search.invoke(query)
    return "\n\n".join([
        f"제목: {r['title']}\n내용: {r['content'][:300]}"
        for r in results
    ])

@tool
def search_statistics(topic: str) -> str:
    """주제 관련 통계 데이터를 검색합니다."""
    return web_search(f"{topic} statistics data 2024")

# tools/file_ops.py
@tool
def save_report(filename: str, content: str) -> str:
    """보고서를 마크다운 파일로 저장합니다."""
    filepath = f"reports/{filename}.md"
    import os
    os.makedirs("reports", exist_ok=True)
    with open(filepath, "w", encoding="utf-8") as f:
        f.write(content)
    return f"✅ 보고서 저장 완료: {filepath}"

# agents/planner.py
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def planner_agent(state: ResearchState):
    """리서치 계획을 수립합니다."""
    response = llm.invoke([
        SystemMessage("""당신은 리서치 계획 전문가입니다.
주어진 주제에 대해 다음을 작성하세요:
1. 리서치 핵심 질문 5가지
2. 필요한 정보 유형 (통계, 사례, 전문가 의견 등)
3. 예상 보고서 구조"""),
        HumanMessage(f"주제: {state['topic']}")
    ])
    
    return {
        "research_plan": response.content,
        "next_step": "researcher"
    }

# agents/researcher.py
from langgraph.prebuilt import ToolNode

research_tools = [web_search, search_statistics]
research_llm = llm.bind_tools(research_tools)

def researcher_agent(state: ResearchState):
    """계획에 따라 정보를 수집합니다."""
    messages = [
        SystemMessage(f"""당신은 리서치 전문가입니다.
아래 계획에 따라 정보를 수집하세요.

리서치 계획:
{state['research_plan']}"""),
        HumanMessage(f"'{state['topic']}'에 대한 정보를 수집해주세요.")
    ]
    
    response = research_llm.invoke(messages)
    
    raw_data = state.get("raw_research", [])
    raw_data.append(response.content)
    
    return {
        "messages": [response],
        "raw_research": raw_data,
        "next_step": "analyst"
    }

# agents/analyst.py
def analyst_agent(state: ResearchState):
    """수집된 정보를 분석합니다."""
    research_text = "\n\n---\n\n".join(state["raw_research"])
    
    response = llm.invoke([
        SystemMessage("""당신은 데이터 분석 전문가입니다.
수집된 정보에서:
1. 핵심 트렌드 3가지
2. 주요 데이터 포인트
3. 이해관계자별 시사점
을 분석해주세요."""),
        HumanMessage(f"분석할 리서치 데이터:\n{research_text}")
    ])
    
    return {
        "analysis": response.content,
        "next_step": "writer"
    }

# agents/writer.py
writer_tools = [save_report]
writer_llm = llm.bind_tools(writer_tools)

def writer_agent(state: ResearchState):
    """분석 결과로 보고서를 작성합니다."""
    response = writer_llm.invoke([
        SystemMessage("""당신은 전문 리서치 작가입니다.
다음 구조로 마크다운 보고서를 작성하세요:
# 제목
## 개요 (3문장)
## 주요 트렌드
## 핵심 데이터
## 시사점
## 결론"""),
        HumanMessage(
            f"주제: {state['topic']}\n\n"
            f"분석 결과:\n{state['analysis']}"
        )
    ])
    
    return {
        "messages": [response],
        "draft_report": response.content,
        "next_step": "reviewer"
    }

def reviewer_agent(state: ResearchState):
    """보고서 품질을 검토합니다."""
    response = llm.invoke([
        SystemMessage("""보고서를 검토하고 다음을 평가하세요:
1. 완성도 (0~10점)
2. 누락된 중요 내용
3. 개선이 필요한 부분

형식: {"score": 숫자, "missing": "내용", "improvements": "내용"}
JSON으로만 응답."""),
        HumanMessage(state["draft_report"])
    ])
    
    import json
    try:
        review = json.loads(response.content)
        score = float(review.get("score", 7))
    except:
        score = 7.0
    
    return {
        "quality_score": score,
        "review_count": state.get("review_count", 0) + 1,
        "next_step": "writer" if score < 8.0 and state.get("review_count", 0) < 2 else "finalize"
    }

def finalize_agent(state: ResearchState):
    """보고서를 최종 저장합니다."""
    import re
    safe_topic = re.sub(r'[^가-힣a-zA-Z0-9]', '_', state['topic'])
    
    # 파일 저장
    save_report.invoke({
        "filename": safe_topic,
        "content": state["draft_report"]
    })
    
    return {
        "final_report": state["draft_report"],
        "next_step": "end"
    }

from langgraph.graph import StateGraph, END

def route(state: ResearchState) -> str:
    return state["next_step"]

graph = StateGraph(ResearchState)

graph.add_node("planner", planner_agent)
graph.add_node("researcher", researcher_agent)
graph.add_node("analyst", analyst_agent)
graph.add_node("writer", writer_agent)
graph.add_node("reviewer", reviewer_agent)
graph.add_node("finalize", finalize_agent)

graph.set_entry_point("planner")

for node in ["planner", "researcher", "analyst", "writer", "reviewer", "finalize"]:
    graph.add_conditional_edges(node, route, {
        "planner": "planner",
        "researcher": "researcher",
        "analyst": "analyst",
        "writer": "writer",
        "reviewer": "reviewer",
        "finalize": "finalize",
        "end": END
    })

research_agent = graph.compile()

def run_research(topic: str):
    print(f"\n🔍 리서치 시작: {topic}\n")
    
    initial_state = {
        "topic": topic,
        "messages": [],
        "research_plan": None,
        "raw_research": [],
        "analysis": None,
        "draft_report": None,
        "final_report": None,
        "next_step": "planner",
        "review_count": 0,
        "quality_score": 0.0
    }
    
    for event in research_agent.stream(initial_state, stream_mode="updates"):
        node_name = list(event.keys())[0]
        print(f"✅ {node_name} 완료")
    
    final = research_agent.invoke(initial_state)
    print(f"\n📄 보고서 완성 (품질 점수: {final['quality_score']}/10)")
    return final["final_report"]

if __name__ == "__main__":
    report = run_research("2024 생성형 AI 시장 동향")
    print(report[:500] + "...")