Claude Agent SDK 子代理并行开发

简介

子代理（Sub-agent）并行，就是启动多个独立 Claude Agent 实例，同时执行不同子任务，主代理负责任务拆分、分发、收集结果、合并输出。

区分概念

串行子代理：逐个执行任务，适合有依赖流程；速度慢

并行子代理：多个任务无依赖，同时发起请求，提升处理效率

和 Claude Code 子代理：Claude Code 内置子代理；SDK 并行是代码层面自主编排，自由度更高

适用场景

批量审查多个源码文件
同时解析多篇文档，分别提取摘要
多维度评估方案（安全性、性能、可读性）
爬虫多页面并行内容分析

限制说明

Anthropic API 存在速率限制（RPM），并发不可无限拉高，需要做并发池控制。

前置准备

Python3.10+ / Node.js 18+
SDK 安装

# Python
pip install anthropic aiohttp
# TS
npm install @anthropic-ai/sdk

API Key 使用环境变量，禁止硬编码
了解：异步编程基础，SDK messages 调用

核心工作流程

主代理：接收原始需求，拆解成若干独立子任务列表
任务分发：通过异步并发，将子任务交给多个子代理执行
子代理：独立执行任务，返回各自结果
结果聚合：收集全部子代理输出，交给主代理汇总、校验、生成最终报告
异常处理：捕获单个子代理失败，不中断全部任务

flowchart LR
    A[用户请求] --> B[主Agent：任务拆分]
    B --> C[子任务1 → 子代理1]
    B --> D[子任务2 → 子代理2]
    B --> E[子任务3 → 子代理3]
    C & D & E --> F[收集所有子任务结果]
    F --> G[主Agent汇总输出最终结果]

实战1：Python 异步并行子代理示例

场景：并行对3个代码片段做安全审查

import os
import asyncio
from anthropic import Anthropic, AsyncAnthropicclient = AsyncAnthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
子代理执行函数
async def sub_agent_review(code_snippet: str) -> str:
    resp = await client.messages.create(
        model="claude-3-5-sonnet-latest",
        max_tokens=1024,
        temperature=0.2,
        system="你是代码安全审计子代理，只输出漏洞清单与风险等级，简洁输出",
        messages=[
            {"role": "user", "content": f"审计下面代码:\n{code_snippet}"}
        ]
    )
    return resp.content[0].text
主代理：拆分任务 + 并行执行
async def main():
    # 待审查代码列表，相互独立，可以并行
    task_list = [
        "user_input = input(); os.system(f'ping {user_input}')",
        "sql = 'SELECT * FROM users where name = ' + user_input",
        "eval(user_input)"
    ]
    # 构建并发任务数组
    tasks = [sub_agent_review(code) for code in task_list]
    # 并发等待全部执行完成
    sub_results = await asyncio.gather(*tasks, return_exceptions=True)
# 过滤异常结果
valid_results = []
for idx, res in enumerate(sub_results):
    if isinstance(res, Exception):
        valid_results.append(f"【任务{idx+1}】执行失败：{str(res)}")
    else:
        valid_results.append(f"【任务{idx+1}结果】\n{res}")

# 主代理汇总全部子代理结果
merge_prompt = "合并下面所有代码审计结果，生成一份总报告，汇总全部风险点：\n" + "\n".join(valid_results)
final_resp = await client.messages.create(
    model="claude-3-5-sonnet-latest",
    max_tokens=1500,
    system="你是报告汇总主代理，整理子代理输出，去重，按风险等级排序",
    messages=[{"role": "user", "content": merge_prompt}]
)
print(final_resp.content[0].text)if name == "main":
    asyncio.run(main())

实战2：TypeScript 并行子代理示例

