10장: 실전 프로젝트 - 리서치 에이전트 시스템 구축

from dataclasses import dataclass
 
@dataclass
class AgentConfig:
    model: str = "claude-sonnet-4-20250514"
    max_tokens: int = 4096
    max_search_results: int = 10
    max_iterations: int = 5
    max_cost_per_run: float = 2.0
    max_reflection_rounds: int = 2
    memory_top_k: int = 5

import anthropic
from config import AgentConfig
 
class BaseAgent:
    """모든 에이전트의 기반 클래스입니다."""
 
    def __init__(self, config: AgentConfig):
        self.client = anthropic.Anthropic()
        self.config = config
 
    def call_llm(self, system: str, messages: list,
                 tools: list = None) -> object:
        """LLM API를 호출합니다."""
        kwargs = {
            "model": self.config.model,
            "max_tokens": self.config.max_tokens,
            "system": system,
            "messages": messages,
        }
        if tools:
            kwargs["tools"] = tools
        return self.client.messages.create(**kwargs)
 
    def extract_text(self, response) -> str:
        """응답에서 텍스트를 추출합니다."""
        for block in response.content:
            if hasattr(block, "text"):
                return block.text
        return ""

import httpx
 
SEARCH_TOOL_SCHEMA = {
    "name": "web_search",
    "description": (
        "웹에서 정보를 검색합니다. "
        "기술 동향, 공식 문서, 블로그 글 등을 검색할 때 사용합니다."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "검색 쿼리. 구체적이고 명확한 검색어를 사용하십시오."
            },
            "num_results": {
                "type": "integer",
                "description": "반환할 결과 수. 기본값 5.",
                "default": 5
            }
        },
        "required": ["query"]
    }
}
 
def web_search(query: str, num_results: int = 5) -> str:
    """웹 검색을 수행합니다."""
    # 실제 구현에서는 Tavily, Serper, Brave Search 등의 API를 사용합니다.
    # 여기서는 인터페이스만 보여줍니다.
    try:
        response = httpx.get(
            "https://api.search-provider.com/search",
            params={"q": query, "num": num_results},
            headers={"Authorization": f"Bearer {API_KEY}"},
            timeout=10,
        )
        results = response.json().get("results", [])
        formatted = []
        for r in results:
            formatted.append(
                f"제목: {r['title']}\n"
                f"URL: {r['url']}\n"
                f"요약: {r['snippet']}\n"
            )
        return "\n---\n".join(formatted)
    except Exception as e:
        return f"검색 오류: {str(e)}"

SCRAPE_TOOL_SCHEMA = {
    "name": "extract_page_content",
    "description": (
        "웹 페이지의 주요 텍스트 내용을 추출합니다. "
        "검색 결과에서 더 자세한 정보가 필요할 때 사용합니다."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "url": {
                "type": "string",
                "description": "텍스트를 추출할 웹 페이지 URL"
            }
        },
        "required": ["url"]
    }
}
 
def extract_page_content(url: str) -> str:
    """웹 페이지에서 주요 텍스트를 추출합니다."""
    try:
        response = httpx.get(url, timeout=15, follow_redirects=True)
        # 실제로는 BeautifulSoup이나 trafilatura를 사용합니다
        from trafilatura import extract
        text = extract(response.text)
        if text:
            return text[:3000]  # 컨텍스트 절약을 위해 길이 제한
        return "페이지에서 텍스트를 추출할 수 없습니다."
    except Exception as e:
        return f"페이지 추출 오류: {str(e)}"

import json
from agents import BaseAgent
 
