1. 引言

Claude Code 是 Anthropic 推出的 AI 编程助手,面向代码理解、修改、调试和项目级协作等开发场景进行了深度优化。随着项目规模扩大,单 Agent 往往难以高效完成复杂任务,多 Agent 协作成为提升开发效率的关键路径。本文将从原理出发,结合可运行的代码实践,详细介绍如何在 Claude Code 中实现多 Agent 协作。

2. 多 Agent 架构基础

多 Agent 架构的核心思想是将复杂任务拆解为多个子任务,由不同 Agent 分别负责,再通过协调机制汇总结果。在 Claude Code 中,多 Agent 实现主要依赖以下三种方式:

  • Subagent 机制:Claude Code 内置的子代理能力,主 Agent 可以委派任务给专门的 Subagent 执行。
  • 多会话协作:通过多个 Claude Code 会话并行处理不同模块,再手动或脚本化合并结果。
  • 外部编排框架:使用 Python、Node.js 等语言编写编排脚本,调用 Claude Code CLI 或 API 实现多 Agent 调度。

选择哪种方式取决于任务复杂度、协作紧密度和自动化程度。下面分别给出代码实践。

3. 使用内置 Subagent 实现多 Agent

Claude Code 从较新版本开始支持 Subagent 功能。主 Agent 可以在对话中直接要求创建 Subagent 处理特定任务,例如代码审查、测试编写或文档生成。以下是一个典型的 Subagent 使用示例:

# 在 Claude Code 交互中直接使用
# 主 Agent 指令示例:
# "创建一个 Subagent 专门负责审查 src/auth/login.ts 的代码质量,
#  重点关注安全漏洞和边界条件,输出审查报告。"
也可以使用 --subagent 参数启动独立子代理
claude --subagent "审查 src/auth/login.ts 的安全问题" 
--allowedTools "Read, Grep, Glob"
--output-format json

Subagent 的优势在于上下文隔离:每个 Subagent 拥有独立的上下文窗口,不会互相污染,主 Agent 负责汇总和决策。这种方式适合任务边界清晰、子任务相对独立的场景。

4. 基于 Claude Code CLI 的多 Agent 编排

当需要更精细的控制时,可以通过脚本调用 Claude Code CLI 实现多 Agent 编排。下面给出一个 Python 示例,演示如何并行启动多个 Agent 处理不同模块的代码审查任务:

import subprocess
import json
import concurrent.futures
def run_agent(module_path, focus):
"""在独立进程中运行 Claude Code 子代理"""
prompt = f"请审查 {module_path} 的代码,重点关注 {focus},输出 JSON 格式的审查报告。"
cmd = [
"claude",
"--print",
"--output-format", "json",
prompt
]
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
return {
"module": module_path,
"focus": focus,
"output": result.stdout,
"error": result.stderr
}
def main():
tasks = [
("src/auth/login.ts", "安全漏洞和输入校验"),
("src/api/order.ts", "异常处理和边界条件"),
("src/db/connection.ts", "连接池管理和超时设置"),
]
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(run_agent, m, f) for m, f in tasks]
    for future in concurrent.futures.as_completed(futures):
        report = future.result()
        print(f"模块 {report['module']} 审查完成")
        if report["error"]:
            print(f"错误: {report['error']}")
        else:
            print(report["output"])
if name == "main":
main()

这个脚本的核心思路是:将不同模块的审查任务分发给独立的 Claude Code 进程,利用线程池实现并行执行,最后统一收集结果。这种方式适合任务之间无强依赖、可以并行处理的场景。

5. 基于 Anthropic API 的多 Agent 协作框架

对于需要深度协作、共享中间状态的复杂场景,推荐直接使用 Anthropic API 构建多 Agent 框架。下面给出一个 Python 实现,演示如何构建一个简单的多 Agent 协作系统:

import anthropic
import json
from typing import List, Dict
class Agent:
def init(self, name: str, system_prompt: str, client: anthropic.Anthropic):
self.name = name
self.system_prompt = system_prompt
self.client = client
self.messages = []
def run(self, task: str) -> str:
    """执行单个任务"""
    self.messages.append({"role": "user", "content": task})
    response = self.client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        system=self.system_prompt,
        messages=self.messages
    )
    result = response.content[0].text
    self.messages.append({"role": "assistant", "content": result})
    return result
class Orchestrator:
def init(self, api_key: str):
self.client = anthropic.Anthropic(api_key=api_key)
self.agents: Dict[str, Agent] = {}
def register_agent(self, name: str, system_prompt: str):
    self.agents[name] = Agent(name, system_prompt, self.client)
def delegate(self, agent_name: str, task: str) -> str:
if agent_name not in self.agents:
raise ValueError(f"Agent {agent_name} 未注册")
return self.agents[agent_name].run(task)
def collaborate(self, workflow: List[Dict]):
"""按工作流顺序调度多个 Agent"""
results = {}
for step in workflow:
agent_name = step["agent"]
task = step["task"]
# 支持引用前序结果
for key, value in results.items():
task = task.replace(f"{{{{{key}}}}}", value)
print(f"调度 {agent_name} 执行任务...")
results[step["name"]] = self.delegate(agent_name, task)
return results
def main():
api_key = "your-api-key"
orchestrator = Orchestrator(api_key)
注册三个专业 Agent
orchestrator.register_agent(
"architect",
"你是一名资深软件架构师,负责系统设计和模块划分。"
)
orchestrator.register_agent(
"developer",
"你是一名高级开发工程师,负责编写高质量代码。"
)
orchestrator.register_agent(
"reviewer",
"你是一名严格的代码审查员,负责发现代码中的问题和改进点。"
)
定义协作工作流
workflow = [
{
"name": "design",
"agent": "architect",
"task": "为一个用户登录模块设计接口和数据结构,输出 JSON 格式设计文档。"
},
{
"name": "code",
"agent": "developer",
"task": "根据以下设计文档编写 Python 实现代码:\n{{design}}"
},
{
"name": "review",
"agent": "reviewer",
"task": "审查以下代码,指出安全问题、性能问题和改进建议:\n{{code}}"
}
]
results = orchestrator.collaborate(workflow)
print("=== 最终协作结果 ===")
for name, result in results.items():
print(f"\n--- {name} ---")
print(result)
if name == "main":
main()

