第13章：多 Agent 协作 —— 一个人干不完就叫帮手
一句话：用 Subagent 实现多 Agent 协作，让一群 AI 一起干活。
本章目标
- 理解为什么需要多个 Agent 协作
- 掌握三种创建 Subagent 的方式
- 理解 Subagent 的四大属性和工作机制
- 学会串行和并行两种协作模式
- 动手搭建代码审查团队、研究小组、翻译工作室三个实战项目
前置知识
- 需要先看完第4章（query 函数）
- 需要先看完第5章（内置工具，尤其是 Task 工具）
- 需要先看完第6章（权限控制）
13.1 为什么需要多个 Agent？
一个 Agent 的烦恼
你让一个 Agent 同时干这些事：
- 读取项目所有源代码文件
- 分析每个文件的代码质量
- 找出潜在的安全漏洞
- 写测试用例
- 生成一份综合报告
结果会怎样？Agent 的上下文窗口（Context Window）是有限的。塞进去太多东西，它就开始"丢三落四" —— 前面分析过的文件内容被挤掉了，后面写报告的时候就忘了前面的分析结果。
这就好比你让一个人同时当厨师、服务员、收银员和保洁员。不是不行，但效率低、容易出错。
公司模式：老板 + 员工
想想一个正常的公司是怎么运作的：
老板（主 Agent / Main Agent）
├── 代码审查员（Subagent 1）—— 专门看代码写得好不好
├── 安全专家（Subagent 2）—— 专门找安全漏洞
├── 测试工程师（Subagent 3）—— 专门写测试
└── 技术文档员（Subagent 4）—— 专门写报告
老板负责分配任务、收集结果、做最终决策。每个员工专注自己擅长的领域。这就是多 Agent 协作的核心思想。
四大好处
| 好处 | 单 Agent | 多 Agent |
|---|---|---|
| 上下文管理 | 所有信息挤在一个窗口里，容易溢出 | 每个 Agent 有独立上下文，互不干扰 |
| 专业分工 | 一个人什么都做，样样稀松 | 每个 Agent 有专门的 prompt 和工具，术业有专攻 |
| 并行效率 | 一个接一个串行处理 | 多个 Agent 同时干活，速度翻倍 |
| 安全隔离 | 所有工具权限混在一起 | 每个 Agent 只有必要的最小权限 |
真实案例
在 AI Agent 生态里，多 Agent 协作已经是主流模式：
- Claude Code 自身就使用 Task 工具来派子任务给 Subagent
- NanoClaw 是第一个基于 Claude Agent SDK 实现 Agent Swarm（Agent 蜂群）模式的项目
- OpenClaw 使用 Agent Swarm 来处理复杂的多步骤任务
说白了，单打独斗能解决的问题有限。真正复杂的任务，就得团队协作。
13.2 三种创建 Subagent 的方式
Claude Agent SDK 提供了三种创建 Subagent 的方式，适合不同场景。
方式一：代码里定义（推荐）
这是最常用也是最灵活的方式。直接在 query() 的 options 里用 agents 参数定义。
Python 版本：
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition
async def main():
async for message in query(
prompt="审查一下 src/ 目录下的代码，找出安全问题",
options=ClaudeAgentOptions(
# 注意：必须包含 Task 工具，Subagent 是通过 Task 工具调用的
allowed_tools=["Read", "Grep", "Glob", "Task"],
agents={
"code-reviewer": AgentDefinition(
description="专门做代码审查的 Agent，当需要检查代码质量和安全性时使用",
prompt="""你是一个严格的代码审查员，专注于安全性和代码质量。
审查代码时请关注：
- SQL 注入、XSS 等安全漏洞
- 硬编码的密钥或密码
- 未处理的异常
- 不安全的文件操作
请给出具体的问题描述和修复建议。""",
tools=["Read", "Grep", "Glob"], # 只读权限，不能修改文件
model="sonnet",
),
"test-writer": AgentDefinition(
description="专门写测试用例的 Agent，当需要为代码编写测试时使用",
prompt="""你是一个测试专家。为给定的代码编写全面的测试用例。
要求：
- 覆盖正常路径和异常路径
- 包含边界条件测试
- 使用 pytest 框架
- 每个测试函数都要有清晰的文档字符串""",
tools=["Read", "Write", "Bash"], # 需要写文件和运行测试
model="haiku", # 测试代码用便宜的模型就够了
),
},
),
):
if hasattr(message, "result"):
print(message.result)
asyncio.run(main())
TypeScript 版本：
import { query, ClaudeAgentOptions } from "@anthropic-ai/claude-agent-sdk";
async function main() {
for await (const message of query({
prompt: "审查一下 src/ 目录下的代码，找出安全问题",
options: {
// 注意：必须包含 Task 工具
allowedTools: ["Read", "Grep", "Glob", "Task"],
agents: {
"code-reviewer": {
description: "专门做代码审查的 Agent，当需要检查代码质量和安全性时使用",
prompt: `你是一个严格的代码审查员，专注于安全性和代码质量。
审查代码时请关注：
- SQL 注入、XSS 等安全漏洞
- 硬编码的密钥或密码
- 未处理的异常
- 不安全的文件操作
请给出具体的问题描述和修复建议。`,
tools: ["Read", "Grep", "Glob"],
model: "sonnet",
},
"test-writer": {
description: "专门写测试用例的 Agent，当需要为代码编写测试时使用",
prompt: `你是一个测试专家。为给定的代码编写全面的测试用例。`,
tools: ["Read", "Write", "Bash"],
model: "haiku",
},
},
},
})) {
if (message.type === "result") {
console.log(message.result);
}
}
}
main();
关键点：
allowedTools（或allowed_tools）里必须包含"Task"，因为主 Agent 是通过 Task 工具来调用 Subagent 的- 每个 Subagent 的
tools里不要包含"Task"，因为 Subagent 不能再派自己的 Subagent（只有一层）
方式二：文件系统定义
如果你的 Subagent 配置比较复杂（比如 prompt 很长），可以把它定义成一个 Markdown 文件，放在 .claude/agents/ 目录下。
文件：.claude/agents/code-reviewer.md
---
description: 专门做代码审查的 Agent，当需要检查代码质量和安全性时使用
tools:
- Read
- Grep
- Glob
model: sonnet
---
# 代码审查员
你是一个严格的代码审查员，有 10 年的代码审查经验。
## 你的职责
1. 检查代码的安全性
- SQL 注入
- XSS 攻击
- CSRF 漏洞
- 硬编码的密钥
2. 检查代码质量
- 函数是否太长（超过 50 行就要拆分）
- 变量命名是否清晰
- 是否有重复代码
- 错误处理是否完善
3. 检查代码风格
- 是否符合项目的编码规范
- 注释是否充分
- 导入是否整理
## 输出格式
请按以下格式输出审查结果：
### 严重问题（必须修复）
- [文件名:行号] 问题描述
### 建议改进（可以优化）
- [文件名:行号] 建议内容
### 总体评价
给出 1-10 的评分，并说明理由。
这种方式的好处是：
- Prompt 可以写得很详细，不受代码格式限制
- 方便非技术人员编辑和维护
- 可以用 Git 管理版本
- 在 Claude Code CLI 中也能直接使用
注意： 如果代码里定义了和文件系统同名的 Agent，代码里的定义优先。
方式三：内置通用 Agent
即使你不定义任何自定义 Subagent，只要 allowedTools 里包含 "Task"，主 Agent 就可以使用内置的 general-purpose（通用）Subagent。
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions
async def main():
async for message in query(
prompt="帮我研究一下 Python 3.12 有哪些新特性，每个特性给一个代码示例",
options=ClaudeAgentOptions(
# 只要有 Task，主 Agent 就能使用通用 Subagent
allowed_tools=["Read", "Bash", "WebSearch", "Task"],
# 不需要定义 agents，会使用内置的通用 Subagent
),
):
if hasattr(message, "result"):
print(message.result)
asyncio.run(main())
这种方式最简单，适合不需要精细控制 Subagent 行为的场景。主 Agent 会自己判断什么时候派子任务，通用 Subagent 会继承主 Agent 的工具和模型设置。
三种方式的对比
| 特性 | 代码定义 | 文件系统定义 | 内置通用 |
|---|---|---|---|
| 灵活性 | 高 | 中 | 低 |
| Prompt 复杂度 | 适合短中 prompt | 适合长 prompt | 无自定义 prompt |
| 工具控制 | 精确控制 | 精确控制 | 继承主 Agent |
| 模型控制 | 可指定 | 可指定 | 继承主 Agent |
| 适合场景 | 程序化使用 SDK | 项目级配置 | 简单委托任务 |
| 优先级 | 最高 | 次之 | 最低 |
13.3 Subagent 的四大属性详解
每个 Subagent 有四个关键属性。搞懂这四个属性，你就搞懂了 Subagent 的全部配置。
属性一：description —— 什么时候用这个 Agent
description 是最重要的属性。主 Agent 根据 description 来决定把任务派给谁。它就像员工的"岗位职责说明"。
好的 description：
# 清晰说明了"做什么"和"什么时候用"
description="Expert code review specialist. Use for quality, security, and maintainability reviews."
# 明确了专业领域和使用场景
description="数据库迁移专家。当需要修改数据库 schema、编写 migration 脚本、或检查数据完整性时使用。"
# 具体列举了能力范围
description="前端性能优化专家。负责分析 Lighthouse 报告、优化 bundle 大小、检查渲染性能和内存泄漏。"
不好的 description：
# 太模糊，主 Agent 不知道什么时候该派给它
description="一个有用的助手"
# 太宽泛，和其他 Agent 分不清
description="写代码的"
# 没有说清什么场景使用
description="处理任务"
小技巧： 如果你发现主 Agent 不把任务派给正确的 Subagent，首先检查 description 写得够不够清楚。
属性二：prompt —— Agent 的"工作手册"
prompt 就是这个 Subagent 的 system prompt，定义了它的角色、行为规范和输出格式。
写好 prompt 的几个要点：
prompt="""你是一个 Python 代码安全审查专家。
## 你的角色
你有 10 年的网络安全经验，专注于 Python Web 应用的安全审计。
## 审查清单
检查以下安全问题：
1. SQL 注入（使用参数化查询了吗？）
2. XSS 攻击（用户输入有转义吗？）
3. 路径遍历（文件路径有验证吗？）
4. 密钥泄露（有硬编码的 API Key 吗？）
5. 不安全的反序列化（用了 pickle.loads 吗？）
## 输出格式
对于每个发现的问题，请按以下格式输出：
- 严重程度：高/中/低
- 文件位置：文件名:行号
- 问题描述：一句话说清楚
- 修复建议：具体怎么改
## 注意事项
- 不要给出模棱两可的建议
- 每个问题都要附带修复代码示例
- 如果没发现问题，也要说明你检查了哪些方面
"""
属性三：tools —— 这个 Agent 能用什么工具
tools 决定了 Subagent 能使用哪些工具。这体现了最小权限原则 —— 只给 Agent 它需要的工具，不多给。
# 只读 Agent：只能看，不能改
tools=["Read", "Grep", "Glob"]
# 代码编辑 Agent：能看也能改
tools=["Read", "Write", "Edit", "Bash"]
# 研究 Agent：能搜索网页
tools=["WebSearch", "WebFetch", "Read"]
# 测试 Agent：能运行命令和读写文件
tools=["Bash", "Read", "Write", "Grep"]
重要提醒：
- 不要在 Subagent 的
tools里放"Task"—— Subagent 不能再派自己的 Subagent - 如果不指定
tools，Subagent 会继承主 Agent 的所有工具 - 工具给得太多会增加安全风险，给得太少 Agent 可能干不了活
属性四：model —— 用哪个大脑
model 可以让你给不同的 Subagent 指定不同的模型。这是成本优化的关键。
agents={
# 复杂的架构分析任务，用最聪明的模型
"architect": AgentDefinition(
description="系统架构分析师",
prompt="分析系统架构并给出改进建议...",
tools=["Read", "Grep", "Glob"],
model="opus", # 最强但最贵
),
# 常规的代码审查，用性价比最高的模型
"reviewer": AgentDefinition(
description="代码审查员",
prompt="检查代码质量...",
tools=["Read", "Grep", "Glob"],
model="sonnet", # 性价比之王
),
# 简单的格式化任务，用最便宜的模型
"formatter": AgentDefinition(
description="代码格式化助手",
prompt="格式化代码输出...",
tools=["Read"],
model="haiku", # 又快又便宜
),
}
| 模型 | 特点 | 适合的 Subagent 任务 | 大概成本 |
|---|---|---|---|
| opus | 最聪明，推理能力最强 | 架构设计、复杂分析 | $$$ |
| sonnet | 聪明 + 快速，性价比高 | 代码审查、bug 查找 | $$ |
| haiku | 最快最便宜 | 格式化、简单分类、摘要 | $ |
| inherit | 继承主 Agent 的模型 | 和主 Agent 一样的任务 | 取决于主 Agent |
13.4 Subagent 的工作机制
理解 Subagent 的内部工作流程，能帮你更好地设计多 Agent 系统。
完整的执行流程
用户 ──→ 主 Agent ──→ 读取任务、思考
│
├──→ "这个任务适合 code-reviewer 来做"
│
├──→ 调用 Task 工具，指定 agent="code-reviewer"
│ 传入任务描述："审查 auth.py 的安全性"
│
│ ┌──────────────────────────────┐
│ │ Subagent: code-reviewer │
│ │ │
│ │ 1. 接收任务 │
│ │ 2. 用 Read 读取 auth.py │
│ │ 3. 用 Grep 搜索敏感模式 │
│ │ 4. 分析、整理结果 │
│ │ 5. 返回审查报告 │
│ └──────────────┬───────────────┘
│ │
├──← 收到 Subagent 的结果
│
├──→ 可能继续派其他 Subagent
│
└──→ 综合所有结果，返回给用户
关键特点
1. 上下文隔离
Subagent 和主 Agent 的上下文是完全隔离的。这意味着：
- Subagent 看不到主 Agent 之前的对话内容
- 主 Agent 看不到 Subagent 内部的"思考过程"
- 主 Agent 只能看到 Subagent 最终返回的结果
这就像公司里，老板把任务分配给员工后，员工怎么做是员工的事，老板只看最终成果。
2. 通过 Task 工具调用
主 Agent 调用 Subagent 并不是什么魔法，它就是调用了一个叫 Task 的内置工具。从主 Agent 的视角看：
主 Agent 的思考过程：
"用户想要代码审查，我有一个专门的 code-reviewer Agent，让它来做。"
主 Agent 调用的工具：
工具名: Task
参数: {
"description": "审查 src/auth.py 文件的安全性，关注 SQL 注入和 XSS 漏洞",
"agent": "code-reviewer" // 指定使用哪个 Subagent
}
Subagent 返回结果后：
Task 工具的返回值 = Subagent 的最终输出
3. 自动选择 vs 手动指定
主 Agent 有两种方式选择 Subagent：
# 方式 A：自动选择
# 主 Agent 根据任务内容和各 Subagent 的 description 自动匹配
prompt = "帮我检查一下代码质量"
# → 主 Agent 会自动选择 description 里提到"代码审查"的 Subagent
# 方式 B：手动指定
# 在 prompt 里明确提到 Subagent 的名字
prompt = "使用 code-reviewer agent 来检查 auth.py 的安全性"
# → 主 Agent 会直接调用名为 code-reviewer 的 Subagent
13.5 并行 vs 串行
串行：一个做完再轮下一个
串行适合任务之间有依赖关系的场景 —— B 的输入依赖 A 的输出。
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition
async def serial_review():
"""串行审查：先审查代码，再根据审查结果写修复方案"""
async for message in query(
prompt="""请按以下步骤执行：
1. 先使用 code-reviewer agent 审查 src/auth.py 的代码
2. 拿到审查结果后，使用 bug-fixer agent 根据审查发现的问题编写修复代码
注意：必须先完成步骤1，才能执行步骤2，因为修复需要基于审查结果。""",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Write", "Edit", "Task"],
agents={
"code-reviewer": AgentDefinition(
description="代码审查专家。审查代码并列出所有问题。",
prompt="你是代码审查员。仔细审查代码，列出所有问题及其位置。",
tools=["Read", "Grep", "Glob"],
model="sonnet",
),
"bug-fixer": AgentDefinition(
description="代码修复专家。根据审查报告修复代码中的问题。",
prompt="你是代码修复专家。根据给定的问题列表，编写修复代码。",
tools=["Read", "Write", "Edit"],
model="sonnet",
),
},
),
):
if hasattr(message, "result"):
print(message.result)
asyncio.run(serial_review())
串行执行的流程：
主 Agent
│
├──→ 调用 code-reviewer（等待完成...）
│ └──→ 返回审查报告
│
├──→ 把审查报告传给 bug-fixer（等待完成...）
│ └──→ 返回修复方案
│
└──→ 汇总结果，返回给用户
并行：多个 Agent 同时干活
并行适合任务之间没有依赖关系的场景 —— A 和 B 可以各干各的。
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition
async def parallel_review():
"""并行审查：代码审查和安全扫描同时进行"""
async for message in query(
prompt="""请同时执行以下两个独立任务：
1. 使用 code-reviewer agent 审查 src/ 目录下所有 Python 文件的代码质量
2. 使用 security-scanner agent 扫描 src/ 目录下所有 Python 文件的安全漏洞
这两个任务是独立的，可以并行执行。最后请汇总两个 agent 的结果，给出一份综合报告。""",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Grep", "Glob", "Task"],
agents={
"code-reviewer": AgentDefinition(
description="代码质量审查专家。检查代码风格、可维护性和最佳实践。",
prompt="你是代码质量专家。审查代码风格、可维护性、命名规范等方面。",
tools=["Read", "Grep", "Glob"],
model="sonnet",
),
"security-scanner": AgentDefinition(
description="安全漏洞扫描专家。检查代码中的安全问题。",
prompt="你是安全专家。扫描代码中的安全漏洞，包括注入攻击、认证缺陷等。",
tools=["Read", "Grep", "Glob"],
model="sonnet",
),
},
),
):
if hasattr(message, "result"):
print(message.result)
asyncio.run(parallel_review())
并行执行的流程：
主 Agent
│
├──→ 同时调用 code-reviewer 和 security-scanner
│ │ │
│ ├──→ 审查代码质量 ├──→ 扫描安全漏洞
│ └──→ 返回质量报告 └──→ 返回安全报告
│ │ │
├──← 收集两份报告 ←──┴────────────────────────┘
│
└──→ 汇总成一份综合报告，返回给用户
如何触发并行执行？
关键在于你给主 Agent 的 prompt。如果你在 prompt 里说"先做 A 再做 B"，主 Agent 就会串行执行。如果你说"同时做 A 和 B"或者"A 和 B 是独立的"，主 Agent 通常会尝试并行执行。
Claude 足够聪明，它会根据任务的性质自动判断。但如果你想确保并行，最好在 prompt 里明确说明。
13.6 Subagent 的限制
在你兴高采烈地搞多 Agent 架构之前，先了解一下这些限制，免得踩坑。
限制一：只有一层，不能套娃
Subagent 不能再创建自己的 Subagent。也就是说，你不能搞出这样的结构：
主 Agent
└── Subagent A
└── Sub-Subagent B ← 这是不行的！
SDK 只支持一层深度。所以不要在 Subagent 的 tools 里放 "Task"。
如果真的需要多层怎么办？ 把所有 Agent 都放在主 Agent 下面，让主 Agent 来协调它们之间的依赖关系。
限制二：上下文隔离
Subagent 和主 Agent 的上下文是完全隔离的：
- Subagent 看不到主 Agent 之前的对话历史
- Subagent 看不到其他 Subagent 的执行结果（除非主 Agent 在 prompt 里传给它）
- 主 Agent 看不到 Subagent 的中间思考过程，只能看到最终结果
这意味着你需要在调用 Subagent 的时候，把所有必要的上下文信息都通过任务描述传递过去。
限制三：不能直接和用户对话
Subagent 不能使用 AskUserQuestion 工具来直接问用户问题。如果 Subagent 遇到拿不准的事情，它只能在结果里标注出来，让主 Agent 来决定是否需要问用户。
限制四：Windows 上的命令行长度限制
在 Windows 上，Subagent 的 prompt 通过命令行传递，受限于 Windows 的命令行长度限制（8191 个字符）。如果你的 prompt 很长，在 Windows 上可能会报错。
解决办法：
- 使用文件系统定义（方式二），把长 prompt 写在
.claude/agents/目录下的文件里 - 精简 prompt，只保留核心指令
限制五：成本可能翻倍
每个 Subagent 都是一个独立的 Agent 循环，有自己的 API 调用。多个 Subagent 并行运行时，Token 消耗会成倍增加。所以要合理规划：
- 简单任务不需要派 Subagent
- 能用一个 Agent 搞定的，就不要用两个
- 给不太重要的 Subagent 用便宜的模型（haiku）
13.7 实战：代码审查团队
现在我们来搭建一个完整的"代码审查团队"，包含三个角色。
团队组成
项目经理（主 Agent）
├── 代码审查员（Code Reviewer）—— 检查代码质量
├── Bug 猎手（Bug Hunter）—— 找潜在的 Bug
└── 测试建议师（Test Suggester）—— 建议测试用例
完整代码
import asyncio
import json
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition
async def code_review_team(target_path: str):
"""
代码审查团队：三个 Agent 协作完成代码审查
参数:
target_path: 要审查的代码目录或文件路径
"""
async for message in query(
prompt=f"""你是项目经理，负责协调代码审查工作。
请对 {target_path} 进行全面的代码审查，具体步骤：
1. 首先使用 code-reviewer agent 检查代码质量（风格、可维护性、最佳实践）
2. 同时使用 bug-hunter agent 查找潜在的 Bug 和逻辑错误
3. 等以上两个 agent 完成后，使用 test-suggester agent 根据代码和发现的问题建议测试用例
注意：步骤1和步骤2可以并行执行（它们是独立的），步骤3需要等前两步完成。
最后，请汇总三个 agent 的报告，生成一份完整的代码审查报告，格式如下：
## 代码审查报告
### 1. 代码质量
（code-reviewer 的发现）
### 2. 潜在 Bug
（bug-hunter 的发现）
### 3. 建议的测试用例
（test-suggester 的建议）
### 4. 总体评价
（你的综合评价和优先级建议）
""",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Grep", "Glob", "Task"],
agents={
"code-reviewer": AgentDefinition(
description="代码质量审查专家。检查代码风格、可维护性、命名规范和最佳实践。当需要评估代码质量时使用。",
prompt="""你是一个资深的代码质量审查员。
## 审查维度
1. **代码风格**
- 命名是否清晰（变量名、函数名、类名）
- 缩进和格式是否一致
- 注释是否充分
2. **可维护性**
- 函数是否太长（>50行应拆分）
- 类的职责是否单一
- 是否有重复代码（DRY原则）
3. **最佳实践**
- 是否正确处理了异常
- 资源是否正确释放（文件句柄、数据库连接等）
- 是否使用了已弃用的 API
## 输出要求
- 按严重程度排序：严重 > 中等 > 轻微
- 每个问题标注文件名和行号
- 给出具体的改进建议""",
tools=["Read", "Grep", "Glob"],
model="sonnet",
),
"bug-hunter": AgentDefinition(
description="Bug 查找专家。分析代码逻辑，找出潜在的 Bug 和运行时错误。当需要查找 Bug 时使用。",
prompt="""你是一个经验丰富的 Bug 猎手，专门找代码里隐藏的问题。
## 搜索维度
1. **逻辑错误**
- 条件判断是否正确
- 循环是否有边界问题
- 空值/空列表处理
2. **并发问题**
- 竞态条件
- 死锁风险
- 线程安全
3. **边界情况**
- 空输入
- 超大输入
- 特殊字符
- 类型不匹配
4. **安全漏洞**
- 注入攻击
- 权限绕过
- 信息泄露
## 输出要求
- 每个 Bug 给出：位置、描述、严重程度、复现条件
- 如果可能，给出修复建议
- 按严重程度排序""",
tools=["Read", "Grep", "Glob"],
model="sonnet",
),
"test-suggester": AgentDefinition(
description="测试用例建议专家。根据代码分析结果建议需要编写的测试用例。当需要测试建议时使用。",
prompt="""你是一个测试策略专家。根据代码和已发现的问题，建议需要编写的测试用例。
## 测试策略
1. **单元测试**
- 每个公开函数至少一个测试
- 覆盖正常路径和异常路径
- 边界值测试
2. **集成测试**
- 模块间交互测试
- 数据库操作测试
- API 端点测试
3. **针对已发现问题的回归测试**
- 为每个发现的 Bug 写一个能复现的测试
- 确保修复后这些测试能通过
## 输出格式
用 pytest 风格写出测试用例的代码框架，包含：
- 测试函数名
- 文档字符串说明测试目的
- 核心的 assert 断言
- 必要的 mock/fixture""",
tools=["Read", "Grep", "Glob"],
model="haiku", # 测试建议用 haiku 就够了
),
},
),
):
# 打印流式输出
if hasattr(message, "content"):
for block in message.content:
if hasattr(block, "text"):
print(block.text, end="", flush=True)
if hasattr(message, "result"):
print("\n\n===== 最终报告 =====")
print(message.result)
# 运行审查
asyncio.run(code_review_team("src/"))
TypeScript 版本
import { query } from "@anthropic-ai/claude-agent-sdk";
async function codeReviewTeam(targetPath: string) {
for await (const message of query({
prompt: `你是项目经理，负责协调代码审查工作。
请对 ${targetPath} 进行全面的代码审查：
1. 使用 code-reviewer agent 检查代码质量
2. 同时使用 bug-hunter agent 查找潜在 Bug
3. 等以上完成后，使用 test-suggester agent 建议测试用例
最后汇总成一份完整的代码审查报告。`,
options: {
allowedTools: ["Read", "Grep", "Glob", "Task"],
agents: {
"code-reviewer": {
description: "代码质量审查专家。检查代码风格、可维护性和最佳实践。",
prompt: "你是资深代码审查员。检查命名规范、函数长度、重复代码、异常处理等。按严重程度排序输出。",
tools: ["Read", "Grep", "Glob"],
model: "sonnet",
},
"bug-hunter": {
description: "Bug 查找专家。分析代码逻辑找出潜在 Bug。",
prompt: "你是 Bug 猎手。查找逻辑错误、并发问题、边界情况和安全漏洞。给出位置、描述和修复建议。",
tools: ["Read", "Grep", "Glob"],
model: "sonnet",
},
"test-suggester": {
description: "测试用例建议专家。建议需要编写的测试用例。",
prompt: "你是测试策略专家。建议单元测试、集成测试和回归测试。用 Jest/Vitest 风格输出测试代码框架。",
tools: ["Read", "Grep", "Glob"],
model: "haiku",
},
},
},
})) {
if (message.type === "result") {
console.log(message.result);
}
}
}
codeReviewTeam("src/");
13.8 实战：研究小组
搭建一个"研究小组"，让多个 Agent 并行搜索信息，然后由一个 Agent 汇总成报告。
团队组成
研究组长（主 Agent）
├── 研究员 A（Researcher）—— 并行搜索不同方面的信息
└── 报告撰写员（Synthesizer）—— 汇总所有研究结果
完整代码
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition
async def research_team(topic: str):
"""
研究小组：多个研究员并行搜索，最后汇总成报告
参数:
topic: 要研究的主题
"""
async for message in query(
prompt=f"""你是研究组长，负责协调一个关于"{topic}"的研究项目。
请按以下步骤执行：
## 第一阶段：并行研究（可以同时进行）
1. 使用 researcher agent 搜索"{topic}"的技术原理和核心概念
任务描述："研究 {topic} 的技术原理、核心概念和基本架构"
2. 使用 researcher agent 搜索"{topic}"的实际应用案例
任务描述："研究 {topic} 的实际应用案例、知名项目和成功案例"
3. 使用 researcher agent 搜索"{topic}"的优缺点和未来趋势
任务描述："研究 {topic} 的优点、缺点、局限性和未来发展趋势"
以上三个研究任务是独立的，请并行执行。
## 第二阶段：汇总报告
等所有研究完成后，使用 synthesizer agent 将三份研究结果汇总成一份结构化的研究报告。
最后把报告展示给我。""",
options=ClaudeAgentOptions(
allowed_tools=["Read", "WebSearch", "WebFetch", "Task"],
agents={
"researcher": AgentDefinition(
description="研究员。负责搜索和收集某个主题的相关信息。当需要调研某个话题时使用。",
prompt="""你是一个专业的研究员。你的任务是针对给定的主题进行深入研究。
## 研究方法
1. 使用 WebSearch 搜索相关信息
2. 使用 WebFetch 获取重要页面的详细内容
3. 整理搜索到的信息，去除重复和不相关内容
## 输出要求
- 列出你找到的关键信息点
- 每个信息点标注来源
- 区分"事实"和"观点"
- 如果信息有矛盾，标注出来
## 注意事项
- 只报告你确认的信息，不要编造
- 如果某方面找不到信息，坦诚说明
- 尽量找多个信息源交叉验证""",
tools=["WebSearch", "WebFetch", "Read"],
model="sonnet",
),
"synthesizer": AgentDefinition(
description="报告撰写员。负责将多份研究结果汇总成一份结构化的报告。当需要整合信息时使用。",
prompt="""你是一个专业的报告撰写员。你的任务是将多份独立的研究结果整合成一份完整、连贯的报告。
## 报告结构
# {topic} 研究报告
## 概述
（一段话概括这个主题）
## 1. 技术原理与核心概念
（来自研究一的内容）
## 2. 实际应用案例
（来自研究二的内容）
## 3. 优缺点分析
（来自研究三的内容）
## 4. 未来展望
（综合分析）
## 5. 参考来源
（列出所有信息来源）
## 撰写要求
- 语言通俗易懂
- 重要结论加粗标注
- 数据和案例要标注来源
- 如果不同研究之间有矛盾，在报告中指出""",
tools=["Read"],
model="sonnet",
),
},
),
):
if hasattr(message, "result"):
print(message.result)
# 使用示例
asyncio.run(research_team("WebAssembly 在服务端的应用"))
13.9 实战：翻译工作室
搭建一个"翻译工作室"，实现翻译 -> 校对 -> 排版的流水线。
团队组成
翻译总监（主 Agent）
├── 翻译师（Translator）—— 做初始翻译
├── 校对员（Proofreader）—— 检查翻译质量
└── 排版师（Formatter）—— 格式化输出
完整代码
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition
async def translation_studio(source_file: str, source_lang: str, target_lang: str):
"""
翻译工作室：翻译 -> 校对 -> 排版 的流水线
参数:
source_file: 源文件路径
source_lang: 源语言（如 "英语"）
target_lang: 目标语言（如 "中文"）
"""
async for message in query(
prompt=f"""你是翻译总监，负责协调翻译工作。
目标：将 {source_file} 从{source_lang}翻译成{target_lang}。
请严格按以下流水线执行（每一步都依赖前一步的结果）：
## 步骤1：翻译
使用 translator agent 翻译 {source_file} 的内容。
将原文和翻译结果一起传给下一步。
## 步骤2：校对
使用 proofreader agent 校对步骤1的翻译结果。
把翻译结果和校对意见传给它。
## 步骤3：排版
使用 formatter agent 将校对后的翻译结果排版为最终输出。
最后，请将排版后的最终结果展示给我，并附上校对过程中发现的主要问题摘要。""",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Write", "Task"],
agents={
"translator": AgentDefinition(
description=f"翻译专家。负责将{source_lang}文本翻译成{target_lang}。当需要翻译时使用。",
prompt=f"""你是一个专业的{source_lang}到{target_lang}翻译专家。
## 翻译原则
1. **准确性优先**：忠实原文含义，不增不减
2. **自然流畅**：翻译结果读起来像{target_lang}原生写作
3. **保留术语**：技术术语保留原文，括号内加{target_lang}解释
4. **风格一致**：保持原文的语气和风格
## 特殊处理
- 代码片段不翻译，保持原样
- 专有名词首次出现时标注原文
- URL 和邮箱地址不翻译
- 表格格式保持不变
## 输出要求
直接输出翻译后的完整文本，不需要额外说明。""",
tools=["Read"],
model="sonnet",
),
"proofreader": AgentDefinition(
description="翻译校对专家。负责检查翻译的准确性和流畅性。当需要校对翻译时使用。",
prompt=f"""你是一个{target_lang}翻译校对专家。
## 校对清单
1. **准确性**：翻译是否忠实原文？有没有漏译或误译？
2. **流畅性**：{target_lang}表达是否自然？有没有翻译腔？
3. **一致性**：同一个术语是否始终使用相同的翻译？
4. **语法**：{target_lang}语法是否正确？
5. **格式**：标点符号是否使用了{target_lang}标点？
## 输出要求
1. 先列出发现的问题（附带修改建议）
2. 然后给出校对后的完整修正版本
3. 最后给出翻译质量评分（1-10分）""",
tools=["Read"],
model="sonnet",
),
"formatter": AgentDefinition(
description="文档排版专家。负责将文本排版为美观的最终格式。当需要排版时使用。",
prompt="""你是一个文档排版专家。
## 排版规则
1. 标题层级清晰，使用 Markdown 格式
2. 中英文之间加空格
3. 代码块使用正确的语言标记
4. 列表项缩进一致
5. 段落之间有合适的间距
6. 引用使用引用块格式
7. 重点内容使用加粗
## 输出要求
输出排版后的 Markdown 格式文本。""",
tools=["Read"],
model="haiku", # 排版任务用 haiku 就够了
),
},
),
):
if hasattr(message, "result"):
print(message.result)
# 使用示例
asyncio.run(translation_studio(
source_file="docs/README.md",
source_lang="英语",
target_lang="中文"
))
13.10 Agent Swarm 模式
什么是 Agent Swarm？
Agent Swarm（Agent 蜂群）是一种更高级的多 Agent 协作模式。和简单的主-子 Agent 不同，Swarm 模式下的多个 Agent 可以通过共享文件来交换信息，实现更复杂的协作。
这个概念最早由 NanoClaw 项目在实践中提出并实现。
Swarm 和普通 Subagent 的区别
| 特性 | 普通 Subagent | Agent Swarm |
|---|---|---|
| 通信方式 | 主 Agent 传递参数和接收结果 | 通过共享文件交换信息 |
| 协调方式 | 主 Agent 负责协调 | 有专门的协调 Agent |
| 知识共享 | 隔离，不共享 | 通过文件系统共享 |
| 适合场景 | 简单的分工任务 | 复杂的、需要多轮协作的项目 |
Swarm 模式的实现思路
核心思想很简单：用文件系统作为 Agent 之间的"公告板"。
共享工作目录/
├── tasks/ # 任务分配文件
│ ├── task-001.md # 任务1的描述和状态
│ └── task-002.md # 任务2的描述和状态
├── results/ # 各 Agent 的产出
│ ├── reviewer.md # 审查员的报告
│ └── researcher.md # 研究员的报告
└── coordination.md # 协调信息（进度、依赖关系等）
完整示例：Swarm 模式的项目分析
import asyncio
import os
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition
async def project_analysis_swarm(project_path: str):
"""
用 Agent Swarm 模式分析一个项目
多个 Agent 协作，通过共享文件交换信息
"""
# 创建共享工作目录
work_dir = os.path.join(project_path, ".analysis")
os.makedirs(os.path.join(work_dir, "results"), exist_ok=True)
async for message in query(
prompt=f"""你是项目分析团队的协调员。
你需要协调多个专家 Agent 来分析 {project_path} 这个项目。
共享工作目录是 {work_dir}，各 Agent 的分析结果都放在 {work_dir}/results/ 下面。
## 执行计划
### 第一阶段（并行）
1. 使用 architecture-analyst agent 分析项目架构
要求它把结果写到 {work_dir}/results/architecture.md
2. 使用 dependency-analyst agent 分析项目依赖
要求它把结果写到 {work_dir}/results/dependencies.md
3. 使用 code-quality-analyst agent 分析代码质量
要求它把结果写到 {work_dir}/results/code-quality.md
以上三个任务是独立的，请并行执行。
### 第二阶段（串行，等第一阶段全部完成）
4. 使用 report-writer agent 读取 {work_dir}/results/ 下所有分析结果，
生成一份综合的项目分析报告，写到 {work_dir}/results/final-report.md
### 第三阶段
5. 读取 {work_dir}/results/final-report.md 的内容，展示给用户。
""",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Write", "Glob", "Grep", "Bash", "Task"],
cwd=project_path,
agents={
"architecture-analyst": AgentDefinition(
description="项目架构分析专家。分析项目的目录结构、模块划分和设计模式。",
prompt="""你是一个软件架构师。分析项目的架构并输出报告。
请分析：
1. 目录结构和模块划分
2. 使用的设计模式
3. 模块间的依赖关系
4. 架构的优点和不足
把分析结果写到指定的文件中。使用 Markdown 格式。""",
tools=["Read", "Glob", "Grep", "Write"],
model="sonnet",
),
"dependency-analyst": AgentDefinition(
description="依赖分析专家。分析项目的外部依赖和版本情况。",
prompt="""你是一个依赖管理专家。分析项目的依赖情况。
请分析：
1. 直接依赖和间接依赖
2. 是否有过时的依赖
3. 是否有已知安全漏洞的依赖
4. 依赖树的复杂度
把分析结果写到指定的文件中。使用 Markdown 格式。""",
tools=["Read", "Glob", "Bash", "Write"],
model="haiku",
),
"code-quality-analyst": AgentDefinition(
description="代码质量分析专家。统计代码的各项质量指标。",
prompt="""你是一个代码质量分析师。统计和分析代码质量指标。
请分析：
1. 代码总行数、文件数量
2. 各语言占比
3. 注释率
4. 复杂度评估
5. 重复代码检测
把分析结果写到指定的文件中。使用 Markdown 格式，多用表格展示数据。""",
tools=["Read", "Glob", "Grep", "Bash", "Write"],
model="haiku",
),
"report-writer": AgentDefinition(
description="报告撰写专家。读取各分析师的结果，生成综合报告。",
prompt="""你是一个技术报告撰写专家。
你的任务是读取其他分析师写的分析结果文件，然后整合成一份完整的项目分析报告。
## 报告格式
# 项目分析报告
## 执行摘要
（3-5句话概括关键发现）
## 详细分析
### 架构分析
### 依赖分析
### 代码质量
## 风险评估
（综合各方面的风险）
## 改进建议
（按优先级排列的改进建议）
## 附录
（数据表格、引用等）
把报告写到指定的文件中。""",
tools=["Read", "Glob", "Write"],
model="sonnet",
),
},
),
):
if hasattr(message, "result"):
print(message.result)
# 使用示例
asyncio.run(project_analysis_swarm("/path/to/your/project"))
Swarm 模式的核心优势
- 可追溯性：每个 Agent 的分析结果都写在文件里，可以随时查看
- 可恢复性：如果中间某步失败了，已完成的结果不会丢失
- 可组合性：可以灵活地增减 Agent，只需修改协调逻辑
- 可观测性：通过查看文件就能知道每个 Agent 的进度和结果
什么时候用 Swarm？
| 场景 | 用普通 Subagent | 用 Swarm |
|---|---|---|
| 简单的一次性分析 | 推荐 | 杀鸡用牛刀 |
| 需要中间结果持久化 | 不适合 | 推荐 |
| 需要多轮迭代协作 | 勉强可以 | 推荐 |
| Agent 数量 > 5 | 主 Agent 上下文吃紧 | 推荐 |
| 需要人工介入中间步骤 | 不方便 | 推荐（查看文件后继续） |
13.11 检测 Subagent 的调用
在开发和调试多 Agent 系统时，你可能想知道主 Agent 什么时候调用了 Subagent、调用了哪个、结果是什么。
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition
async def monitored_multi_agent():
"""带监控的多 Agent 系统"""
async for message in query(
prompt="审查 src/ 目录的代码",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Grep", "Glob", "Task"],
agents={
"reviewer": AgentDefinition(
description="代码审查专家",
prompt="审查代码质量",
tools=["Read", "Grep", "Glob"],
model="sonnet",
),
},
),
):
# 检查是否是 AssistantMessage
if hasattr(message, "content"):
for block in message.content:
# 检查是否调用了 Task 工具（即调用 Subagent）
if hasattr(block, "type") and block.type == "tool_use":
if block.name == "Task":
print(f"[监控] 主 Agent 正在调用 Subagent")
print(f" 任务描述: {block.input.get('description', 'N/A')}")
agent_name = block.input.get("agent", "general-purpose")
print(f" 使用的 Agent: {agent_name}")
# 检查消息是否来自 Subagent 内部
if hasattr(message, "subagent_id") and message.subagent_id:
print(f" (这条消息来自 Subagent 内部)")
# 最终结果
if hasattr(message, "result"):
print(f"\n[监控] 执行完成")
print(message.result)
asyncio.run(monitored_multi_agent())
13.12 恢复 Subagent
Subagent 执行完后可以被恢复，让它继续之前的工作。这在需要对 Subagent 的结果进行追问时很有用。
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition
async def resumable_subagent():
"""可恢复的 Subagent 示例"""
session_id = None
agent_id = None
# 第一次查询：让 Subagent 分析代码
async for message in query(
prompt="使用 reviewer agent 分析 src/auth.py 的安全性",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Grep", "Task"],
agents={
"reviewer": AgentDefinition(
description="安全审查专家",
prompt="你是安全审查专家，分析代码的安全性。",
tools=["Read", "Grep"],
model="sonnet",
),
},
),
):
# 从 init 消息获取 session_id
if hasattr(message, "session_id"):
session_id = message.session_id
# 从 Task 工具的结果中获取 agent_id
if hasattr(message, "content"):
for block in message.content:
if hasattr(block, "text") and "agentId" in str(block.text):
# 解析出 agent_id（实际代码需要根据具体返回格式解析）
agent_id = block.text # 简化示例
if hasattr(message, "result"):
print("第一次分析结果:", message.result)
# 第二次查询：恢复之前的 Subagent，追问更多细节
if session_id:
async for message in query(
prompt="针对你之前发现的安全问题，给出具体的修复代码示例",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Grep", "Task"],
resume=session_id, # 恢复之前的会话
),
):
if hasattr(message, "result"):
print("追问结果:", message.result)
asyncio.run(resumable_subagent())
恢复 Subagent 的关键点：
- Subagent 的对话历史是持久化存储的，独立于主 Agent
- 即使主 Agent 的上下文被压缩，Subagent 的记录不受影响
- 恢复后的 Subagent 保留完整的之前对话记录
动手练习
练习1：搭建"代码审查团队"
目标： 创建一个包含 reviewer + fixer + tester 三个 Agent 的代码审查团队。
要求：
- reviewer 负责找出问题
- fixer 根据 reviewer 的发现编写修复代码
- tester 为修复的代码编写测试用例
- 三步串行执行（有依赖关系）
- 最后输出一份综合报告
提示：
agents = {
"reviewer": AgentDefinition(
description="...",
prompt="...",
tools=["Read", "Grep", "Glob"], # 只读
model="sonnet",
),
"fixer": AgentDefinition(
description="...",
prompt="...",
tools=["Read", "Write", "Edit"], # 能修改文件
model="sonnet",
),
"tester": AgentDefinition(
description="...",
prompt="...",
tools=["Read", "Write", "Bash"], # 能写测试并运行
model="haiku",
),
}
练习2：搭建"研究小组"
目标： 创建多个 researcher Agent 并行研究不同方面，然后由 synthesizer 汇总。
要求：
- 选择一个技术话题（比如"Rust vs Go 在微服务中的对比"）
- 定义 3 个并行的 researcher，分别研究不同维度
- 1 个 synthesizer 汇总为结构化报告
- 使用 Swarm 模式，中间结果写到文件
挑战： 尝试让 synthesizer 不仅汇总结果，还要标注不同 researcher 之间的观点差异。
练习3：搭建"翻译工作室"
目标： 实现 translator -> proofreader -> formatter 的翻译流水线。
要求：
- 准备一份英文 README 文件
- translator 翻译成中文
- proofreader 校对翻译质量，给出评分和修改意见
- formatter 排版为标准 Markdown 格式
- 如果 proofreader 评分低于 7 分，让 translator 重新翻译
挑战： 如何实现"评分不够就重来"的逻辑？提示 —— 在主 Agent 的 prompt 里设计条件分支。
本章小结
这一章我们学了很多，来回顾一下关键知识点：
为什么需要多 Agent？ 上下文隔离、专业分工、并行效率、安全隔离四大好处。
三种创建方式：
- 代码定义（推荐，最灵活）
- 文件系统定义（适合长 prompt）
- 内置通用 Agent（最简单，但不可控）
四大属性：
description：决定什么时候用这个 Agentprompt：定义 Agent 的行为规范tools：限制 Agent 能用的工具（最小权限原则）model：选择合适的模型（成本优化）
两种协作模式：
- 串行：有依赖关系时，一个做完再做下一个
- 并行：独立任务，同时执行，效率翻倍
Agent Swarm 模式： 通过文件系统共享信息，适合复杂的多 Agent 协作项目。
五个限制要记住： 只有一层、上下文隔离、不能直接和用户对话、Windows 命令行长度限制、成本可能翻倍。
设计原则：
- 不是越多 Agent 越好，够用就行
- description 写清楚，主 Agent 才知道派谁
- 给 Agent 最小权限，安全第一
- 简单任务用便宜模型，省钱
下一章预告
下一章我们学习 Agent Skills —— 教 Agent 学会新技能。如果说 MCP 给 Agent 装上了"新手臂"（工具），那 Skills 就是给 Agent 装上了"新大脑"（知识）。我们会学习如何创建 Skills 文件，让 Agent 掌握项目特有的知识和流程，比如你公司的代码规范、部署流程、审批规则等等。