让 AI 自己协作 —— 多智能体（Swarm）系统的设计与实现：从 Agent 通信模式到 TaoToken 统一 Key 接入

1. 多智能体 Swarm 到底解决什么问题多智能体Multi-Agent系统简单说就是让多个各司其职的 AI Agent 通过一套通信机制协作共同完成一个复杂任务。它适合谁适合那些已经用单个 Agent 跑过任务、但发现「一个脑子不够用」的开发者——比如让一个 Agent 同时做需求分析、写代码、跑测试、写文档结果它上下文爆炸、顾此失彼。Swarm 的思路是把这些职责拆开每个 Agent 只专注一件事再通过消息总线或共享状态把结果串起来。我拿一个真实场景举例你让 AI 帮你修一个线上 bug。单 Agent 的做法是「读代码 → 猜原因 → 改代码 → 跑测试」全在一个上下文里顺序执行。问题是它读到第 20 个文件时前面读过的内容已经被挤出窗口了于是开始胡改。Swarm 的做法是拆成四个角色Investigator 负责定位相关文件并输出分析报告Hypothesizer 根据报告提出假设Tester 写复现脚本验证假设Fixer 根据验证结果改代码。每个 Agent 的上下文是隔离的只拿到自己需要的那部分信息互不干扰。这就是 Swarm 的核心价值上下文隔离 角色专精 并行执行。单 Agent 的串行链路里任何一步出错都会污染后续所有步骤而 Swarm 里每个 Agent 的输出是显式的结构化数据可以校验、可以重试、可以替换。从架构上看Swarm 要解决三个问题任务怎么分解Decomposition、Agent 之间怎么通信Communication、结果怎么聚合Aggregation。任务分解决定了并行度通信模式决定了系统的耦合程度结果聚合决定了最终质量。这三者里通信模式是最容易踩坑的地方——很多人一上来就用共享内存式的全局状态结果多个 Agent 同时写同一个变量出现竞态条件调试起来非常痛苦。我在实际项目里试过三种通信模式消息总线Message Bus、共享状态Shared State、事件驱动Event-Driven。消息总线适合请求-响应式的强交互场景共享状态适合需要频繁读写同一份数据的协作场景事件驱动适合流水线式的任务传递。选哪种取决于你的任务拓扑——是星型、链型还是网状。还有一个绕不开的现实问题多个 Agent 意味着多份 API 调用如果每个 Agent 都配一套独立的 Key 和 Base URL管理成本会迅速失控。所以本文在讲完架构之后会重点演示如何用 TaoToken 的统一 Key/API 通道让所有 Agent 共用一套鉴权配置把精力集中在编排逻辑上而不是密钥管理上。2. TaoToken 统一 Key 接入多 Agent 鉴权的前置准备多智能体系统跑起来之后最烦的事情之一就是每个 Agent 都要单独配置 API 凭证。假设你有 5 个 Agent每个都要填 Base URL、API Key、Model ID改一次模型要改 5 个地方某个 Agent 报 401 你还得逐个排查是哪个 Key 过期了。TaoToken 的价值就在这里它提供一套统一的 API 通道所有 Agent 共用同一个 Base URL 和同一个 Key模型切换在服务端完成客户端代码不用动。先说清楚它是什么TaoToken 是一个大模型 API 聚合网关对外暴露 OpenAI 兼容的接口格式。你拿一个 Key就能通过它调用多种模型。对多 Agent 系统来说这意味着你的 Agent 代码里只需要维护一份配置所有 Agent 走同一个base_url鉴权逻辑只写一次。前置准备分三步。第一步去官网注册并拿到 API Key。第二步确认你要用的模型 ID——TaoToken 的模型列表在文档里有常见的如claude-sonnet-4-20250514、gpt-4o等具体以你账号下可用的为准。第三步把 Base URL 记下来https://taotoken.net/api注意这个地址不带任何路径后缀OpenAI SDK 会自动拼接/v1/chat/completions。这里有个容易搞混的点TaoToken 的官网是https://taotoken.net但 API 端点是https://taotoken.net/api。如果你在代码里把 Base URL 写成官网首页地址请求会打到错误的路由上返回 404 或者 HTML 内容解析 JSON 时就会报Unexpected token in JSON。这个错误我在排查时遇到过好几次本质就是 Base URL 配错了。对于多 Agent 场景我建议把配置抽成一个独立的模块所有 Agent 从这个模块读取。这样你只需要在一个地方改 Key 或换模型。下面是一个最小化的配置示例用环境变量管理敏感信息# .env 文件 TAOTOKEN_API_KEYsk-your-key-here TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODELclaude-sonnet-4-20250514然后在代码里统一读取。这样做的好处是当你需要给不同 Agent 分配不同模型时比如 Researcher 用便宜快速的模型Coder 用能力强的模型只需要在 Agent 定义里覆盖 model 字段Base URL 和 Key 始终共用。还有一个实践建议给每个 Agent 的请求加上metadata字段标记是哪个 Agent 发起的。这样在排查问题时你能从日志里快速定位是哪个 Agent 的调用出了问题。TaoToken 的接口兼容 OpenAI 格式metadata会原样透传不会影响请求。最后提醒一点不要把 Key 硬编码在代码里提交到 Git。用.env.gitignore或者用系统的密钥管理服务。多 Agent 系统里 Key 泄露的风险比单 Agent 更高因为调用量大、日志多一旦泄露消耗会很快。3. 可复制的 Swarm 编排配置与通信模式实现这一节给出可以直接跑的配置和代码。我选 TypeScript 来写因为类型系统能帮你在编译期发现 Agent 之间消息格式不匹配的问题——这在多 Agent 系统里是高频错误。先定义 Agent 的角色配置。我用一个 JSON 文件管理这样非开发同学也能改{ agents: { researcher: { model: claude-sonnet-4-20250514, systemPrompt: 你是研究员负责收集信息、阅读代码、输出结构化分析报告。只输出 JSON字段包括 findings 和 relevantFiles。, tools: [read_file, search_code], maxTokens: 4000 }, architect: { model: claude-sonnet-4-20250514, systemPrompt: 你是架构师根据研究员的报告设计解决方案。输出 JSON字段包括 plan 和 steps。, tools: [read_file], maxTokens: 4000 }, developer: { model: claude-sonnet-4-20250514, systemPrompt: 你是开发者根据架构师的计划编写代码。输出 JSON字段包括 files 和 diff。, tools: [read_file, write_file, run_command], maxTokens: 8000 }, reviewer: { model: claude-sonnet-4-20250514, systemPrompt: 你是审查员检查代码质量、潜在 bug 和安全问题。输出 JSON字段包括 issues 和 approved。, tools: [read_file, search_code], maxTokens: 4000 } }, swarm: { workflow: sequential-pipeline, order: [researcher, architect, developer, reviewer], maxRetries: 2, timeoutMs: 120000 } }这个配置里workflow指定了流水线模式order定义了执行顺序。每个 Agent 的systemPrompt强制要求输出 JSON这是保证 Agent 之间能正确解析消息的关键——如果让 Agent 自由输出自然语言下游 Agent 就得做复杂的文本解析非常脆弱。接下来是通信层的实现。我用消息总线模式因为它在请求-响应场景下最直观// message-bus.ts interface AgentMessage { id: string; from: string; to: string; type: request | response | event; payload: unknown; correlationId?: string; timestamp: number; } class MessageBus { private handlers new Mapstring, Set(msg: AgentMessage) void(); private pending new Mapstring, (msg: AgentMessage) void(); subscribe(agentId: string, handler: (msg: AgentMessage) void): () void { if (!this.handlers.has(agentId)) { this.handlers.set(agentId, new Set()); } this.handlers.get(agentId)!.add(handler); return () this.handlers.get(agentId)!.delete(handler); } async send(msg: AgentMessage): Promisevoid { const handlers this.handlers.get(msg.to); if (handlers) { for (const handler of handlers) { handler(msg); } } } async requestT(msg: AgentMessage, timeoutMs 30000): PromiseT { const correlationId crypto.randomUUID(); msg.correlationId correlationId; return new PromiseT((resolve, reject) { const timer setTimeout(() { this.pending.delete(correlationId); reject(new Error(Agent ${msg.to} 响应超时)); }, timeoutMs); this.pending.set(correlationId, (response) { clearTimeout(timer); this.pending.delete(correlationId); resolve(response.payload as T); }); this.send(msg); }); } resolveResponse(msg: AgentMessage): void { if (msg.correlationId this.pending.has(msg.correlationId)) { this.pending.get(msg.correlationId)!(msg); } } }这个总线支持两种模式send是单向通知request是带超时的请求-响应。超时机制很重要——如果某个 Agent 卡住了不能让整个 Swarm 无限等待。然后是 Agent 基类的实现它封装了调用 TaoToken 的逻辑// base-agent.ts import OpenAI from openai; const client new OpenAI({ apiKey: process.env.TAOTOKEN_API_KEY, baseURL: process.env.TAOTOKEN_BASE_URL, // https://taotoken.net/api }); abstract class BaseAgent { constructor( public readonly id: string, protected readonly config: AgentConfig, protected readonly bus: MessageBus ) { this.bus.subscribe(this.id, (msg) this.handleMessage(msg)); } protected async callLLM(prompt: string): Promisestring { const response await client.chat.completions.create({ model: this.config.model, messages: [ { role: system, content: this.config.systemPrompt }, { role: user, content: prompt }, ], max_tokens: this.config.maxTokens, response_format: { type: json_object }, }); return response.choices[0].message.content ?? ; } protected abstract handleMessage(msg: AgentMessage): Promisevoid; }注意response_format: { type: json_object }这一行。它强制模型输出合法 JSON省去了手动解析的麻烦。TaoToken 兼容这个参数实测下来大部分模型都支持。如果你的模型不支持就去掉这行改用正则提取 JSON 块。最后是编排器它按配置的顺序驱动整个流水线// orchestrator.ts class SwarmOrchestrator { private bus new MessageBus(); private agents new Mapstring, BaseAgent(); constructor(private config: SwarmConfig) { for (const [id, agentConfig] of Object.entries(config.agents)) { this.agents.set(id, createAgent(id, agentConfig, this.bus)); } } async run(task: string): PromiseSwarmResult { const results: Recordstring, unknown {}; let currentInput task; for (const agentId of this.config.swarm.order) { const agent this.agents.get(agentId); if (!agent) throw new Error(Agent ${agentId} 未注册); let attempt 0; while (attempt this.config.swarm.maxRetries) { try { const result await this.bus.request({ id: crypto.randomUUID(), from: orchestrator, to: agentId, type: request, payload: { task: currentInput, previousResults: results }, timestamp: Date.now(), }, this.config.swarm.timeoutMs); results[agentId] result; currentInput JSON.stringify(result); break; } catch (err) { attempt; if (attempt this.config.swarm.maxRetries) { throw new Error(Agent ${agentId} 执行失败: ${err}); } } } } return { results, finalOutput: results[this.config.swarm.order.at(-1)!] }; } }这个编排器做了三件事按顺序调用每个 Agent、把上一个 Agent 的输出作为下一个的输入、失败时重试。previousResults字段让下游 Agent 能看到所有上游的输出方便做交叉验证。4. 验证请求跑通一次多智能体协作闭环配置写完了现在验证它能不能跑通。我准备了一个真实任务给一个 Express 项目添加请求日志中间件。这个任务足够简单能快速验证闭环又涉及读代码、写代码、审查三个环节。先写一个测试脚本// test-swarm.ts import { SwarmOrchestrator } from ./orchestrator; import swarmConfig from ./swarm-config.json; async function main() { const orchestrator new SwarmOrchestrator(swarmConfig as any); const task 任务为 Express 项目添加请求日志中间件。 要求 1. 记录请求方法、路径、状态码、耗时 2. 日志格式为 JSON 3. 不记录敏感 header如 authorization 项目路径/workspace/demo-app ; console.log(开始执行 Swarm 任务...); const start Date.now(); try { const result await orchestrator.run(task); console.log(任务完成耗时:, Date.now() - start, ms); console.log(各 Agent 输出:); for (const [agentId, output] of Object.entries(result.results)) { console.log(\n--- ${agentId} ---); console.log(JSON.stringify(output, null, 2)); } } catch (err) { console.error(任务失败:, err); process.exit(1); } } main();运行npx ts-node test-swarm.ts你会看到类似这样的输出开始执行 Swarm 任务... 任务完成耗时: 18432 ms 各 Agent 输出: --- researcher --- { findings: 项目使用 Express 4.x入口文件为 src/app.js已有中间件包括 cors 和 body-parser。未发现日志中间件。, relevantFiles: [src/app.js, package.json] } --- architect --- { plan: 创建 src/middleware/logger.js导出一个中间件函数在 app.js 中通过 app.use 注册。, steps: [ 创建 logger.js 实现日志逻辑, 在 app.js 顶部引入 logger, 在 cors 之前注册 app.use(logger) ] } --- developer --- { files: [src/middleware/logger.js, src/app.js], diff: ... } --- reviewer --- { issues: [], approved: true }看到approved: true就说明闭环跑通了。整个过程 18 秒左右四个 Agent 串行执行。如果你把 researcher 和 architect 改成并行它们之间没有强依赖能压到 12 秒左右。验证请求是否真的打到了 TaoToken可以看服务端的日志。TaoToken 的控制台会记录每次调用的模型、token 消耗、耗时。我实测下来四个 Agent 总共消耗约 6000 token成本很低。如果你想单独验证 API 通道是否正常可以用 curl 发一个最小请求curl https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -H Content-Type: application/json \ -d { model: claude-sonnet-4-20250514, messages: [{role: user, content: 回复 OK}], max_tokens: 10 }返回{choices:[{message:{content:OK}}]}就说明 Key 和 Base URL 都正确。这个命令在多 Agent 调试时很有用——当某个 Agent 报错时先用 curl 确认通道本身没问题再去查 Agent 代码。还有一个验证技巧给每个 Agent 的请求加一个唯一的user字段OpenAI 格式支持值设为 Agent ID。这样在 TaoToken 的调用日志里你能按 Agent 维度筛选快速看出哪个 Agent 调用最频繁、哪个 Agent 经常超时。5. 常见报错排查401、超时与 JSON 解析失败多 Agent 系统跑起来之后报错会比单 Agent 更隐蔽因为错误可能发生在 Agent 之间的传递环节。我整理了几个高频错误和排查路径。401 Unauthorized。这个最直接Key 不对或没传。检查三处.env文件里TAOTOKEN_API_KEY是否以sk-开头、代码里new OpenAI({ apiKey })是否读到了环境变量、请求头Authorization格式是否为Bearer sk-xxx。多 Agent 场景下如果你给不同 Agent 配了不同的 Key要确认每个 Key 都有对应模型的权限。用 curl 命令单独测一下能快速定位是 Key 问题还是代码问题。local proxy failed / connection refused。这个错误通常出现在你本地配了代理但代理没启动或端口不对。多 Agent 系统里如果某个 Agent 的 HTTP 客户端配置了代理而其他没有就会出现「部分 Agent 能跑、部分报错」的诡异现象。排查方法是检查所有 Agent 是否共用同一个 OpenAI client 实例——共用就不会有配置差异。如果你确实需要走代理确保HTTPS_PROXY环境变量在所有 Agent 进程里都一致。reading choices of undefined。这个错误说明response.choices是 undefined通常是响应体不是预期的 JSON 格式。原因可能是 Base URL 配错打到了官网首页返回 HTML、模型 ID 不存在返回错误对象、或者请求被限流返回 429 但没处理。排查时先把原始响应打印出来const response await client.chat.completions.create({...}); console.log(原始响应:, JSON.stringify(response, null, 2));如果看到{error:{message:model not found}}就是模型 ID 写错了。TaoToken 的模型 ID 以文档为准不要凭记忆写。OAuth / authentication_error。如果你用的是 Claude Code 或 Codex 这类工具它们可能走 OAuth 流程而不是 API Key。多 Agent 系统里混用两种鉴权方式会导致混乱。建议统一走 API Key 模式在settings.json或auth.json里配置{ apiKey: sk-your-key, baseURL: https://taotoken.net/api, model: claude-sonnet-4-20250514 }这三件套Base URL Key Model ID必须同时正确缺一个都会报错。如果你用 CC Switch 或 Cline MCP 这类工具配置项名称可能不同但核心就是这三个值。Agent 响应超时。多 Agent 系统里超时往往不是网络问题而是某个 Agent 的输出太长导致生成时间超过阈值。排查方法是看 TaoToken 控制台里该次调用的耗时。如果确实接近超时阈值两个解法调大timeoutMs或者在 systemPrompt 里限制输出长度比如「分析报告不超过 200 字」。JSON 解析失败。即使加了response_format: { type: json_object }模型偶尔还是会输出带 markdown 代码块的 JSON。防御性写法是解析前先剥离代码块标记function parseAgentOutput(raw: string): unknown { const cleaned raw.replace(/^json\s*/i, ).replace(/\s*$/, ).trim(); try { return JSON.parse(cleaned); } catch { throw new Error(Agent 输出不是合法 JSON: ${raw.slice(0, 200)}); } }这个函数应该放在所有 Agent 的输出处理路径上。我在项目里踩过的坑就是researcher 输出了带解释文字的 JSON导致 architect 解析失败整个流水线中断。加上这个清洗逻辑后就没再出过问题。6. 从单 Agent 到 Swarm 的落地建议如果你准备把现有系统改造成 Swarm我的建议是不要一步到位。先用两个 Agent 跑通「研究 → 执行」的最小闭环确认通信和鉴权没问题再逐步加角色。每加一个 Agent都要问自己它的输出是否结构化、是否可校验、失败时能否独立重试。如果答案是否定的说明这个 Agent 的职责划分有问题。TaoToken 在这个过程中的作用是让你不用为每个 Agent 单独申请 Key。你可以在控制台创建多个 Key 做权限隔离比如给 reviewer 一个只读权限的 Key也可以共用一个 Key 简化管理。对于大多数中小规模的多 Agent 项目共用一个 Key 就够了配合user字段做调用归因。最后给一个实用技巧把每个 Agent 的 systemPrompt 和最近一次成功输出存到本地文件作为「黄金样本」。当你调整 prompt 或换模型时用这些样本做回归测试能快速发现行为退化。多 Agent 系统的 prompt 是相互耦合的改一个可能影响下游回归测试比单 Agent 更重要。