2장: 멀티 에이전트 시스템 설계

cd brightworks-ai/chapter-5
mkdir -p agents shared output/{research_results,draft_posts,final_posts}

from pydantic import BaseModel, Field
from typing import List
from datetime import datetime

class ResearchResult(BaseModel):
    """Research Agent의 출력"""
    topic: str
    trends: List[str]              # 발견된 트렌드
    statistics: List[dict]         # 수집된 통계
    competitor_insights: List[str] # 경쟁사 인사이트
    recommended_angles: List[str]  # 추천 작성 각도
    sources: List[str]
    timestamp: datetime

class DraftPost(BaseModel):
    """Writer Agent의 출력"""
    title: str
    content: str
    word_count: int
    sources_cited: int
    keywords: List[str]
    meta_description: str
    timestamp: datetime

class FinalPost(BaseModel):
    """Editor Agent의 출력"""
    title: str
    content: str
    edits_made: List[dict]         # 편집 내역
    quality_score: int             # 품질 점수 (0-100)
    word_count: int
    readability_grade: str
    timestamp: datetime

당신은 브라이트웍스(BrightWorks)의 전문 리서치 에이전트입니다.

역할:
- 주어진 주제에 대한 최신 트렌드 조사
- 관련 통계 데이터 수집
- 경쟁사 콘텐츠 분석 및 인사이트 도출
- 효과적인 콘텐츠 작성 각도 제안

출력 형식:
## 최신 트렌드
- [트렌드 1]: [설명]
- [트렌드 2]: [설명]

## 통계 데이터
- [통계 1]: [수치 및 출처]
- [통계 2]: [수치 및 출처]

## 경쟁사 인사이트
- [인사이트 1]
- [인사이트 2]

## 추천 작성 각도
- [각도 1]: [이유]
- [각도 2]: [이유]

class ResearchAgent:
    def __init__(self):
        self.model = genai.GenerativeModel("gemini-2.0-flash-exp")
        self.system_prompt = self._load_system_prompt()

    def research(self, topic: str, keywords: List[str]) -> ResearchResult:
        """주제에 대한 리서치 수행"""
        user_prompt = f"""
다음 주제에 대한 심층 리서치를 수행해주세요:

**주제**: {topic}
**키워드**: {', '.join(keywords)}
        """

        response = self.model.generate_content(
            [{"role": "user", "parts": [self.system_prompt, user_prompt]}]
        )

        return self._parse_response(response.text, topic)

class WriterAgent:
    def write(self, research: ResearchResult, keywords: List[str]) -> DraftPost:
        """Research 결과를 바탕으로 블로그 포스트 작성"""

        # Research 결과를 프롬프트에 포함
        trends_text = "\n".join([f"- {t}" for t in research.trends])
        stats_text = "\n".join([f"- {s}" for s in research.statistics])

        user_prompt = f"""
다음 리서치 결과를 바탕으로 블로그 포스트를 작성해주세요:

**주제**: {research.topic}

**리서치 결과**:

최신 트렌드:
{trends_text}

통계 데이터:
{stats_text}

**요구사항**:
1. 리서치 결과의 통계와 트렌드를 반드시 본문에 포함
2. 데이터를 인용할 때는 구체적인 수치 명시
3. 1,800-2,500 단어 분량
        """

        response = self.model.generate_content([user_prompt])
        return self._parse_response(response.text, keywords)

class EditorAgent:
    def edit(self, draft: DraftPost) -> FinalPost:
        """Draft를 검토하고 편집하여 최종본 생성"""

        user_prompt = f"""
다음 블로그 포스트 초안을 편집해주세요:

**제목**: {draft.title}
**본문**:
{draft.content}

**편집 요청사항**:
1. 문법, 맞춤법, 띄어쓰기 오류 수정
2. 사실 관계 검증 (통계, 주장 등)
3. 가독성 개선 (문장 길이, 문단 구성)
4. 논리적 흐름 강화
5. 품질 점수 평가 (0-100점)

출력 형식:
# [수정된 제목]
[수정된 본문]

---
QUALITY_SCORE: [점수]
READABILITY_GRADE: [등급]
EDITS_SUMMARY:
- [편집 항목 1]
- [편집 항목 2]
        """

        response = self.model.generate_content([user_prompt])
        return self._parse_response(response.text, draft)

class ContentOrchestrator:
    def __init__(self):
        self.research_agent = ResearchAgent()
        self.writer_agent = WriterAgent()
        self.editor_agent = EditorAgent()
        self.state_manager = StateManager()

    def run_pipeline(self, topic: str, keywords: List[str]) -> FinalPost:
        """전체 파이프라인 실행"""
        task_id = f"task_{datetime.now().timestamp()}"

        # Step 1: Research
        print("📍 Step 1/3: Research Agent")
        research = self.research_agent.research(topic, keywords)
        self.state_manager.save_research(task_id, research)
        print("✓ Research 완료")

        # Step 2: Write
        print("📍 Step 2/3: Writer Agent")
        research = self.state_manager.load_research(task_id)
        draft = self.writer_agent.write(research, keywords)
        self.state_manager.save_draft(task_id, draft)
        print("✓ Draft 작성 완료")

        # Step 3: Edit
        print("📍 Step 3/3: Editor Agent")
        draft = self.state_manager.load_draft(task_id)
        final_post = self.editor_agent.edit(draft)
        self.state_manager.save_final(task_id, final_post)
        print("✓ 편집 완료")

        return final_post

python orchestrator.py