多代理模式：编排器、Worker 与管道

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:
    # 第一步：规划工作
    plan_response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        system="""你是一个规划代理。给定用户请求，将其分解为
2-4 个具体子任务。仅返回任务描述的 JSON 数组。
示例：["研究 X", "分析 Y", "综合发现"]""",
        messages=[{"role": "user", "content": user_request}]
    )

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

    # 第二步：用专业 Worker 执行每个任务
    results = []
    for task in tasks:
        result = run_worker(
            system_prompt="你是一个专注的执行代理。彻底完成分配的任务。",
            task=task
        )
        results.append({"task": task, "result": result})

    # 第三步：综合结果
    synthesis_prompt = f"""原始请求：{user_request}

Worker 结果：
{json.dumps(results, indent=2, ensure_ascii=False)}

将这些结果综合为一个连贯的最终回答。"""

    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": "研究员",
            "system": "从输入中提取并组织所有关键事实。"
                      "格式化为结构化列表，在可用时注明来源。",
        },
        {
            "name": "写作者",
            "system": "将研究笔记转化为清晰、可读的散文。"
                      "保持所有事实内容。面向技术受众。",
        },
        {
            "name": "编辑",
            "system": "提高清晰度和简洁性。删除冗余内容。"
                      "不要改变事实。只返回改进后的文本。",
        },
        {
            "name": "事实核查员",
            "system": "审查内部一致性。标记任何相互矛盾或"
                      "似乎缺乏支撑的声明。"
                      "如果没有问题，返回'已验证：'后跟原始文本。",
        },
    ]

    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']}] 完成（{len(current)} 字符）")

    return current

import asyncio

async def analyze_document(doc: str, index: int) -> dict:
    """异步分析单个文档。"""
    system = """分析此文档并返回包含以下内容的 JSON 对象：
- "sentiment"：positive/negative/neutral
- "key_topics"：3-5 个主要话题的列表
- "summary"：2-3 句话的摘要
- "flags"：关注点列表（如果没有则为空列表）"""

    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:
    # 扇出：并发分析所有文档
    tasks = [analyze_document(doc, i) for i, doc in enumerate(documents)]
    analyses = await asyncio.gather(*tasks)

    # 用专用综合代理聚合
    synthesis_input = json.dumps({
        "document_count": len(documents),
        "analyses": analyses
    }, ensure_ascii=False)

    aggregate_result = await asyncio.to_thread(
        lambda: client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            system="将文档分析综合为：总体情感分布、所有文档中的顶级主题，"
                   "以及标记中的显著模式。以 JSON 格式返回。",
            messages=[{"role": "user", "content": synthesis_input}]
        ).content[0].text
    )

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