멀티에이전트 패턴: 오케스트레이터, 워커, 파이프라인

import anthropic
import json

client = anthropic.Anthropic()

def run_worker(system_prompt: str, task: str) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        system=system_prompt,
        messages=[{"role": "user", "content": task}]
    )
    return response.content[0].text

def orchestrator(user_request: str) -> str:
    # Step 1: plan the work
    plan_response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        system="""You are a planning agent. Given a user request, break it into
2-4 specific subtasks. Return a JSON array of task descriptions only.
Example: ["Research X", "Analyze Y", "Synthesize findings"]""",
        messages=[{"role": "user", "content": user_request}]
    )

    tasks = json.loads(plan_response.content[0].text)

    # Step 2: run each task with a specialized worker
    results = []
    for task in tasks:
        result = run_worker(
            system_prompt="You are a focused execution agent. Complete the assigned task thoroughly.",
            task=task
        )
        results.append({"task": task, "result": result})

    # Step 3: synthesize
    synthesis_prompt = f"""Original request: {user_request}

Worker results:
{json.dumps(results, indent=2)}

Synthesize these results into a cohesive final response."""

    final = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{"role": "user", "content": synthesis_prompt}]
    )
    return final.content[0].text

def run_pipeline(input_text: str) -> str:
    stages = [
        {
            "name": "Researcher",
            "system": "Extract and organize all key facts from the input. "
                      "Format as a structured list with sources noted where available.",
        },
        {
            "name": "Writer",
            "system": "Transform the research notes into clear, readable prose. "
                      "Maintain all factual content. Target a technical audience.",
        },
        {
            "name": "Editor",
            "system": "Improve clarity and concision. Remove redundancy. "
                      "Do not change facts. Return only the improved text.",
        },
        {
            "name": "Fact Checker",
            "system": "Review for internal consistency. Flag any claims that "
                      "contradict each other or seem unsupported. "
                      "If no issues, return 'VERIFIED: ' followed by the original text.",
        },
    ]

    current = input_text
    for stage in stages:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system=stage["system"],
            messages=[{"role": "user", "content": current}]
        )
        current = response.content[0].text
        print(f"[{stage['name']}] complete ({len(current)} chars)")

    return current

import asyncio

async def analyze_document(doc: str, index: int) -> dict:
    """Analyze a single document asynchronously."""
    system = """Analyze this document and return a JSON object with:
- "sentiment": positive/negative/neutral
- "key_topics": list of 3-5 main topics
- "summary": 2-3 sentence summary
- "flags": list of any concerns (empty list if none)"""

    # asyncio.to_thread lets you call synchronous code in a thread pool
    result = await asyncio.to_thread(
        lambda: client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=512,
            system=system,
            messages=[{"role": "user", "content": doc}]
        ).content[0].text
    )
    return {"index": index, **json.loads(result)}

async def parallel_analysis(documents: list[str]) -> dict:
    # Fan out: analyze all documents concurrently
    tasks = [analyze_document(doc, i) for i, doc in enumerate(documents)]
    analyses = await asyncio.gather(*tasks)

    # Aggregate with a dedicated synthesis agent
    synthesis_input = json.dumps({
        "document_count": len(documents),
        "analyses": analyses
    })

    aggregate_result = await asyncio.to_thread(
        lambda: client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            system="Synthesize document analyses into: overall sentiment distribution, "
                   "top themes across all documents, and notable patterns in flags. "
                   "Return as JSON.",
            messages=[{"role": "user", "content": synthesis_input}]
        ).content[0].text
    )

    return {
        "individual": analyses,
        "aggregate": json.loads(aggregate_result)
    }