☰
多 Agent 协作系统设计:从拓扑结构到一套能跑起来的编排器(TaoToken 统一 Key 接入版)
2026/10/1 13:18:45 网站建设 项目流程

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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API 地址是 https://taotoken.net/api 。

你需要准备三样东西:Base URL、API Key、Model ID。Base URL 填https://taotoken.net/api,Key 在控制台的 API Keys 页面创建,Model ID 按你实际要用的模型填。这三件套在后面的配置里会反复出现,先记牢。控制台地址是 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API Keys 管理页是 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。如果你还没决定用哪个模型,可以先去模型对话页 https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_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后面不要带/v1,OpenAI SDK 会自己拼路径。Key 从环境变量读更安全,配置文件里可以留空,代码里优先读环境变量。下面是编排器主体:

import os import json import time from dataclasses import dataclass, field CONFIG = json.load(open("orchestrator.json", encoding="utf-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_url=CONFIG["base_url"], api_key=API_KEY, timeout=CONFIG["request_timeout"]) last_err = None for attempt in range(CONFIG["max_retries"]): try: resp = client.chat.completions.create( model=model, messages=[ {"role": "system", "content": system}, {"role": "user", "content": user}, ], temperature=0.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_factory=dict) def context(self, *keys) -> str: picked = {k: self.artifacts.get(k, "") for k in keys} return json.dumps(picked, ensure_ascii=False, indent=2) @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(task=topic) 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"第 {attempt+1} 轮通过" break feedback = review["comment"] board.artifacts["status"] = f"第 {attempt+1} 轮被打回:{feedback}" else: board.artifacts["status"] = "达到最大重写次数,强制交付" return board if __name__ == "__main__": board = Orchestrator().run("多 Agent 系统的工程价值") print(json.dumps(board.artifacts, ensure_ascii=False, indent=2))

这套骨架里,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.json,Codex 走auth.json,Cline 走 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_KEY,Model ID 通过model字段指定。Cline 的 MCP 配置里则是baseUrl、apiKey、modelId。字段名对不上就会报鉴权失败,看起来像 OAuth 问题,其实是配置键写错了。

审校循环停不下来。这是多 Agent 特有的坑。检查max_revisions是否生效,以及审校 Agent 是否真的返回了布尔值pass。如果审校返回的是字符串"true"而不是布尔true,if review["pass"]会永远为假。在解析后加一层类型转换:pass_flag = review["pass"] is True or str(review["pass"]).lower() == "true"。

排障时如果拿不准是通道问题还是代码问题,先去模型对话页 https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 手动发一条消息,能通说明通道没问题,问题在代码;不能通就去看接入文档 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 核对参数格式。

6. 从 demo 到生产:终止条件、成本熔断与可观测性

demo 跑通容易,上生产难。下面几条比拓扑选择更影响最终成败。

终止条件是第一优先级。多 Agent 最大的事故来源不是答错,而是停不下来——两个 Agent 互相打回、规划器无限拆任务。上面代码里那个max_revisions和for...else不是装饰,是保命的。任何反馈闭环都必须有硬上限,而且上限要写在编排器层,不能指望每个 Agent 自觉。

成本会失控。一次单 Agent 调用,在多 Agent 里可能变成十几次。审校循环、上下文重复传递,token 烧得飞快。务必在编排器层做全局计数和熔断,比如累计 token 超过阈值就中止并返回当前最佳结果。这类全局计量更适合沉到平台层统一做,业务侧不用每个项目重复造一遍。如果你要长期跑编码类或 Agent 类任务,Coding Plan https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 这种按周期计费的方式比按量付费更好控预算。

把黑板做成可观测的。出问题时,你需要的不是模型日志,而是「第几步、哪个 Agent、输入输出是什么」。结构化的共享状态天然就是最好的 trace。建议在Agent.run里加一行日志,把name、instruction前 100 字、返回前 100 字记下来,排查一条链路通常几分钟就能定位到出错的那一环。

错误隔离。单个执行器失败,不该拖垮整条链路。给每个 Agent 调用包上重试和降级,失败时编排器要能决定是跳过、重试还是中止。上面call_llm里的指数退避只是第一层,编排器层还要有「这个 Agent 连续失败两次就换备用模型或跳过」的逻辑。

最后一步实操:把max_revisions改成 0,跑一遍,观察审校不通过时链路是否直接强制交付;再改成 5,观察成本增长曲线。这两个极端值跑过之后,你对这套骨架的边界就有手感了。接下来要做的,是把角色提示词换成你真实业务里的分工,把黑板字段换成你的数据结构,其余骨架不用动。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询