class PlannerAgent(BaseAgent):
    SYSTEM_PROMPT = """당신은 기술 리서치 계획을 수립하는 전문가입니다.
 
주어진 주제에 대해 체계적인 리서치 계획을 수립합니다.
계획은 구체적이고 실행 가능한 검색 쿼리 목록을 포함해야 합니다."""
 
    def plan(self, topic: str, existing_knowledge: str = "") -> dict:
        """리서치 계획을 생성합니다."""
        prompt = f"""다음 주제에 대한 리서치 계획을 수립하십시오.
 
주제: {topic}
 
{f'이미 알고 있는 정보:\n{existing_knowledge}' if existing_knowledge else ''}
 
다음 JSON 형식으로 계획을 제시하십시오:
{{
  "research_questions": ["답변해야 할 핵심 질문들"],
  "search_queries": ["실행할 구체적 검색 쿼리들"],
  "aspects": ["다루어야 할 측면/관점들"],
  "expected_sources": ["기대되는 정보 소스 유형"]
}}"""
 
        response = self.call_llm(
            system=self.SYSTEM_PROMPT,
            messages=[{"role": "user", "content": prompt}],
        )
        text = self.extract_text(response)
        try:
            return json.loads(text)
        except json.JSONDecodeError:
            return {
                "research_questions": [topic],
                "search_queries": [topic],
                "aspects": [],
                "expected_sources": [],
            }

from agents import BaseAgent
from tools.search import SEARCH_TOOL_SCHEMA, web_search
from tools.scraper import SCRAPE_TOOL_SCHEMA, extract_page_content
 
class ResearcherAgent(BaseAgent):
    SYSTEM_PROMPT = """당신은 기술 리서처입니다.
주어진 검색 쿼리를 실행하고, 관련 자료를 체계적으로 수집합니다.
 
수집 원칙:
1. 공식 문서와 신뢰할 수 있는 소스를 우선합니다.
2. 최신 정보를 우선합니다.
3. 중복된 정보는 제거합니다.
4. 핵심 사실과 데이터를 정리합니다."""
 
    TOOLS = [SEARCH_TOOL_SCHEMA, SCRAPE_TOOL_SCHEMA]
 
    def research(self, queries: list[str]) -> list[dict]:
        """검색 쿼리를 실행하고 자료를 수집합니다."""
        all_findings = []
 
        for query in queries:
            findings = self._search_and_extract(query)
            all_findings.extend(findings)
 
        return all_findings
 
    def _search_and_extract(self, query: str) -> list[dict]:
        """하나의 쿼리에 대해 검색하고 핵심 정보를 추출합니다."""
        messages = [{
            "role": "user",
            "content": (
                f"다음 쿼리로 검색하고 핵심 정보를 수집하십시오: {query}\n"
                f"검색 후 가장 관련성 높은 결과 2-3개의 상세 내용을 추출하십시오."
            ),
        }]
 
        findings = []
 
        for _ in range(self.config.max_iterations):
            response = self.call_llm(
                system=self.SYSTEM_PROMPT,
                messages=messages,
                tools=self.TOOLS,
            )
 
            if response.stop_reason == "end_turn":
                summary = self.extract_text(response)
                findings.append({
                    "query": query,
                    "summary": summary,
                })
                break
 
            # 도구 호출 처리
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    if block.name == "web_search":
                        result = web_search(**block.input)
                    elif block.name == "extract_page_content":
                        result = extract_page_content(**block.input)
                    else:
                        result = "알 수 없는 도구"
 
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result,
                    })
 
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})
 
        return findings

from agents import BaseAgent
 
class AnalystAgent(BaseAgent):
    SYSTEM_PROMPT = """당신은 기술 분석가입니다.
수집된 자료에서 패턴, 관계, 인사이트를 도출합니다.
 
분석 원칙:
1. 사실과 의견을 명확히 구분합니다.
2. 데이터로 뒷받침되는 인사이트를 우선합니다.
3. 상충되는 정보는 양쪽을 모두 제시합니다.
4. 불확실한 부분은 명시합니다."""
 
    def analyze(self, findings: list[dict], topic: str) -> dict:
        """수집된 자료를 분석합니다."""
        findings_text = "\n\n".join(
            f"### 쿼리: {f['query']}\n{f['summary']}"
            for f in findings
        )
 
        prompt = f"""다음 수집된 자료를 분석하십시오.
 
주제: {topic}
 
수집된 자료:
{findings_text}
 
다음 형식으로 분석 결과를 제시하십시오:
 
## 핵심 발견
- 가장 중요한 발견 3-5개
 
## 트렌드와 패턴
- 식별된 트렌드
 
## 비교 분석
- 주요 항목 간 비교 (해당되는 경우)
 
## 결론과 시사점
- 실무에 적용 가능한 시사점
 
## 추가 조사가 필요한 영역
- 현재 자료로 답변할 수 없는 질문"""
 
        response = self.call_llm(
            system=self.SYSTEM_PROMPT,
            messages=[{"role": "user", "content": prompt}],
        )
        analysis = self.extract_text(response)
        return {
            "topic": topic,
            "analysis": analysis,
            "source_count": len(findings),
        }

