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

최종 프로젝트: 자율 리서치 에이전트

주제를 입력하면 스스로 리서치하고 보고서를 작성하는 에이전트를 만듭니다.

flowchart TB
    USER["주제 입력\n'AI 반도체 시장 동향'"] --> PLAN

    subgraph AGENT["자율 리서치 에이전트"]
        PLAN["① 계획 수립\n리서치 범위·방향 결정"]
        SEARCH["② 정보 수집\n웹 검색 + 문서 검색"]
        ANALYZE["③ 분석·정리\n핵심 인사이트 추출"]
        WRITE["④ 보고서 작성\n구조화된 문서 생성"]
        REVIEW["⑤ 자기 검토\n오류·누락 확인"]
        
        PLAN --> SEARCH --> ANALYZE --> WRITE --> REVIEW
        REVIEW -->|"개선 필요"| WRITE
        REVIEW -->|"완료"| DONE
    end
    
    DONE["📄 최종 보고서 저장"]

---

프로젝트 구조

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

---

공유 상태 (state.py)

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/)

# 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/)

# 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"
    }

---

그래프 조립 (graph.py)

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()

---

실행 (main.py)

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] + "...")

---

시리즈 전체 요약

flowchart LR
    E1["편 1\n에이전트 개념"] --> E2["편 2\nLangChain 기초"]
    E2 --> E3["편 3\n도구+LangGraph"]
    E3 --> E4["편 4\n메모리·상태"]
    E4 --> E5["편 5\n멀티에이전트"]
    E5 --> E6["편 6\n실전 프로젝트"]

편	핵심 개념
1	에이전트 = LLM + 도구 + 메모리 + 계획
2	LCEL로 체인 조합, 메모리 관리
3	@tool, LangGraph 상태 그래프
4	단기/장기 메모리, 대화 요약
5	오케스트레이터-서브에이전트 패턴
6	전체 통합: 자율 리서치 에이전트

이것으로 AI 에이전트 기초 시리즈를 마칩니다.