import Anthropic from "@anthropic-ai/sdk";
const anthropic = new Anthropic({
  apiKey: process.env.ANTHROPIC_API_KEY,
});// 子代理函数
async function subAgentReview(code: string): Promise<string> {
  const res = await anthropic.messages.create({
    model: "claude-3-5-sonnet-latest",
    max_tokens: 1024,
    temperature: 0.2,
    system: "代码安全审计子代理，输出漏洞清单和风险等级",
    messages: [{ role: "user", content: 审计代码：${code} }],
  });
  return res.content[0].text;
}
async function main() {
  const taskList = [
    "user_input = input(); os.system(f'ping {user_input}')",
    "sql = 'SELECT * FROM users where name = ' + user_input",
    "eval(user_input)"
  ];
  // 并发执行所有子任务
  const tasks = taskList.map(item => subAgentReview(item));
  const subResults = await Promise.allSettled(tasks);
  const output: string[] = [];
  for (let i = 0; i < subResults.length; i++) {
    const r = subResults[i];
    if (r.status === "fulfilled") {
      output.push(【任务${i+1}结果】${r.value});
    } else {
      output.push(【任务${i+1}失败】${r.reason});
    }
  }
  // 主代理汇总
  const final = await anthropic.messages.create({
    model: "claude-3-5-sonnet-latest",
    max_tokens: 1500,
    system: "汇总所有子代理审计结果，去重并按风险等级排序",
    messages: [{ role: "user", content: output.join("\n") }],
  });
  console.log(final.content[0].text);
}
main();

并发控制（重要）

直接一次性开启大量并发，极易触发 API 限流 429。需要限制最大并发数量。

Python 简易并发池示例（使用 asyncio.Semaphore）

# 限制最多同时2个并发
sem = asyncio.Semaphore(2)
async def sub_agent_review(code_snippet: str):
    async with sem:
        resp = await client.messages.create(...)

TS 可以使用 p-limit 包做并发限制：

npm install p-limit

子代理设计规范

任务必须独立：并行子任务之间不能存在数据依赖；有依赖必须串行。
子代理职责单一：每个子代理只做一件事，system prompt 精简，减少 token。
结果结构化：要求子代理输出固定格式（markdown / JSON），方便主代理合并。
超时设置：每个子代理请求增加超时时间，防止单个任务卡死整体流程。
隔离上下文：每个子代理会话完全独立，消息历史互不干扰。

错误与重试策略

使用 asyncio.gather(return_exceptions=True) / Promise.allSettled，单个失败不整体崩溃
捕获 429 限流错误，增加指数退避重试
对子代理结果做校验：主代理判断子代理输出是否为空、格式错误，必要时重新派发任务

并行子代理 + 自定义工具组合

子代理同样可以传入 tools 参数，子代理内部也可以调用自定义工具。
典型场景：每个子代理读取不同文件，并行分析，再汇总。

注意：子代理的工具调用也是独立的，每个子代理各自完成 tool_use → tool_result 循环。

并行子代理 vs Claude Code 子代理对比

项目	SDK 并行子代理	Claude Code 内置子代理
运行环境	程序代码内编排	本地 IDE / CLI 会话内
并发控制	开发者代码自主控制	Claude Code 内部调度
上下文	完全隔离	可共享项目上下文
适用场景	后端服务自动化、批量任务	本地项目交互开发
工具能力	接入 SDK Tools	原生支持 MCP

常见问题

Q：并行越多速度越快？

A：不是。受 API 速率限制，并发超过阈值会触发429，大量重试反而变慢，建议并发控制在 2~5 起步。

Q：子代理之间能不能共享记忆？

A：默认不共享。如需共享，需要手动把公共信息传入每个子代理的 system prompt。

Q：主代理会不会丢失大量细节？

A：子代理返回结果尽量摘要，不要返回原始长文本，降低汇总阶段 token 开销。

小结

SDK 子代理并行开发核心：异步任务分发 + 并发限制 + 独立子任务执行 + 结果聚合。适合批量、无依赖的任务场景。一定要做好限流、超时、异常捕获，避免 API 账单暴增与请求报错。

简介