8.0
热点
AI SCORE
编程提效2026-08-16 19:37
Claude Code 多 Agent 编排：如何构建 AI 代理团队
dev.to · AI#Claude Code#多Agent#架构
Editor brief · 编辑速览
在 Claude Code 中构建 subagent 团队，由 orchestrator 分发任务并汇聚结果，适合复杂项目的生产级 AI 工作流。
一个 AI 智能体只是起点。本文将展示如何在 Claude Code 中构建子代理团队，实现工作分担和并行运作。
Claude Code 课程 - 第 21 课（共 28 课）
本文是 Claude Code 免费课程的一部分——这是面向程序员的最大的波兰语系列教程，共 28 课。从首次启动到自主 AI 智能体。
<-- 全部课程列表
<-- 第 20 课：Claude API（Python/TypeScript） 第 22 课：调度器 /loop
Claude Code 系列 · JSystems
一个 AI 智能体很好。五个专业智能体并行工作则所向披靡。这正是 2026 年最佳 AI 系统的运作方式——你也可以用同样的方式构建自己的系统。
通过本文你将学到：
什么是编排（orchestration）以及为什么一个智能体不足以完成大型任务
如何编写多智能体并行工作的代码（asyncio 逐步讲解）
模式：Orchestrator→子代理、专业智能体、反思循环
多智能体何时真正值得——何时只会增加问题
编排——orchestrator 如何将任务分配给子代理
🤖 Orchestrator claude_code · 主智能体 规划 · 分配 · 聚合
🔍 智能体 1 需求分析 并行工作
⚙ 智能体 2 编写代码 并行工作
🧪 智能体 3 单元测试 并行工作
📋 智能体 4 代码审查 并行工作
Result（收集） 智能体并行工作 → orchestrator 综合结果
单个 AI 智能体有其局限性：一个上下文、一次处理一个任务、一种专业能力。当项目复杂时——你需要的是一个团队。
多智能体系统允许将工作分配给由 orchestrator 协调的专业智能体。这是 Claude 的高级用法——这正是 hobby 项目与生产级 AI 系统之间的区别所在。
开始之前——你应该知道什么
本文涉及 AI 架构的高级模式。建议先阅读关于 Claude API 的文章（第 7 课）并理解 API 调用的基础。你需要：
Python 基础（函数、类、asyncio）
了解什么是 API 以及 Claude API 的工作原理
对异步编程的基本概念（下面会解释）
最简单的方式：通过 prompt 调用子代理
在深入代码之前——Claude Code 可以在不写一行代码的情况下启动子代理。只需要在 prompt 中描述要做什么：
用三个视角同时分析这段代码：
智能体 1（安全性）：检查漏洞和 SQL 注入
智能体 2（性能）：找出瓶颈和 N+1 查询
智能体 3（可读性）：评估命名和文档 将结果汇总为优先级列表。
Claude 自动启动子代理、等待结果并综合回复——零配置。当你需要完全控制（日志、重试、与系统集成）时，再使用 API。下面我们就来讲解。
什么是编排？（经理和团队）
编排（Orchestration，源自英文"指挥"）是一种架构模式，其中一个组件（orchestrator）协调多个其他组件（智能体、服务、函数）的工作。Orchestrator 知道需要做什么、指派任务并收集结果。
类比：项目经理和专家
想象一家建筑公司。项目经理（orchestrator）接到任务："建一栋房子"。他不会亲自动手做所有事——而是联系电工、水管工和泥瓦匠（子代理）。每个人都是各自领域的专家。他们并行工作（电工和水管工可以同时作业）。项目经理收集结果并做出最终决策。AI 中的编排也是如此：一个 Claude-orchestrator 协调多个 Claude-子代理的工作。
为什么这对程序员很重要？当你对一个智能体说"对整个项目做代码审查"时——它有上下文限制（100 个文件放不下）。但 orchestrator 可以将每个文件夹发送给不同的子代理，每个子代理只分析自己的部分，然后 orchestrator 收集结果。这是单个智能体无法达到的规模和quality。
Claude Code 中的子代理是什么
Claude Code 内置了 Task 工具——允许智能体"生成"子代理，这些子代理并行或顺序执行任务。生成（spawning）是创建新进程实例——即：一个带有自己上下文的新 Claude 会话。Orchestrator（主智能体）规划工作，子代理执行工作，orchestrator 聚合结果。
下图可以看到 orchestrator 同时指派三个独立任务。在看代码之前——先理解关键概念：
这意味着什么？"并行"在 AI 智能体语境下的含义：三个任务几乎同时启动，我们等待它们全部完成。整体时间 = 最长任务的耗时——而不是依次累加。如果每个智能体需要 5 秒，并行执行只需 5 秒，而不是 15 秒。
# Orchestrator 看到这个工具：
# Tool: Task
# Description: 运行一个子代理来执行指定任务
# Parameters:
# description: 任务的描述（简短摘要）
# prompt: 详细指令（子代理需要逐步做什么）
# Orchestrator 可以并行启动多个子代理
# 每个都有独立上下文——彼此看不到对方的結果！
Task("分析 auth 模块", "检查 src/auth/ 中的安全测试")
Task("分析 payments 模块", "检查 src/payments/ 中的错误处理")
Task("分析 users 模块", "检查 src/users/ 中的验证")
# 三个并行运行！耗时 = max(auth_time, payments_time, users_time)
# 而不是 auth_time + payments_time + users_time
运行上述代码后，orchestrator"等待"三个分析返回，然后可以将它们合并成一份摘要。使用结果前确保每个智能体都有返回内容。
asyncio——面向初学者的异步编程
asyncio 是什么意思？想象餐厅里的服务员。同步服务员（普通 Python）走到厨房、等待食物做好、回来、再服务下一桌。异步服务员（asyncio）接受第 1 桌的点单、送到厨房——在厨房做饭的同时去服务第 2 桌和第 3 桌——等食物好了再回来取。asyncio 让 Python 在等待 API 响应时可以"处理其他事务"。
asyncio 关键字详解——每个的作用：
async def - "此函数可以被暂停和恢复"。就像接听电话时保持通话去接第二个来电。
await - "等待这个结果，但不要阻塞整个程序"。就像分配任务后继续做其他事情。
asyncio.gather(*tasks) - "同时启动所有任务并等待每个完成"。就像 fork-join：分叉到多条路径、等待每条到达终点、收集结果。
Orchestrator → 子代理模式——逐步讲解
下面的示例展示了一个 orchestrator 启动 4 个子代理进行代码审查——每个专注于不同方面，全部并行工作。让我们逐步讲解。
第 1 步：定义子代理函数——这是每个专家的"模板"。每个子代理收到自己的 task_description 来确定其专业方向，以及 context 即要分析的材料。
第 2 步：创建 4 个任务的列表——每个对应不同的"专家"。此时还没有任何一个启动，只是计划。
第 3 步：asyncio.gather(*tasks) 同时启动所有 4 个并等待每个完成。程序在此处暂停。
第 4 步：Orchestrator（这次用更贵的模型）收到 4 个独立结果，合并成一份连贯的摘要。
import anthropic
import asyncio # 异步编程库
client = anthropic.Anthropic()
# 第 1 步：子代理模板
# "async def" = 异步函数 - 等待时可以暂停
async def run_subagent(task_description: str, context: str) -> str:
"""为特定任务运行子代理"""
# 每个子代理都是独立的 API 调用 - 独立上下文、独立"头脑"
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=2048,
# System prompt 定义子代理的专业方向
system=f"""你是执行以下任务的专家：{task_description}。
执行任务并以 JSON 格式返回结果。""",
messages=[{"role": "user", "content": context}]
)
return response.content[0].text
async def orchestrate_code_review(pr_diff: str):
"""编排器，启动并行的代码审查"""
# 步骤 2：任务列表 - 4 个专家，每个专注不同领域
tasks = [
run_subagent("安全分析", pr_diff), # 查找漏洞
run_subagent("性能分析", pr_diff), # 查找瓶颈
run_subagent("规范一致性分析", pr_diff), # 检查代码风格
run_subagent("测试覆盖率分析", pr_diff), # 检查测试是否充分
]
# 步骤 3：asyncio.gather(*tasks) = "并行运行所有任务，等待完成"
# 这就像同时打开 4 个浏览器标签页，而不是逐个打开
results = await asyncio.gather(*tasks)
# 步骤 4：编排器收到 4 个独立分析结果，合并为一份
# 这里使用更贵的模型（Opus），因为这是更重要的综合任务
aggregation_prompt = f"""
你收到了对同一 PR 的 4 份独立分析：
1. 安全性：{results[0]}
2. 性能：{results[1]}
3. 规范：{results[2]}
4. 测试：{results[3]}
请创建一份连贯的代码审查总结，包含按优先级排列的意见。
"""
final = client.messages.create(
# 编排器可以使用更贵/更智能的模型，因为它做更复杂的综合工作
model="claude-opus-4-7",
max_tokens=1024,
messages=[{"role": "user", "content": aggregation_prompt}]
)
return final.content[0].text
结果：4 个智能体并行启动，2.2 秒完成，而非串行的约 8 秒
运行后：终端会短暂"沉默"——这是正常的，4 个智能体正在并行工作。几秒钟后你会收到合并后的结果。你可以在 gather 之后添加 print(f"Agent {i} skończył") 来确认所有 4 个结果都已返回。
你想在现场构建这样的系统吗？由讲师指导？
Claude Code 培训 - 3 天：基础、工具、多智能体系统。保证开课日期。
查看培训日期
不要一个通才，而是几个专才。每个智能体收到不同的 system prompt，定义其角色和"个性"。流水线：一个接一个传递结果。
为什么专业化效果更好？当智能体只有一个清晰的目标（"查找安全漏洞"）时，它的上下文更集中，响应也更精准。就像全科医生和专科医生的区别——两者都知道很多，但专科医生会问不同的问题、注意到不同的东西。
步骤 1：定义专家字典 - 每个专家有自己的 system prompt 和选定的模型。
步骤 2：run_agent 函数获取配置并调用 API。
步骤 3：串行流水线 - coder 写代码，reviewer 评审，documenter 仅在代码通过 review 后才写文档。
# 步骤 1：每个专家的配置字典
AGENTS = {
"coder": {
"system": "你是一名高级 Python 开发者。编写简洁、可测试的代码。",
"model": "claude-sonnet-4-6" # 写代码需要完整能力的模型
},
"reviewer": {
"system": "你是一名安全工程师。查找漏洞和安全错误。",
"model": "claude-sonnet-4-6"
},
"documenter": {
"system": "你编写简洁的技术文档。关注'为什么'而非'是什么'。",
"model": "claude-haiku-4-5" # 更便宜的模型足以完成简单的文档任务
}
}
# 步骤 2：运行特定专家的函数
def run_agent(agent_name: str, task: str) -> str:
"""运行特定专家智能体"""
config = AGENTS[agent_name] # 获取专家的配置
response = client.messages.create(
model=config["model"], # 每个专家可以使用不同的模型
max_tokens=1024,
system=config["system"], # System prompt 定义专业化方向
messages=[{"role": "user", "content": task}]
)
return response.content[0].text
# 步骤 3：串行流水线 - coder -> reviewer -> documenter
# 每个后续智能体收到前一个的结果
code = run_agent("coder", f"实现功能：{feature_description}")
review = run_agent("reviewer", f"检查安全性：{code}")
# 条件：仅当代码通过安全 review 后才生成文档
if "PASS" in review:
docs = run_agent("documenter", f"为以下代码编写文档：{code}")
结果：串行流水线 Coder -> Reviewer (PASS) -> Documenter
如果你想验证流水线是否正常工作：在第一次调用后添加 print(f"Coder 返回了 {len(code)} 个字符")。Reviewer 应该返回包含 "PASS" 或 "FAIL" 的文本——你可以打印 review[:100] 来查看响应的开头。
Reflection 是一种智能体在将结果传递出去之前，在循环中评估和改进自身输出的模式。类似于"发送前先检查你的文本"。
类比：想象你给员工一个任务。你不是接受他带来的第一个结果，而是说："自己检查一下，评估它是否够好，如果不够——改进它"。只有当他本人认为工作完成得好时，你才接受结果。AI 中的 Reflection 正是这样自我批评和改进的循环。
步骤 1：智能体执行第一个解决方案。
步骤 2：第二个智能体（或使用不同 prompt 的同一智能体）评估结果 - 返回 ACCEPTED 或 REJECTED 并附带原因。
步骤 3：如果是 REJECTED - 根据批评意见进行改进。最多循环 max_iterations 次，以避免无限循环。
def agent_with_reflection(task: str, max_iterations: int = 3) -> str:
# 步骤 1：第一次尝试 - 生成解决方案
result = run_agent("coder", task)
# 反思循环 - 最多 max_iterations 次
for i in range(max_iterations):
# 步骤 2：第二个智能体评估结果
critique = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=512,
messages=[{
"role": "user",
"content": f"""评估这段代码是否正确解决了问题。
问题：{task}
解决方案：{result}
回答：如果解决方案良好则 ACCEPTED，
如果需要改进则 REJECTED: [原因]。"""
}]
).content[0].text
# 如果评估为正面 - 退出循环，返回结果
if "ACCEPTED" in critique:
break
# 步骤 3：如果评估为负面 - 根据批评意见改进
result = run_agent("coder", f"{task}\n\n之前的解决方案有问题：{critique}\n请修复。")
return result # 返回最后一次被接受的或最好的结果
结果：反思循环 - 第 1 次迭代 REJECTED，第 2 次迭代改进后 ACCEPTED
如何验证 reflection 是否正常工作？在循环内部添加 print(f"迭代 {i}：{critique[:50]}") - 你会看到智能体如何从之前的错误中学习。
何时使用 multi-agent，何时使用 single agent
Multi-agent 并不总是更好 - 每个 subagent 都意味着额外的 token 消耗和额外的调试复杂度。原则：当收益（并行、专业化）明显超过成本时才使用 multi-agent。
最常见的错误：认为 subagent B 会自动知道 subagent A 做了什么。每个 subagent 有独立的上下文 - 如果你不明确地把其他智能体的结果传递给它，它就不会知道。始终通过编排器显式传递结果。
正确做法 - 编排器显式传递结果：
# 错误 - agent B 不知道 agent A 做了什么：
result_a = run_subagent("agent-A", code)
result_b = run_subagent("agent-B", code) # 缺少上下文！
# 正确 - 编排器显式传递结果：
security = run_subagent("安全专家",
f"分析以下代码的漏洞：{code}")
performance = run_subagent("性能专家",
f"分析性能：{code}\n"
f"来自 agent-A 的上下文：{security}\n"
f"不要提出会降低安全性的改动。")
final = run_subagent("编排器",
f"合并结果：\n安全性：{security}\n性能：{performance}")
Orchestrator 是唯一能看见所有结果的地方。Subagenci 之间只能通过它来通信，永远不能直接交流。
错误 #2：reflection loop 中没有设置迭代上限
如果没有 max_iterations，Reflection 会无限运行——尤其当"ACCEPTED"标准过于严苛时。务必设置上限，并确保退出条件是可以达成的。
错误 #3：为了"酷"而构建 multi-agent
如果你添加第二个智能体，却没有它要解决的特定问题——这说明你根本不需要 multi-agent。额外的智能体意味着额外的成本、额外的故障点，以及更难以调试的结构。先从一个智能体开始，只有在有具体理由时才添加下一个。
Context 污染 — subagenci 如果不显式传递，就看不到彼此的结果。如果 agent A 发现了一个 bug，agent B 不会知道——你得自己在 prompt 里传递这条信息。
成本 — N 个智能体 = N × token 费用。有 5 个 subagents 时，你付的费用是单个智能体的 5 倍。要监控消耗量。
调试 — 如果某个 subagent 失败了怎么办？你需要对每个 subagent 单独记录日志，并在 orchestrator 层面处理错误。
无限循环 — orchestrator 和 subagent 会互相"ping"对方，永不停止。务必设置 max_iterations 和退出条件。
给初级开发者的黄金法则： 先从一个智能体开始。只有当你发现了 multi-agent 能解决的特定问题时，才添加第二个。不要为了"酷"而去构建多智能体系统——每多一个智能体，就多一层复杂性和成本。
一旦你掌握了 orchestration 和 multi-agent patterns——那些原本需要花几个小时的任务突然只需要几分钟。你的代码会做得更多、更快、更好，远超单个智能体所能提供的。这就是竞争者没有的优势。
智能体团队确实比单个智能体更能干——但必须遵守以下原则：
在 .claude/agents/ 中定义 subagents。 在那里创建一个描述角色定位的文件（如"reviewer"、"tester"）——Claude Code 会自动调用它，当任务匹配时。
Multi-agent 只用于独立任务。 彼此依赖的步骤用一个智能体完成。真正互不干扰的任务才并行跑。
Reflection 模式。 让第二个智能体在第一个智能体完成工作后进行批评，再做确认——质量会显著提升。
每个 subagent = 独立的上下文。 任务分离可以减轻主上下文的负担，限制长会话中的"遗忘"问题。
JSystems 博客 Newsletter
每一篇新课直接发送到你的邮箱
新课程在每周一和周四上线。订阅 JSystems 博客的 Newsletter——每次发布后你都会立即收到通知。
订阅 Newsletter
3 天课程：第 1 天 = 基础 + CLAUDE.md，第 2 天 = MCP + 工具，第 3 天 = subagenci + orchestration。这是你在文档里找不到的进阶知识。名额有限，先到先得。
Claude Code 培训
<-- 第 20 课：Claude API (Python/TypeScript)
第 22 课：Scheduler /loop
Claude Code 全套课程列表