1. 多 Agent 协作系统到底解决什么问题什么时候不该上多 Agent 协作系统简单说就是让多个各司其职的模型实例每个实例带独立的系统提示词和工具权限通过一套调度逻辑协同完成一个任务。它能做的事很具体把「调研→写作→审校」这种有明显阶段差异的流程拆开让每个阶段用最合适的提示词把装不下的大上下文拆给专门的 Agent 去压缩筛选让生成和审查由两个不同角色承担避免自己给自己打分。适合谁适合已经用单 Agent 跑过一轮、发现提示词越写越长、职责越来越混、输出质量开始飘的开发者。但先说清楚什么时候不该上。如果你的任务能用一次 prompt 加几个工具调用搞定那就别折腾多 Agent 只会增加延迟和成本。真正需要拆分的信号通常是这几个任务本身有清晰阶段每个阶段需要的能力和提示词差异很大单次上下文装不下需要有人专门负责筛选信息、压缩中间结果需要不同的「人格」或权限比如一个负责生成、一个负责挑刺让它们互相制衡比让同一个模型既当运动员又当裁判更可靠。最后这一点我体会最深。让同一个 Agent 自己写完自己审它几乎总是给自己打高分。把审校单独拆出来、用不同的系统提示词质量立刻就上去了。所以本文的目标很明确先讲清楚拓扑怎么选再给出一套不依赖重型框架、用标准库就能跑起来的编排器骨架最后用统一的 API 通道把模型调用接上让你从零跑通一条多 Agent 协作链路。2. 四种协作拓扑对比与 TaoToken 统一 Key 前置准备多 Agent 系统设计80% 的功夫在拓扑选择上。选错了结构后面写再多代码都是在补窟窿。常见的就四种我按生产可用性从高到低排拓扑结构适合场景主要风险编排器–执行器中心 Agent 规划调度子 Agent 执行需要动态决策的生产任务中心节点逻辑要写扎实流水线固定链路上一步输出是下一步输入阶段明确、无需动态决策不灵活中间出错难回退对等协作多 Agent 共享会话自由发言高度开放的头脑风暴容易陷入互相恭维死循环层级编排器下再挂子编排器超大型任务调试成本指数级上升选型经验任务流程固定就用流水线需要根据中间结果动态决策就用编排器–执行器只有当任务高度开放、确实需要「头脑风暴」时才考虑对等协作。层级能两层解决就别上三层。Agent 之间怎么「说话」也有讲究。我见过不少实现把上一个 Agent 的完整输出原封不动塞给下一个结果上下文越滚越大到链路末端早就爆了。更干净的做法是黑板模式所有 Agent 不直接对话而是读写一块共享状态每个 Agent 只取自己需要的字段只写自己负责的产出。好处是上下文可控、产出有结构、调试时把黑板打印出来整个协作过程一目了然。在动手写编排器之前先把模型调用通道准备好。多 Agent 意味着一次任务会发起十几次甚至几十次模型请求如果每个 Agent 各自配一套 Key、各自处理鉴权和重试编排器代码会被这些杂事淹没。我的做法是统一走一个兼容 OpenAI 协议的 API 通道所有 Agent 共用同一个 Base URL 和 Key模型 ID 按角色需要切换。这里我用 TaoToken 来做这层统一接入官网入口是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 地址是 https://taotoken.net/api 。你需要准备三样东西Base URL、API Key、Model ID。Base URL 填https://taotoken.net/apiKey 在控制台的 API Keys 页面创建Model ID 按你实际要用的模型填。这三件套在后面的配置里会反复出现先记牢。控制台地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API Keys 管理页是 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。如果你还没决定用哪个模型可以先去模型对话页 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 试几个确认输出风格符合你的角色设定再写进配置。3. 可复制的编排器骨架配置Agent 注册、消息路由、失败重试这一节给你一套能直接跑的骨架。我刻意不依赖 LangGraph、AutoGen 之类的框架就用标准库目的是把协作的骨架暴露出来——很多人用了框架却说不清里面到底发生了什么。场景是输入一个主题产出一篇结构清晰的短文拆成规划→检索→写作→审校四个角色审校不通过就打回重写带最大轮次保护。先写配置文件。把模型接入信息抽成独立的 JSON编排器和 Agent 都从这里读换模型时只改一处{ base_url: https://taotoken.net/api, api_key: sk-你的Key, models: { planner: gpt-4o-mini, searcher: gpt-4o-mini, writer: gpt-4o-mini, reviewer: gpt-4o-mini }, max_revisions: 2, request_timeout: 60, max_retries: 3 }注意base_url后面不要带/v1OpenAI SDK 会自己拼路径。Key 从环境变量读更安全配置文件里可以留空代码里优先读环境变量。下面是编排器主体import os import json import time from dataclasses import dataclass, field CONFIG json.load(open(orchestrator.json, encodingutf-8)) API_KEY os.getenv(TAOTOKEN_API_KEY) or CONFIG[api_key] def call_llm(system: str, user: str, model: str) - str: from openai import OpenAI client OpenAI(base_urlCONFIG[base_url], api_keyAPI_KEY, timeoutCONFIG[request_timeout]) last_err None for attempt in range(CONFIG[max_retries]): try: resp client.chat.completions.create( modelmodel, messages[ {role: system, content: system}, {role: user, content: user}, ], temperature0.3, ) return resp.choices[0].message.content.strip() except Exception as e: last_err e time.sleep(1.5 ** attempt) raise RuntimeError(f模型调用失败: {last_err}) dataclass class Blackboard: task: str artifacts: dict field(default_factorydict) def context(self, *keys) - str: picked {k: self.artifacts.get(k, ) for k in keys} return json.dumps(picked, ensure_asciiFalse, indent2) dataclass class Agent: name: str role_prompt: str model_key: str def run(self, instruction: str) - str: return call_llm(self.role_prompt, instruction, CONFIG[models][self.model_key]) class Orchestrator: def __init__(self): self.max_revisions CONFIG[max_revisions] self.planner Agent(规划, 你是规划 Agent。把主题拆成 3 个写作要点逐行输出。, planner) self.searcher Agent(检索, 你是检索 Agent。针对要点给出关键事实简洁罗列。, searcher) self.writer Agent(写作, 你是写作 Agent。根据要点和事实写一篇 300 字短文。, writer) self.reviewer Agent(审校, 你是审校 Agent。判断短文是否合格只返回 JSON{\pass\: bool, \comment\: str}, reviewer) def run(self, topic: str) - Blackboard: board Blackboard(tasktopic) board.artifacts[outline] self.planner.run(f主题:{topic}) board.artifacts[facts] self.searcher.run(f要点:\n{board.context(outline)}) feedback for attempt in range(self.max_revisions 1): draft self.writer.run( f主题:{topic}\n素材:\n{board.context(outline, facts)}\n f上一轮审校意见(若有):{feedback}) board.artifacts[draft] draft review json.loads(self.reviewer.run(f短文:\n{draft})) if review[pass]: board.artifacts[status] f第 {attempt1} 轮通过 break feedback review[comment] board.artifacts[status] f第 {attempt1} 轮被打回:{feedback} else: board.artifacts[status] 达到最大重写次数强制交付 return board if __name__ __main__: board Orchestrator().run(多 Agent 系统的工程价值) print(json.dumps(board.artifacts, ensure_asciiFalse, indent2))这套骨架里Agent 注册就是构造Agent对象时传入角色提示词和模型键消息路由靠黑板每个 Agent 只读自己需要的字段失败重试在call_llm里用指数退避包了三层。跑一遍你会看到四个角色各司其职黑板里清晰记录了每一步产出审校不通过会带着意见打回去重写而且永远不会无限循环。4. 本地验证请求与成功结果解读配置写好后先别急着跑完整链路分三步验证出问题好定位。第一步单独验证 API 通道通不通。用 curl 发一个最小请求curl https://taotoken.net/api/chat/completions \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -H Content-Type: application/json \ -d { model: gpt-4o-mini, messages: [{role: user, content: 只回复两个字通了}] }返回体里choices[0].message.content是「通了」说明 Base URL、Key、Model ID 三件套都对。如果这里就报错先看第 5 节的排查表别往下走。第二步跑单 Agent。把编排器里planner.run单独拎出来执行确认角色提示词能产出预期格式。规划 Agent 应该逐行输出三个要点如果它输出一大段散文说明提示词约束不够加一句「只输出要点不要解释」。第三步跑完整链路。执行python orchestrator.py正常输出类似{ outline: 1. 多 Agent 降低单点提示词复杂度\n2. 黑板模式控制上下文膨胀\n3. 终止条件保障生产可用, facts: 多 Agent 单次任务请求数可达十几次反馈闭环需硬上限结构化状态便于 trace, draft: 多 Agent 协作系统的工程价值体现在……, status: 第 1 轮通过 }看到status是「第 1 轮通过」说明规划、检索、写作、审校四个环节全部跑通审校 Agent 返回了合法 JSON 且pass为 true。如果status是「第 2 轮被打回」说明审校给了意见、写作 Agent 带着意见重写了一次这也是正常路径只要最终能收敛就行。如果连续打回到「达到最大重写次数」那要去看draft和审校意见通常是写作 Agent 的提示词和审校标准对不上。验证通过后你可以把models里不同角色换成不同模型比如规划用推理强的、检索用便宜的观察成本和质量的平衡点。这一步的调整不需要改编排器代码只改 JSON 配置。5. 本篇常见错误排查401、local proxy failed、reading choices、OAuth多 Agent 链路比单 Agent 更容易出问题因为错误会在 Agent 之间传递和放大。下面是我实际遇到过的几类报错和对应处理。401 Unauthorized。最常见Key 没读到或写错了。检查三处环境变量TAOTOKEN_API_KEY是否真的导出echo $TAOTOKEN_API_KEY看有没有值配置文件里的 Key 有没有多余空格或换行Key 是否已在控制台启用。如果用的是sk-开头的 Key注意别把前后引号也复制进去。local proxy failed / connection refused。这类报错通常是 Base URL 写错或网络层拦截。确认base_url是https://taotoken.net/api不要带/v1也不要带末尾斜杠。如果你本地有环境变量HTTP_PROXY、HTTPS_PROXY指向了不可用的地址SDK 会尝试走它然后失败临时unset掉再试。reading choices / KeyError choices。返回体里没有choices字段说明请求根本没到模型层或者返回的是错误结构。先打印完整响应体看error字段写了什么。常见原因是 Model ID 拼错比如把gpt-4o-mini写成gpt-4o_mini服务端会返回错误对象而不是正常补全结果。另一个原因是审校 Agent 返回的 JSON 被json.loads解析失败这时要检查审校提示词是否严格约束了「只返回 JSON」。OAuth / authentication 相关报错。如果你在 Claude Code 或 Codex 这类工具里配置注意它们各自有独立的鉴权文件。Claude Code 走settings.jsonCodex 走auth.jsonCline 走 MCP 配置。这三件套Base URL、Key、Model ID在每个工具里的字段名不一样别混用。以 Codex 的auth.json为例{ base_url: https://taotoken.net/api, api_key: sk-你的Key, model: gpt-4o-mini }Claude Code 的settings.json里对应字段是env下的ANTHROPIC_BASE_URL和ANTHROPIC_API_KEYModel ID 通过model字段指定。Cline 的 MCP 配置里则是baseUrl、apiKey、modelId。字段名对不上就会报鉴权失败看起来像 OAuth 问题其实是配置键写错了。审校循环停不下来。这是多 Agent 特有的坑。检查max_revisions是否生效以及审校 Agent 是否真的返回了布尔值pass。如果审校返回的是字符串true而不是布尔trueif review[pass]会永远为假。在解析后加一层类型转换pass_flag review[pass] is True or str(review[pass]).lower() true。排障时如果拿不准是通道问题还是代码问题先去模型对话页 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 手动发一条消息能通说明通道没问题问题在代码不能通就去看接入文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 核对参数格式。6. 从 demo 到生产终止条件、成本熔断与可观测性demo 跑通容易上生产难。下面几条比拓扑选择更影响最终成败。终止条件是第一优先级。多 Agent 最大的事故来源不是答错而是停不下来——两个 Agent 互相打回、规划器无限拆任务。上面代码里那个max_revisions和for...else不是装饰是保命的。任何反馈闭环都必须有硬上限而且上限要写在编排器层不能指望每个 Agent 自觉。成本会失控。一次单 Agent 调用在多 Agent 里可能变成十几次。审校循环、上下文重复传递token 烧得飞快。务必在编排器层做全局计数和熔断比如累计 token 超过阈值就中止并返回当前最佳结果。这类全局计量更适合沉到平台层统一做业务侧不用每个项目重复造一遍。如果你要长期跑编码类或 Agent 类任务Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 这种按周期计费的方式比按量付费更好控预算。把黑板做成可观测的。出问题时你需要的不是模型日志而是「第几步、哪个 Agent、输入输出是什么」。结构化的共享状态天然就是最好的 trace。建议在Agent.run里加一行日志把name、instruction前 100 字、返回前 100 字记下来排查一条链路通常几分钟就能定位到出错的那一环。错误隔离。单个执行器失败不该拖垮整条链路。给每个 Agent 调用包上重试和降级失败时编排器要能决定是跳过、重试还是中止。上面call_llm里的指数退避只是第一层编排器层还要有「这个 Agent 连续失败两次就换备用模型或跳过」的逻辑。最后一步实操把max_revisions改成 0跑一遍观察审校不通过时链路是否直接强制交付再改成 5观察成本增长曲线。这两个极端值跑过之后你对这套骨架的边界就有手感了。接下来要做的是把角色提示词换成你真实业务里的分工把黑板字段换成你的数据结构其余骨架不用动。