1. 从单 Agent 到多 Agent商用协作系统到底解决了什么问题多 Agent 协作系统简单说就是让多个各有所长的 AI Agent 像一支团队一样分工干活由一个“管理者”负责拆任务、派活、验收、汇总。它适合谁适合那些单 Agent 已经明显扛不住的场景一份需要调研数据分析写作评审的深度报告、一个要同时改前端后端还要跑测试的工程任务、一套需要多轮交叉验证的客服或风控流程。单 Agent 的三大瓶颈很现实——上下文窗口装不下超长文档一个模型很难同时精通代码和创意复杂多步推理容易在中途跑偏产生幻觉。多 Agent 的思路借鉴了人类组织把复杂任务拆开让专业的人做专业的事再有人统一协调和把关最终实现 112。但真到商用落地问题就来了。我见过太多团队卡在同一个地方Agent 编排逻辑写好了消息队列也搭起来了结果每个 Agent 各自去配一套模型 Key、各自处理鉴权和限流配置散落在十几个文件里换一个模型要改一遍全局排查一个报错要翻五个服务的日志。这篇就围绕 LangGraph、消息队列和 Agent 编排把架构设计讲清楚同时给出一套可复制的统一 Key/API 接入骨架让你从零搭出一个能跑起来的协作系统而不是停留在 demo 阶段。2. 前置准备用 TaoToken 统一管理多 Agent 的模型入口多 Agent 系统里每个 Agent 本质上都是一次或多次大模型调用。如果 Orchestrator、Planner、Worker、Critic 各自维护一套 API Key 和 base_url配置会迅速失控。我的做法是引入一个统一的模型接入层把所有 Agent 的模型请求收敛到同一个入口。这里用 TaoToken 来做这件事它的价值在于一个 Key 就能覆盖多个主流模型Agent 侧只需要认一个 base_url切换模型时改配置而不是改代码。你需要先拿到一个可用的 API Key。访问控制台创建https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 创建后复制保存。接口地址统一用 https://taotoken.net/api 注意这个地址不带任何查询参数。想先确认模型是否可用可以直接在模型对话页试一条请求https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 。如果你打算长期跑编码类或 Agent 类任务Coding Plan 会更划算入口在https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。接入细节和参数说明统一看文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。注意API Key 属于敏感凭证不要硬编码进提交到仓库的代码里建议用环境变量或本地配置文件加载并在 .gitignore 中排除。3. 可复制配置settings.json 与 config.toml 接入骨架统一接入层的关键是让所有 Agent 都从同一份配置读取模型信息。下面给两套骨架Python 生态用 config.tomlNode/通用场景用 settings.json按你的技术栈选一套即可。3.1 config.tomlPython 多 Agent 项目配置# config.toml —— 多 Agent 统一模型接入配置 [llm] # 统一入口所有 Agent 共用 base_url https://taotoken.net/api api_key ${TAOTOKEN_API_KEY} # 从环境变量读取避免明文 timeout 60 max_retries 3 # 不同 Agent 角色可指定不同模型但共用同一个 base_url 和 key [llm.orchestrator] model claude-3-5-sonnet temperature 0.2 [llm.planner] model claude-3-5-sonnet temperature 0.3 [llm.worker] model gpt-4o-mini # 简单子任务用轻量模型控成本 temperature 0.5 [llm.critic] model claude-3-5-sonnet temperature 0.1 # 评审要稳定温度调低 [queue] broker amqp://guest:guestlocalhost:5672/ task_queue agent.tasks result_queue agent.results对应的 Python 加载逻辑用 pydantic 或 tomli 都行核心是把 base_url 和 api_key 注入到每个 Agent 的客户端里import os import tomli from openai import OpenAI def load_config(pathconfig.toml): with open(path, rb) as f: cfg tomli.load(f) # 环境变量覆盖避免明文入库 cfg[llm][api_key] os.environ.get(TAOTOKEN_API_KEY, cfg[llm][api_key]) return cfg def build_client(cfg, role: str) - OpenAI: 为指定角色的 Agent 构建统一客户端 role_cfg cfg[llm].get(role, {}) return OpenAI( base_urlcfg[llm][base_url], api_keycfg[llm][api_key], timeoutcfg[llm][timeout], max_retriescfg[llm][max_retries], ) # 各 Agent 复用同一入口只是模型不同 cfg load_config() orchestrator_client build_client(cfg, orchestrator) worker_client build_client(cfg, worker)3.2 settings.json通用/Node 场景配置{ llm: { baseUrl: https://taotoken.net/api, apiKey: ${TAOTOKEN_API_KEY}, timeout: 60000, maxRetries: 3, roles: { orchestrator: { model: claude-3-5-sonnet, temperature: 0.2 }, planner: { model: claude-3-5-sonnet, temperature: 0.3 }, worker: { model: gpt-4o-mini, temperature: 0.5 }, critic: { model: claude-3-5-sonnet, temperature: 0.1 } } }, queue: { broker: amqp://guest:guestlocalhost:5672/, taskQueue: agent.tasks, resultQueue: agent.results } }Node 侧读取时把${TAOTOKEN_API_KEY}替换为 process.env 的值即可。这样做的收益很直接新增一个 Agent 角色只在 roles 里加一段换模型改 model 字段换供应商只动 baseUrl 一处。4. 架构落地LangGraph 编排 消息队列异步通信配置统一之后进入编排层。这里分两块用 LangGraph 管状态流转用消息队列管 Agent 之间的异步解耦。4.1 LangGraph 定义协作状态图LangGraph 的核心是“状态 节点 条件边”。下面这段把 Orchestrator、Planner、Worker、Aggregator 串成一个可循环调度的图Worker 每完成一步就回到 Orchestrator 继续派活直到所有子任务完成再汇总。from typing import TypedDict, List, Dict, Any, Optional from langgraph.graph import StateGraph, END class AgentState(TypedDict): task: str plan: Optional[List[str]] current_step: int results: Dict[str, Any] final_output: Optional[str] def orchestrator_node(state: AgentState) - AgentState: # 中枢只做状态判断与路由不碰业务逻辑 return state def planner_node(state: AgentState) - AgentState: # 实际项目这里调用统一客户端做任务分解 plan [收集数据, 分析数据, 生成图表, 撰写报告] return {**state, plan: plan, current_step: 0} def worker_node(state: AgentState) - AgentState: idx state[current_step] desc state[plan][idx] results dict(state.get(results, {})) results[fstep_{idx}] {desc: desc, output: f{desc} 完成} return {**state, results: results, current_step: idx 1} def aggregator_node(state: AgentState) - AgentState: lines [f- {v[desc]}: {v[output]} for v in state[results].values()] return {**state, final_output: \n.join(lines)} workflow StateGraph(AgentState) workflow.add_node(orchestrator, orchestrator_node) workflow.add_node(planner, planner_node) workflow.add_node(worker, worker_node) workflow.add_node(aggregator, aggregator_node) workflow.set_entry_point(orchestrator) workflow.add_conditional_edges( orchestrator, lambda s: planner if not s.get(plan) else (aggregator if s[current_step] len(s.get(plan, [])) else worker), {planner: planner, worker: worker, aggregator: aggregator}, ) workflow.add_edge(planner, orchestrator) workflow.add_edge(worker, orchestrator) workflow.add_edge(aggregator, END) app workflow.compile()4.2 消息队列做 Agent 间异步解耦LangGraph 管的是单个任务内部的状态流转而多个独立 Agent 服务之间用消息队列解耦更合适。生产者投任务不等结果消费者并行取任务处理吞吐量随 Worker 数量线性扩展。import queue import threading import time def producer(q: queue.Queue, n: int 5): for i in range(1, n 1): q.put({task_id: i, type: analysis, payload: f任务{i}}) print(f[Producer] 投放任务 {i}队列大小 {q.qsize()}) time.sleep(0.3) q.put(None) # 终止信号 def consumer(q: queue.Queue, wid: int): while True: task q.get() if task is None: q.task_done() break print(f[Worker-{wid}] 处理任务 {task[task_id]}) time.sleep(0.5) # 模拟推理耗时 q.task_done() q queue.Queue() threads [threading.Thread(targetconsumer, args(q, i)) for i in range(1, 3)] for t in threads: t.start() producer(q, 6) q.join() for t in threads: t.join() print(全部任务处理完毕)生产环境把 queue.Queue 换成 RabbitMQ 或 Kafka把线程换成独立部署的 Agent 服务即可接口契约不变。5. 验证请求与常见报错排查配置和代码就位后先做一次最小验证确认统一入口是通的。用 curl 直接打一次对话接口curl https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -H Content-Type: application/json \ -d { model: claude-3-5-sonnet, messages: [{role: user, content: 回复 OK 两个字母}] }返回里能看到正常的 choices 结构说明 Key 和 base_url 都没问题。接着跑一遍上面的 LangGraph 示例预期输出是四个子任务依次完成并汇总成一段报告文本。下面是我在实际搭建中踩过的几个高频坑对照排查能省不少时间报错现象常见原因处理方式401 UnauthorizedKey 未加载或环境变量为空检查 TAOTOKEN_API_KEY 是否导出配置里是否做了变量替换404 Not Foundbase_url 多写或少写了 /v1统一用 https://taotoken.net/api路径由 SDK 拼接连接超时网络或 timeout 设置过短把 timeout 调到 60s重试次数设 3Worker 卡死不退出终止信号未发送或未消费确认哨兵值 None 被每个消费者读到LangGraph 无限循环条件边判断条件写反检查 current_step 与 plan 长度的比较逻辑结果冲突多 Agent 结论矛盾引入 Critic 节点做仲裁或加权投票提示多 Agent 系统排查问题时给每次调用打一个贯穿全链路的 trace_id日志里带上 Agent 角色名定位效率会高很多。6. 下一步把统一接入沉淀成团队规范搭到这里你已经有了一个能跑的多 Agent 协作骨架配置统一、编排清晰、通信解耦、验证可复现。真正决定它能不能上商用的往往不是算法多花哨而是这些工程细节是否稳定。我的建议是把统一 Key 接入这件事固化成团队规范——所有 Agent 只认一个 base_url模型差异通过配置区分新增角色不改核心代码。想继续深入接入细节可以翻接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 需要管理多个 Key 或做配额隔离去控制台https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 长期跑编码和 Agent 任务Coding Plan 的入口在https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。先把最小闭环跑通再按业务反馈逐个扩展 Agent 能力比一上来追求“万能系统”靠谱得多。
1. 从单 Agent 到多 Agent商用协作系统到底解决了什么问题多 Agent 协作系统简单说就是让多个各有所长的 AI Agent 像一支团队一样分工干活由一个“管理者”负责拆任务、派活、验收、汇总。它适合谁适合那些单 Agent 已经明显扛不住的场景一份需要调研数据分析写作评审的深度报告、一个要同时改前端后端还要跑测试的工程任务、一套需要多轮交叉验证的客服或风控流程。单 Agent 的三大瓶颈很现实——上下文窗口装不下超长文档一个模型很难同时精通代码和创意复杂多步推理容易在中途跑偏产生幻觉。多 Agent 的思路借鉴了人类组织把复杂任务拆开让专业的人做专业的事再有人统一协调和把关最终实现 112。但真到商用落地问题就来了。我见过太多团队卡在同一个地方Agent 编排逻辑写好了消息队列也搭起来了结果每个 Agent 各自去配一套模型 Key、各自处理鉴权和限流配置散落在十几个文件里换一个模型要改一遍全局排查一个报错要翻五个服务的日志。这篇就围绕 LangGraph、消息队列和 Agent 编排把架构设计讲清楚同时给出一套可复制的统一 Key/API 接入骨架让你从零搭出一个能跑起来的协作系统而不是停留在 demo 阶段。2. 前置准备用 TaoToken 统一管理多 Agent 的模型入口多 Agent 系统里每个 Agent 本质上都是一次或多次大模型调用。如果 Orchestrator、Planner、Worker、Critic 各自维护一套 API Key 和 base_url配置会迅速失控。我的做法是引入一个统一的模型接入层把所有 Agent 的模型请求收敛到同一个入口。这里用 TaoToken 来做这件事它的价值在于一个 Key 就能覆盖多个主流模型Agent 侧只需要认一个 base_url切换模型时改配置而不是改代码。你需要先拿到一个可用的 API Key。访问控制台创建https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 创建后复制保存。接口地址统一用 https://taotoken.net/api 注意这个地址不带任何查询参数。想先确认模型是否可用可以直接在模型对话页试一条请求https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 。如果你打算长期跑编码类或 Agent 类任务Coding Plan 会更划算入口在https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。接入细节和参数说明统一看文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。注意API Key 属于敏感凭证不要硬编码进提交到仓库的代码里建议用环境变量或本地配置文件加载并在 .gitignore 中排除。3. 可复制配置settings.json 与 config.toml 接入骨架统一接入层的关键是让所有 Agent 都从同一份配置读取模型信息。下面给两套骨架Python 生态用 config.tomlNode/通用场景用 settings.json按你的技术栈选一套即可。3.1 config.tomlPython 多 Agent 项目配置# config.toml —— 多 Agent 统一模型接入配置 [llm] # 统一入口所有 Agent 共用 base_url https://taotoken.net/api api_key ${TAOTOKEN_API_KEY} # 从环境变量读取避免明文 timeout 60 max_retries 3 # 不同 Agent 角色可指定不同模型但共用同一个 base_url 和 key [llm.orchestrator] model claude-3-5-sonnet temperature 0.2 [llm.planner] model claude-3-5-sonnet temperature 0.3 [llm.worker] model gpt-4o-mini # 简单子任务用轻量模型控成本 temperature 0.5 [llm.critic] model claude-3-5-sonnet temperature 0.1 # 评审要稳定温度调低 [queue] broker amqp://guest:guestlocalhost:5672/ task_queue agent.tasks result_queue agent.results对应的 Python 加载逻辑用 pydantic 或 tomli 都行核心是把 base_url 和 api_key 注入到每个 Agent 的客户端里import os import tomli from openai import OpenAI def load_config(pathconfig.toml): with open(path, rb) as f: cfg tomli.load(f) # 环境变量覆盖避免明文入库 cfg[llm][api_key] os.environ.get(TAOTOKEN_API_KEY, cfg[llm][api_key]) return cfg def build_client(cfg, role: str) - OpenAI: 为指定角色的 Agent 构建统一客户端 role_cfg cfg[llm].get(role, {}) return OpenAI( base_urlcfg[llm][base_url], api_keycfg[llm][api_key], timeoutcfg[llm][timeout], max_retriescfg[llm][max_retries], ) # 各 Agent 复用同一入口只是模型不同 cfg load_config() orchestrator_client build_client(cfg, orchestrator) worker_client build_client(cfg, worker)3.2 settings.json通用/Node 场景配置{ llm: { baseUrl: https://taotoken.net/api, apiKey: ${TAOTOKEN_API_KEY}, timeout: 60000, maxRetries: 3, roles: { orchestrator: { model: claude-3-5-sonnet, temperature: 0.2 }, planner: { model: claude-3-5-sonnet, temperature: 0.3 }, worker: { model: gpt-4o-mini, temperature: 0.5 }, critic: { model: claude-3-5-sonnet, temperature: 0.1 } } }, queue: { broker: amqp://guest:guestlocalhost:5672/, taskQueue: agent.tasks, resultQueue: agent.results } }Node 侧读取时把${TAOTOKEN_API_KEY}替换为 process.env 的值即可。这样做的收益很直接新增一个 Agent 角色只在 roles 里加一段换模型改 model 字段换供应商只动 baseUrl 一处。4. 架构落地LangGraph 编排 消息队列异步通信配置统一之后进入编排层。这里分两块用 LangGraph 管状态流转用消息队列管 Agent 之间的异步解耦。4.1 LangGraph 定义协作状态图LangGraph 的核心是“状态 节点 条件边”。下面这段把 Orchestrator、Planner、Worker、Aggregator 串成一个可循环调度的图Worker 每完成一步就回到 Orchestrator 继续派活直到所有子任务完成再汇总。from typing import TypedDict, List, Dict, Any, Optional from langgraph.graph import StateGraph, END class AgentState(TypedDict): task: str plan: Optional[List[str]] current_step: int results: Dict[str, Any] final_output: Optional[str] def orchestrator_node(state: AgentState) - AgentState: # 中枢只做状态判断与路由不碰业务逻辑 return state def planner_node(state: AgentState) - AgentState: # 实际项目这里调用统一客户端做任务分解 plan [收集数据, 分析数据, 生成图表, 撰写报告] return {**state, plan: plan, current_step: 0} def worker_node(state: AgentState) - AgentState: idx state[current_step] desc state[plan][idx] results dict(state.get(results, {})) results[fstep_{idx}] {desc: desc, output: f{desc} 完成} return {**state, results: results, current_step: idx 1} def aggregator_node(state: AgentState) - AgentState: lines [f- {v[desc]}: {v[output]} for v in state[results].values()] return {**state, final_output: \n.join(lines)} workflow StateGraph(AgentState) workflow.add_node(orchestrator, orchestrator_node) workflow.add_node(planner, planner_node) workflow.add_node(worker, worker_node) workflow.add_node(aggregator, aggregator_node) workflow.set_entry_point(orchestrator) workflow.add_conditional_edges( orchestrator, lambda s: planner if not s.get(plan) else (aggregator if s[current_step] len(s.get(plan, [])) else worker), {planner: planner, worker: worker, aggregator: aggregator}, ) workflow.add_edge(planner, orchestrator) workflow.add_edge(worker, orchestrator) workflow.add_edge(aggregator, END) app workflow.compile()4.2 消息队列做 Agent 间异步解耦LangGraph 管的是单个任务内部的状态流转而多个独立 Agent 服务之间用消息队列解耦更合适。生产者投任务不等结果消费者并行取任务处理吞吐量随 Worker 数量线性扩展。import queue import threading import time def producer(q: queue.Queue, n: int 5): for i in range(1, n 1): q.put({task_id: i, type: analysis, payload: f任务{i}}) print(f[Producer] 投放任务 {i}队列大小 {q.qsize()}) time.sleep(0.3) q.put(None) # 终止信号 def consumer(q: queue.Queue, wid: int): while True: task q.get() if task is None: q.task_done() break print(f[Worker-{wid}] 处理任务 {task[task_id]}) time.sleep(0.5) # 模拟推理耗时 q.task_done() q queue.Queue() threads [threading.Thread(targetconsumer, args(q, i)) for i in range(1, 3)] for t in threads: t.start() producer(q, 6) q.join() for t in threads: t.join() print(全部任务处理完毕)生产环境把 queue.Queue 换成 RabbitMQ 或 Kafka把线程换成独立部署的 Agent 服务即可接口契约不变。5. 验证请求与常见报错排查配置和代码就位后先做一次最小验证确认统一入口是通的。用 curl 直接打一次对话接口curl https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -H Content-Type: application/json \ -d { model: claude-3-5-sonnet, messages: [{role: user, content: 回复 OK 两个字母}] }返回里能看到正常的 choices 结构说明 Key 和 base_url 都没问题。接着跑一遍上面的 LangGraph 示例预期输出是四个子任务依次完成并汇总成一段报告文本。下面是我在实际搭建中踩过的几个高频坑对照排查能省不少时间报错现象常见原因处理方式401 UnauthorizedKey 未加载或环境变量为空检查 TAOTOKEN_API_KEY 是否导出配置里是否做了变量替换404 Not Foundbase_url 多写或少写了 /v1统一用 https://taotoken.net/api路径由 SDK 拼接连接超时网络或 timeout 设置过短把 timeout 调到 60s重试次数设 3Worker 卡死不退出终止信号未发送或未消费确认哨兵值 None 被每个消费者读到LangGraph 无限循环条件边判断条件写反检查 current_step 与 plan 长度的比较逻辑结果冲突多 Agent 结论矛盾引入 Critic 节点做仲裁或加权投票提示多 Agent 系统排查问题时给每次调用打一个贯穿全链路的 trace_id日志里带上 Agent 角色名定位效率会高很多。6. 下一步把统一接入沉淀成团队规范搭到这里你已经有了一个能跑的多 Agent 协作骨架配置统一、编排清晰、通信解耦、验证可复现。真正决定它能不能上商用的往往不是算法多花哨而是这些工程细节是否稳定。我的建议是把统一 Key 接入这件事固化成团队规范——所有 Agent 只认一个 base_url模型差异通过配置区分新增角色不改核心代码。想继续深入接入细节可以翻接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 需要管理多个 Key 或做配额隔离去控制台https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 长期跑编码和 Agent 任务Coding Plan 的入口在https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。先把最小闭环跑通再按业务反馈逐个扩展 Agent 能力比一上来追求“万能系统”靠谱得多。