from agents import BaseAgent
 
class WriterAgent(BaseAgent):
    SYSTEM_PROMPT = """당신은 기술 보고서 작성 전문가입니다.
분석 결과를 바탕으로 구조화되고 읽기 쉬운 보고서를 작성합니다.
 
작성 원칙:
1. 경어체(합니다/습니다)를 사용합니다.
2. 전문적이면서 이해하기 쉬운 문체를 사용합니다.
3. 핵심 내용을 먼저 제시합니다 (역피라미드 구조).
4. 도표나 목록을 활용하여 정보를 구조화합니다.
5. 출처를 명시합니다."""
 
    def write(self, analysis: dict) -> str:
        """분석 결과를 보고서로 작성합니다."""
        prompt = f"""다음 분석 결과를 바탕으로 기술 리서치 보고서를 작성하십시오.
 
주제: {analysis['topic']}
 
분석 결과:
{analysis['analysis']}
 
보고서 구조:
1. 요약 (Executive Summary) - 핵심 결론을 3-4문장으로
2. 배경 - 왜 이 주제가 중요한지
3. 주요 발견 - 핵심 발견 사항을 상세히
4. 비교 분석 - 비교표 포함 (해당되는 경우)
5. 결론 및 권고사항 - 실무 적용 가이드
6. 참고 자료
 
마크다운 형식으로 작성하십시오."""
 
        response = self.call_llm(
            system=self.SYSTEM_PROMPT,
            messages=[{"role": "user", "content": prompt}],
        )
        return self.extract_text(response)

import json
from agents import BaseAgent
 
class ReviewerAgent(BaseAgent):
    SYSTEM_PROMPT = """당신은 엄격한 기술 문서 검토자입니다.
보고서의 정확성, 완전성, 명확성을 검증합니다.
 
반드시 비판적으로 검토하십시오. 문제가 없더라도 개선점을 찾으십시오."""
 
    def review(self, report: str, topic: str) -> dict:
        """보고서를 검토합니다."""
        prompt = f"""다음 기술 보고서를 검토하십시오.
 
주제: {topic}
 
보고서:
{report}
 
평가 기준:
1. 정확성 (1-5): 기술적 내용이 정확한가?
2. 완전성 (1-5): 주제를 충분히 다루었는가?
3. 구조 (1-5): 논리적으로 구성되었는가?
4. 명확성 (1-5): 이해하기 쉽게 작성되었는가?
5. 실용성 (1-5): 실무에 적용 가능한가?
 
JSON으로 응답:
{{
  "scores": {{
    "accuracy": N,
    "completeness": N,
    "structure": N,
    "clarity": N,
    "practicality": N
  }},
  "overall": N,
  "issues": ["발견된 문제"],
  "suggestions": ["개선 제안"],
  "pass": true/false
}}
 
overall 4.0 이상이면 pass: true로 평가하십시오."""
 
        response = self.call_llm(
            system=self.SYSTEM_PROMPT,
            messages=[{"role": "user", "content": prompt}],
        )
        text = self.extract_text(response)
        try:
            return json.loads(text)
        except json.JSONDecodeError:
            return {"overall": 3.0, "pass": False, "suggestions": [text]}

