멀티에이전트 시스템 설계: 여러 AI가 협업하는 아키텍처 패턴

flowchart LR
    Problem[복잡한 태스크] --> Single[단일 LLM]
    Single --> Limit1[컨텍스트 한계<br/>128K~200K 토큰]
    Single --> Limit2[전문성 부족<br/>모든 것을 잘 할 수 없음]
    Single --> Limit3[순차 처리<br/>병렬화 불가]

    Problem --> Multi[멀티에이전트]
    Multi --> Sol1[태스크 분할<br/>각자 작은 컨텍스트]
    Multi --> Sol2[역할 특화<br/>전문 에이전트]
    Multi --> Sol3[병렬 실행<br/>속도 향상]

flowchart TD
    User[사용자] --> Orch[오케스트레이터<br/>Claude Opus]
    Orch --> W1[리서치 에이전트<br/>웹 검색 전담]
    Orch --> W2[분석 에이전트<br/>데이터 처리 전담]
    Orch --> W3[작성 에이전트<br/>문서 생성 전담]
    W1 --> Orch
    W2 --> Orch
    W3 --> Orch
    Orch --> Result[최종 결과]

import anthropic

client = anthropic.Anthropic()

def orchestrator(task: str) -> str:
    # 중앙 오케스트레이터: 태스크를 분석하고 워커에게 위임
    response = client.messages.create(
        model="claude-opus-4-5-20251001",  # 강력한 모델로 조율
        max_tokens=2048,
        system='''당신은 태스크 조율자입니다.
복잡한 태스크를 분석하고 다음 형식으로 서브태스크를 정의하세요:
{"subtasks": [{"agent": "research|analysis|writing", "task": "..."}]}''',
        messages=[{"role": "user", "content": task}]
    )
    import json
    plan = json.loads(response.content[0].text)

    results = {}
    for subtask in plan["subtasks"]:
        agent_type = subtask["agent"]
        results[agent_type] = worker(agent_type, subtask["task"])

    # 결과 통합
    return synthesize(task, results)

def worker(agent_type: str, task: str) -> str:
    # 특화된 워커 에이전트
    system_prompts = {
        "research": "당신은 정보 검색 전문가입니다. 정확한 사실만 제공하세요.",
        "analysis": "당신은 데이터 분석 전문가입니다. 수치와 패턴을 분석하세요.",
        "writing": "당신은 콘텐츠 작성 전문가입니다. 명확하고 간결하게 작성하세요.",
    }
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",  # 워커는 빠른 모델
        max_tokens=1024,
        system=system_prompts[agent_type],
        messages=[{"role": "user", "content": task}]
    )
    return response.content[0].text

flowchart LR
    Raw[원본 데이터] --> A1[수집 에이전트<br/>웹 스크래핑]
    A1 --> A2[정제 에이전트<br/>노이즈 제거]
    A2 --> A3[분석 에이전트<br/>인사이트 추출]
    A3 --> A4[리포트 에이전트<br/>문서 생성]
    A4 --> Output[최종 리포트]

from dataclasses import dataclass
from typing import Callable

@dataclass
class PipelineStage:
    name: str
    agent: Callable
    model: str = "claude-haiku-4-5-20251001"

def run_pipeline(input_data: str, stages: list[PipelineStage]) -> str:
    current = input_data
    for stage in stages:
        print(f"[{stage.name}] 처리 중...")
        current = stage.agent(current, stage.model)
    return current

# 뉴스 처리 파이프라인 예시
pipeline = [
    PipelineStage("수집", collect_articles),
    PipelineStage("필터링", filter_relevant, model="claude-haiku-4-5-20251001"),
    PipelineStage("요약", summarize_articles, model="claude-sonnet-4-6-20251001"),
    PipelineStage("번역", translate_to_korean, model="claude-haiku-4-5-20251001"),
]

result = run_pipeline("AI 최신 뉴스", pipeline)

def draft_and_review(task: str, max_iterations: int = 3) -> str:
    draft = None
    feedback = None

    for i in range(max_iterations):
        # 1. 초안 작성
        draft_response = client.messages.create(
            model="claude-sonnet-4-6-20251001",
            max_tokens=2048,
            system="당신은 전문 작가입니다. 요청받은 내용을 작성하세요.",
            messages=[
                {"role": "user", "content": task},
                *([{"role": "assistant", "content": draft},
                   {"role": "user", "content": f"피드백 반영해서 수정:
{feedback}"}]
                  if draft and feedback else [])
            ]
        )
        draft = draft_response.content[0].text

        # 2. 품질 검토
        review_response = client.messages.create(
            model="claude-opus-4-5-20251001",  # 검토는 더 강한 모델
            max_tokens=1024,
            system="당신은 엄격한 편집자입니다. 개선점을 구체적으로 지적하세요. 충분히 좋으면 'APPROVED'라고만 답하세요.",
            messages=[{"role": "user", "content": f"원본 요청: {task}

초안:
{draft}"}]
        )
        feedback = review_response.content[0].text

        if "APPROVED" in feedback:
            print(f"✓ {i+1}번 만에 승인됨")
            break

    return draft

from crewai import Agent, Task, Crew

# 에이전트 정의
researcher = Agent(
    role="AI 리서처",
    goal="최신 AI 동향을 정확하게 조사한다",
    backstory="5년 경력의 AI 연구원. 논문과 기술 블로그를 빠르게 파악하는 능력이 있다.",
    tools=[web_search_tool, arxiv_tool],
    llm="claude-haiku-4-5-20251001",
    verbose=True
)

analyst = Agent(
    role="데이터 분석가",
    goal="수집된 정보에서 핵심 인사이트를 도출한다",
    backstory="통계와 데이터 분석 전문가. 숫자에서 의미를 찾는다.",
    llm="claude-sonnet-4-6-20251001",
)

writer = Agent(
    role="콘텐츠 라이터",
    goal="분석 결과를 읽기 쉬운 한국어 보고서로 작성한다",
    backstory="IT 전문 기자 출신. 복잡한 기술 내용을 쉽게 설명한다.",
    llm="claude-sonnet-4-6-20251001",
)

# 태스크 정의
research_task = Task(
    description="이번 주 발표된 AI 모델 중 주목할 만한 것 5개를 조사하세요",
    expected_output="모델명, 주요 특징, 성능 지표가 포함된 조사 결과",
    agent=researcher
)

analysis_task = Task(
    description="조사 결과를 바탕으로 한국 개발자에게 가장 중요한 트렌드 3가지를 분석하세요",
    expected_output="트렌드별 중요도와 활용 가능성 분석",
    agent=analyst,
    context=[research_task]  # 이전 태스크 결과를 컨텍스트로 사용
)

write_task = Task(
    description="분석 결과를 2,000자 내외의 한국어 뉴스레터로 작성하세요",
    expected_output="마크다운 형식의 뉴스레터",
    agent=writer,
    context=[research_task, analysis_task]
)

# 크루 실행
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, write_task],
    verbose=True
)

result = crew.kickoff()

flowchart LR
    Single[단일 GPT-4o 호출<br/>1초, $0.02] --> Multi[멀티에이전트 5개 호출<br/>5-15초, $0.10]