第 7 课：多 Agent 协作

为什么需要多个 Agent？

一个人干所有活，效率低、容易出错。DevOps 也一样：

code

单 Agent 模式：
  一个 AI 做所有事：看代码、跑测试、写文档、部署...
  问题：提示词太长、角色不清晰、容易搞混

多 Agent 模式：
  代码审查员 → 专门找 bug
  测试员 → 专门跑测试
  部署员 → 专门做部署
  各司其职，效率翻倍

这就是 AgentDefinition 的设计思想：让每个 Agent 专注做一件事。

AgentDefinition 基础

定义一个 Agent

code
from claude_agent_sdk import AgentDefinition, ClaudeAgentOptions, query

options = ClaudeAgentOptions(
    agents={
        "code-reviewer": AgentDefinition(
            description="审查代码质量，找出 bug 和安全问题",
            prompt="你是代码审查专家。检查代码中的 bug、性能问题、安全漏洞。给出具体的改进建议。",
            tools=["Read", "Grep"],       # 只读，不能改代码
            model="sonnet",               # 用 Sonnet 模型
        ),
    },
)

AgentDefinition 有四个字段：

字段	作用	示例
`description`	告诉主 Agent 这个子 Agent 干什么	"审查代码质量"
`prompt`	子 Agent 的系统提示词	"你是代码审查专家..."
`tools`	子 Agent 能用哪些工具	["Read", "Grep"]
`model`	用哪个模型（可选）	"sonnet"

使用 Agent

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

async def review_code():
    options = ClaudeAgentOptions(
        agents={
            "code-reviewer": AgentDefinition(
                description="审查代码质量，找出 bug 和安全问题",
                prompt="你是代码审查专家。检查代码的 bug、性能、安全问题，用中文回复。",
                tools=["Read", "Grep"],
                model="sonnet",
            ),
        },
    )

    async for msg in query(
        prompt="用 code-reviewer 审查 src/ 目录下的代码",
        options=options,
    ):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    print(block.text)
        elif isinstance(msg, ResultMessage):
            if msg.total_cost_usd and msg.total_cost_usd > 0:
                print(f"\n💰 费用: ${msg.total_cost_usd:.4f}")

anyio.run(review_code)

执行流程：

graph TD A["你：用 code-reviewer 审查 src/ 目录下的代码"] --> B["主 Agent 收到任务"] B --> C["主 Agent 看到有 code-reviewer 可用"] C --> D["主 Agent 用 Task 工具启动 code-reviewer"] D --> E["code-reviewer 子 Agent 启动"] E --> E1["系统提示词：你是代码审查专家..."] E --> E2["可用工具：Read, Grep"] E --> E3["开始工作：读代码 → 分析 → 输出报告"] E3 --> F["子 Agent 完成，结果返回给主 Agent"] F --> G["主 Agent 整理输出给你"]

多 Agent 协作

定义多个 Agent

bash
options = ClaudeAgentOptions(
    agents={
        "analyzer": AgentDefinition(
            description="分析代码结构和架构",
            prompt="你是代码架构分析师。分析代码的结构、模式、依赖关系。用中文回复。",
            tools=["Read", "Grep", "Glob"],
        ),
        "tester": AgentDefinition(
            description="编写和运行测试",
            prompt="你是测试工程师。编写全面的测试用例，确保代码质量。用中文回复。",
            tools=["Read", "Write", "Bash"],
            model="sonnet",
        ),
        "doc-writer": AgentDefinition(
            description="编写技术文档",
            prompt="你是技术文档专家。根据代码写清晰、完整的文档。用中文回复。",
            tools=["Read", "Write"],
            model="sonnet",
        ),
    },
)

使用时，主 Agent 会根据你的指令自动选择合适的子 Agent：

code
# 主 Agent 自动选择 analyzer
await query(prompt="分析一下项目的整体架构", options=options)

# 主 Agent 自动选择 tester
await query(prompt="给 utils.py 写单元测试", options=options)

# 主 Agent 可能同时用 analyzer + doc-writer
await query(prompt="分析代码结构，然后写一份架构文档", options=options)

DevOps 场景：审查→测试→部署流水线

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