from config import AgentConfig
from agents.planner import PlannerAgent
from agents.researcher import ResearcherAgent
from agents.analyst import AnalystAgent
from agents.writer import WriterAgent
from agents.reviewer import ReviewerAgent
from memory.store import MemoryStore
from guardrails.cost import CostController
 
class ResearchOrchestrator:
    def __init__(self, config: AgentConfig = None):
        self.config = config or AgentConfig()
        self.planner = PlannerAgent(self.config)
        self.researcher = ResearcherAgent(self.config)
        self.analyst = AnalystAgent(self.config)
        self.writer = WriterAgent(self.config)
        self.reviewer = ReviewerAgent(self.config)
        self.memory = MemoryStore()
        self.cost = CostController(
            max_cost=self.config.max_cost_per_run
        )
 
    def run(self, topic: str) -> dict:
        """리서치 파이프라인을 실행합니다."""
        print(f"\n[리서치 시작] 주제: {topic}")
 
        # 1. 기존 메모리 확인
        existing = self.memory.recall(topic)
        print(f"[메모리] 관련 기록 {len(existing)}건 발견")
 
        # 2. 리서치 계획 수립
        print("[계획] 리서치 계획 수립 중...")
        plan = self.planner.plan(
            topic,
            existing_knowledge="\n".join(existing) if existing else "",
        )
        queries = plan.get("search_queries", [topic])
        print(f"[계획] {len(queries)}개 검색 쿼리 생성")
 
        # 3. 자료 수집
        print("[수집] 자료 수집 중...")
        findings = self.researcher.research(queries)
        print(f"[수집] {len(findings)}건의 자료 수집 완료")
 
        # 4. 분석
        print("[분석] 자료 분석 중...")
        analysis = self.analyst.analyze(findings, topic)
        print("[분석] 완료")
 
        # 5. 보고서 작성 + 리플렉션 루프
        report = None
        for attempt in range(self.config.max_reflection_rounds + 1):
            print(f"[작성] 보고서 작성 중... (시도 {attempt + 1})")
 
            if attempt == 0:
                report = self.writer.write(analysis)
            else:
                # 이전 피드백을 반영하여 재작성
                report = self.writer.rewrite(report, review["suggestions"])
 
            # 검증
            print("[검증] 보고서 검증 중...")
            review = self.reviewer.review(report, topic)
            score = review.get("overall", 0)
            print(f"[검증] 점수: {score}/5.0")
 
            if review.get("pass", False):
                print(f"[검증] 통과 (시도 {attempt + 1})")
                break
            elif attempt < self.config.max_reflection_rounds:
                print(f"[검증] 미통과 - 개선 필요: {review.get('suggestions', [])}")
 
        # 6. 메모리 저장
        self.memory.store(topic, analysis["analysis"])
        print("[메모리] 리서치 결과 저장 완료")
 
        # 7. 결과 반환
        result = {
            "topic": topic,
            "report": report,
            "review_score": review.get("overall", 0),
            "sources_used": len(findings),
            "cost": self.cost.get_summary(),
        }
 
        print(f"\n[완료] 보고서 생성 완료")
        print(f"  점수: {result['review_score']}/5.0")
        print(f"  소스: {result['sources_used']}건")
        print(f"  비용: {result['cost']}")
 
        return result

from orchestrator import ResearchOrchestrator
from config import AgentConfig
 
def main():
    config = AgentConfig(
        model="claude-sonnet-4-20250514",
        max_cost_per_run=2.0,
        max_reflection_rounds=2,
    )
    orchestrator = ResearchOrchestrator(config)
 
    result = orchestrator.run(
        "2026년 AI 에이전트 프레임워크 비교 분석: "
        "LangGraph vs CrewAI vs OpenAI Agents SDK"
    )
 
    # 보고서 저장
    with open("report.md", "w") as f:
        f.write(result["report"])
    print(f"\n보고서가 report.md에 저장되었습니다.")
 
if __name__ == "__main__":
    main()