这个框架的核心设计包括:

  • Agent 类:封装了独立的系统提示词、消息历史和执行方法,每个 Agent 拥有独立的上下文。
  • Orchestrator 类:负责 Agent 注册、任务委派和工作流调度。
  • 结果传递:通过 {{变量名}} 占位符机制,将前序 Agent 的输出注入后续任务,实现信息流转。

6. 多 Agent 协作模式与最佳实践

根据任务特性,多 Agent 协作通常采用以下几种模式:

协作模式 适用场景 优点 缺点
流水线模式 任务有明确先后顺序,如设计到编码到审查 流程清晰,易于追踪 串行执行,整体耗时较长
并行模式 多个独立模块可同时处理 效率高,充分利用资源 结果合并需要额外处理
主从模式 一个主 Agent 协调多个专业 Subagent 职责清晰,上下文隔离好 主 Agent 可能成为瓶颈
辩论模式 需要多角度评估决策,如方案选型 结论更全面,减少偏见 可能陷入争论,收敛慢

在实际项目中,建议遵循以下最佳实践:

  • 明确任务边界:每个 Agent 的职责要清晰,避免重叠导致冲突。
  • 控制上下文长度:传递给 Agent 的中间结果要精简,避免上下文过长影响质量。
  • 设计容错机制:单个 Agent 失败不应导致整个流程中断,应支持重试或降级。
  • 记录执行日志:保留每个 Agent 的输入输出,便于问题排查和结果追溯。
  • 渐进式引入:先从两个 Agent 的简单协作开始,验证效果后再扩展。

7. 实战案例:多 Agent 实现代码库重构

下面给出一个完整的实战案例,演示如何使用多 Agent 协作完成一个小型代码库的重构任务。假设我们有一个遗留的 Python 项目,需要将其拆分为清晰的模块结构:

import anthropic
import os
import json
class RefactorOrchestrator:
def init(self, api_key: str):
self.client = anthropic.Anthropic(api_key=api_key)
self.agents = {}
def register(self, name: str, system: str):
    self.agents[name] = {
        "system": system,
        "history": []
    }
def call(self, name: str, task: str) -> str:
agent = self.agents[name]
agent["history"].append({"role": "user", "content": task})
resp = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=8192,
system=agent["system"],
messages=agent["history"]
)
text = resp.content[0].text
agent["history"].append({"role": "assistant", "content": text})
return text
def run_refactor(self, source_code: str):
# 第一步:分析现有代码
analysis = self.call(
"analyzer",
f"分析以下代码的结构、职责和问题,输出 JSON 格式的分析报告:\n{source_code}"
)
print("=== 分析报告 ===")
print(analysis)
# 第二步:设计目标结构
design = self.call(
    "architect",
    f"根据分析报告设计重构后的模块划分和接口定义:\n{analysis}"
)
print("\n=== 重构设计 ===")
print(design)
第三步:生成重构代码
refactored = self.call(
"developer",
f"根据设计文档生成重构后的完整代码,保持功能不变:\n{design}"
)
print("\n=== 重构代码 ===")
print(refactored)
第四步:质量审查
review = self.call(
"reviewer",
f"审查以下重构代码,检查是否保持原有功能、有无遗漏和错误:\n{refactored}"
)
print("\n=== 审查意见 ===")
print(review)
return {
"analysis": analysis,
"design": design,
"code": refactored,
"review": review
}
def main():
api_key = os.environ.get("ANTHROPIC_API_KEY")
if not api_key:
raise ValueError("请设置 ANTHROPIC_API_KEY 环境变量")
orch = RefactorOrchestrator(api_key)
orch.register("analyzer", "你是一名代码分析专家,擅长识别代码结构、职责和潜在问题。")
orch.register("architect", "你是一名软件架构师,擅长模块划分和接口设计。")
orch.register("developer", "你是一名高级开发工程师,擅长编写清晰、可维护的代码。")
orch.register("reviewer", "你是一名代码审查专家,擅长发现功能遗漏和代码缺陷。")
示例遗留代码
legacy_code = '''
import json
def process(data):
result = []
for item in data:
if item["type"] == "user":
result.append({"id": item["id"], "name": item["name"]})
elif item["type"] == "order":
result.append({"id": item["id"], "total": item["total"]})
return json.dumps(result)
def save(data, filename):
with open(filename, "w") as f:
f.write(data)
def load(filename):
with open(filename, "r") as f:
return json.loads(f.read())
'''
orch.run_refactor(legacy_code)
if name == "main":
main()

这个案例展示了多 Agent 协作的完整链路:分析、设计、编码、审查四个环节由不同专业 Agent 接力完成,每个 Agent 专注于自己的职责,最终输出经过审查的重构代码。这种模式特别适合代码重构、功能开发、文档生成等需要多角色协作的任务。

8. 总结

Claude Code 多 Agent 实现的核心在于任务分解和协作编排。通过内置 Subagent、CLI 脚本编排或 API 框架构建,开发者可以根据任务复杂度选择合适的方式。多 Agent 协作能够显著提升复杂任务的完成质量和效率,但也需要精心设计任务边界、上下文传递和容错机制。建议从简单场景入手,逐步积累经验,最终构建适合自己团队的多 Agent 工作流。

更多推荐