async def devops_pipeline():
    """代码审查 → 测试 → 部署 三步流水线"""

    options = ClaudeAgentOptions(
        agents={
            "reviewer": AgentDefinition(
                description="审查代码变更，找出问题",
                prompt="""你是高级代码审查员。你的职责：
1. 检查最近的 Git 变更
2. 找出潜在的 bug 和安全问题
3. 评估代码质量
4. 给出"通过"或"不通过"的结论
用中文回复。""",
                tools=["Bash", "Read", "Grep"],
                model="sonnet",
            ),
            "tester": AgentDefinition(
                description="运行测试并分析结果",
                prompt="""你是测试工程师。你的职责：
1. 运行项目的测试套件
2. 分析测试结果
3. 如果有失败，分析失败原因
4. 给出"测试通过"或"测试失败"的结论
用中文回复。""",
                tools=["Bash", "Read"],
                model="sonnet",
            ),
            "deployer": AgentDefinition(
                description="执行部署操作",
                prompt="""你是部署工程师。你的职责：
1. 只在审查通过且测试通过后才部署
2. 执行部署脚本
3. 验证部署结果
4. 报告部署状态
用中文回复。""",
                tools=["Bash", "Read"],
                model="sonnet",
            ),
        },
    )

    print("🚀 启动 DevOps 流水线\n")

    async for msg in query(
        prompt="""按顺序执行以下流程：
1. 用 reviewer 审查最近的代码变更
2. 如果审查通过，用 tester 跑测试
3. 如果测试通过，用 deployer 执行部署
任何一步失败就停止并报告原因。""",
        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}")

anyio.run(devops_pipeline)

Agent 的权限隔离

不同 Agent 给不同权限——这是安全的关键：

bash
agents={
    # 审查员：只读，不能改任何东西
    "reviewer": AgentDefinition(
        description="代码审查",
        prompt="...",
        tools=["Read", "Grep", "Glob"],  # 只读工具
    ),

    # 测试员：能执行命令，但不能写文件
    "tester": AgentDefinition(
        description="运行测试",
        prompt="...",
        tools=["Bash", "Read"],  # 能跑命令，能读文件
    ),

    # 修复员：能读写文件
    "fixer": AgentDefinition(
        description="修复代码问题",
        prompt="...",
        tools=["Read", "Write", "Edit"],  # 能改文件，但不能执行命令
    ),

    # 部署员：能执行命令和读文件
    "deployer": AgentDefinition(
        description="部署应用",
        prompt="...",
        tools=["Bash", "Read"],  # 能跑部署脚本
    ),
}

权限矩阵：

bash

          Read  Grep  Glob  Bash  Write  Edit
reviewer   ✅    ✅    ✅    ❌    ❌     ❌
tester     ✅    ❌    ❌    ✅    ❌     ❌
fixer      ✅    ❌    ❌    ❌    ✅     ✅
deployer   ✅    ❌    ❌    ✅    ❌     ❌

每个 Agent 只有完成自己任务所需的最小权限，这就是最小权限原则。

Agent + Hook 组合

Agent 和 Hook 可以组合使用，实现更精细的安全控制：

code
from claude_agent_sdk.types import HookInput, HookContext, HookJSONOutput, HookMatcher

# 审计 Hook：记录哪个 Agent 做了什么
async def agent_audit_hook(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """记录所有 Agent 的工具调用"""
    tool_name = input_data["tool_name"]
    tool_input = input_data["tool_input"]
    print(f"  📝 审计: {tool_name} → {str(tool_input)[:100]}")
    return {}

options = ClaudeAgentOptions(
    agents={
        "reviewer": AgentDefinition(...),
        "tester": AgentDefinition(...),
    },
    hooks={
        "PostToolUse": [
            HookMatcher(matcher=None, hooks=[agent_audit_hook]),
        ],
    },
)

模型选择策略

不同 Agent 可以用不同模型，平衡成本和能力：

bash
agents={
    # 简单任务：用快速便宜的模型
    "file-lister": AgentDefinition(
        description="列出和搜索文件",
        prompt="...",
        tools=["Glob", "Grep"],
        model="haiku",          # 快速、便宜
    ),

    # 复杂分析：用强力模型
    "architect": AgentDefinition(
        description="分析系统架构",
        prompt="...",
        tools=["Read", "Grep", "Glob"],
        model="sonnet",         # 强力、均衡
    ),

    # 关键决策：用最强模型
    "security-auditor": AgentDefinition(
        description="安全审计",
        prompt="...",
        tools=["Read", "Grep", "Bash"],
        model="opus",           # 最强、最贵
    ),
}

本课小结

AgentDefinition 定义子 Agent 的角色、提示词、工具、模型
多个 Agent 各司其职，主 Agent 自动调度
权限隔离：每个 Agent 只给需要的工具（最小权限原则）
Agent + Hook 组合实现审计和安全控制
不同 Agent 可以用不同模型，平衡成本和能力

课后练习

定义一个三 Agent 系统：分析器、修复器、验证器
给每个 Agent 设置不同的权限，确保修复器不能执行 Bash 命令
加一个 PostToolUse Hook 记录每个 Agent 的操作日志