在 Agent SDK 中实现子代理——并行处理多个独立任务。
什么是子代理
子代理是从主代理中派生出的独立代理实例。每个子代理：
- 有自己的上下文和消息历史
- 可以并行运行
- 完成后返回结果给主代理
- 不共享主代理的上下文窗口
使用场景
- 并行分析多个文件
- 同时处理不同的子任务
- 独立审查不同模块
- 批量处理重复任务
实现方式
import concurrent.futures
from anthropic import Anthropic
client = Anthropic()
def sub_agent(task):
"""运行一个子代理处理特定任务。"""
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
system=f"你是子代理。你的任务是：{task}",
messages=[
{"role": "user", "content": "请完成你的任务并返回结果。"}
]
)
return response.content[0].text
def parallel_tasks(tasks):
"""并行运行多个子代理。"""
with concurrent.futures.ThreadPoolExecutor() as executor:
futures = {
executor.submit(sub_agent, task): task
for task in tasks
}
results = {}
for future in concurrent.futures.as_completed(futures):
task = futures[future]
results[task] = future.result()
return results
# 使用
tasks = [
"审查 auth.py 的安全问题",
"检查 utils.py 的性能问题",
"分析 models.py 的架构设计"
]
results = parallel_tasks(tasks)
for task, result in results.items():
print(f"任务：{task}
结果：{result}
")注意事项
- 每个子代理消耗独立的 API 调用
- 注意 API 速率限制
- 子代理之间无法直接通信
- 主代理负责整合结果