第 7 课：多 Agent 协作

from claude_agent_sdk import AgentDefinition

# 1. 分类员：快速判断工单类型和紧急度
classifier = AgentDefinition(
    description="快速分类工单，判断类型、紧急度和应分配的团队",
    prompt="""你是工单分类专家。收到工单后：
1. 判断类别: bug / feature_request / billing / account / consulting / complaint
2. 判断紧急度: critical / high / medium / low
3. 判断情绪: angry / frustrated / neutral / positive
4. 推荐分配团队: engineering / support / billing / sales

输出格式（JSON）：
{"category": "...", "urgency": "...", "sentiment": "...", "team": "..."}
只输出 JSON。""",
    tools=["Read"],
    model="haiku",  # 分类用便宜快速的模型就行
)

# 2. 客服专员：回答客户问题
support_agent = AgentDefinition(
    description="回答客户问题，使用知识库和工单系统为客户提供帮助",
    prompt="""你是专业客服"小智"。工作规则：
1. 友好、专业、简洁
2. 先在知识库搜索答案
3. 需要时查询工单或订单
4. 信息不够就追问客户
5. 解决不了的转人工
6. 用中文回复，每次不超过 3 句话""",
    tools=["Read", "Bash"],  # 会结合 MCP 工具
    model="sonnet",  # 对话质量要好
)

# 3. 质检员：检查回复质量
quality_checker = AgentDefinition(
    description="检查客服回复的质量，确保专业性和准确性",
    prompt="""你是客服质检员。检查标准：
1. 礼貌性：是否友好、有同理心
2. 准确性：信息是否正确
3. 完整性：是否回答了客户的问题
4. 合规性：是否泄露敏感信息
5. 专业性：用词是否专业

输出格式：
- 评分: 1-10 分
- 问题: 发现的问题（如果有）
- 建议: 改进建议（如果有）""",
    tools=["Read"],
    model="haiku",  # 质检不需要最强模型
)

# 4. 升级判断员：决定是否转人工
escalation_judge = AgentDefinition(
    description="判断工单是否需要升级到人工处理",
    prompt="""你是升级判断专家。以下情况需要转人工：
1. 客户明确要求人工服务
2. 客户非常愤怒（连续表达不满）
3. 涉及法律、投诉、安全问题
4. AI 无法在 3 轮对话内解决
5. 涉及超出权限的操作（大额退款等）

输出格式（JSON）：
{"should_escalate": true/false, "reason": "原因", "suggested_team": "建议转给的团队"}
只输出 JSON。""",
    tools=["Read"],
    model="haiku",
)

import anyio
from claude_agent_sdk import (
    query, ClaudeAgentOptions,
    AssistantMessage, ResultMessage, TextBlock, ToolUseBlock,
)

async def smart_support(ticket_text: str):
    """智能客服流程：分类 → 回复 → 质检"""

    options = ClaudeAgentOptions(
        system_prompt="""你是客服系统的调度中心。你有以下 Agent 可以调度：

1. classifier - 分类员：快速分类工单
2. support-agent - 客服专员：回答客户问题
3. quality-checker - 质检员：检查回复质量
4. escalation-judge - 升级判断员：判断是否转人工

工作流程：
1. 先用 classifier 分类工单
2. 根据分类结果，用 support-agent 生成回复
3. 用 quality-checker 检查回复质量
4. 如果质量不达标，让 support-agent 重写
5. 用 escalation-judge 判断是否需要转人工

请按以上流程处理每个工单。用中文输出最终结果。""",
        agents={
            "classifier": classifier,
            "support-agent": support_agent,
            "quality-checker": quality_checker,
            "escalation-judge": escalation_judge,
        },
    )

    print(f"\n📋 收到工单: {ticket_text}")
    print("-" * 40)

    async for msg in query(prompt=f"处理这个客户工单: {ticket_text}", options=options):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    print(block.text)
                elif isinstance(block, ToolUseBlock):
                    print(f"  🔧 调度 {block.name}")
        elif isinstance(msg, ResultMessage):
            if msg.total_cost_usd and msg.total_cost_usd > 0:
                print(f"\n💰 总费用: ${msg.total_cost_usd:.4f}")


async def main():
    tickets = [
        "你们的 API 一直报 500 错误，线上服务挂了！赶紧处理！",
        "请问企业版和基础版有什么区别？",
        "我要投诉！已经三次了，每次都解决不了问题！我要找你们经理！",
    ]

    for ticket in tickets:
        await smart_support(ticket)
        print("\n" + "=" * 50)

anyio.run(main)

工单: "API 报 500 错误，线上挂了"
  │
  ├── 🔧 调度 classifier
  │   └── {"category": "bug", "urgency": "critical", "sentiment": "frustrated"}
  │
  ├── 🔧 调度 support-agent
  │   └── "非常抱歉给您带来困扰！我们已收到您的反馈..."
  │
  ├── 🔧 调度 quality-checker
  │   └── 评分: 8/10，建议加上预计解决时间
  │
  └── 🔧 调度 escalation-judge
      └── {"should_escalate": true, "reason": "critical级bug", "suggested_team": "engineering"}

agents={
    # 分类员：只读，不能改任何东西
    "classifier": AgentDefinition(
        description="分类工单",
        prompt="...",
        tools=["Read"],       # 只读
        model="haiku",        # 便宜快速
    ),

    # 客服：能查数据，但不能直接改
    "support-agent": AgentDefinition(
        description="回答问题",
        prompt="...",
        tools=["Read", "Bash"],  # 能查能搜
        model="sonnet",          # 质量好
    ),

    # 质检员：只读
    "quality-checker": AgentDefinition(
        description="检查质量",
        prompt="...",
        tools=["Read"],
        model="haiku",
    ),
}

              Read  Bash  Write  Edit
classifier     ✅    ❌    ❌     ❌
support-agent  ✅    ✅    ❌     ❌
quality-check  ✅    ❌    ❌     ❌
escalation     ✅    ❌    ❌     ❌

分类员 → Haiku（快、便宜，分类够用）
客服 → Sonnet（质量好，对话自然）
质检员 → Haiku（检查清单式工作）
升级判断 → Haiku（规则判断）

from claude_agent_sdk.types import HookMatcher

async def agent_audit(input_data, tool_use_id, context):
    """记录哪个 Agent 干了什么"""
    tool = input_data.get("tool_name", "unknown")
    print(f"  📝 审计: Agent 调用了 {tool}")
    return {}

options = ClaudeAgentOptions(
    agents={...},
    hooks={
        "PostToolUse": [
            HookMatcher(matcher=None, hooks=[agent_audit]),
        ],
    },
)