1. 从单 Agent 到多 Agent：商用协作系统到底解决了什么问题
多 Agent 协作系统，简单说就是让多个各有所长的 AI Agent 像一支团队一样分工干活，由一个“管理者”负责拆任务、派活、验收、汇总。它适合谁？适合那些单 Agent 已经明显扛不住的场景：一份需要调研+数据分析+写作+评审的深度报告、一个要同时改前端后端还要跑测试的工程任务、一套需要多轮交叉验证的客服或风控流程。单 Agent 的三大瓶颈很现实——上下文窗口装不下超长文档，一个模型很难同时精通代码和创意，复杂多步推理容易在中途跑偏产生幻觉。多 Agent 的思路借鉴了人类组织：把复杂任务拆开，让专业的人做专业的事，再有人统一协调和把关，最终实现 1+1>2。
但真到商用落地，问题就来了。我见过太多团队卡在同一个地方：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_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite ，创建后复制保存。接口地址统一用 https://taotoken.net/api ，注意这个地址不带任何查询参数。想先确认模型是否可用，可以直接在模型对话页试一条请求：https://taotoken.net/model-chat?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite 。如果你打算长期跑编码类或 Agent 类任务，Coding Plan 会更划算，入口在：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite 。接入细节和参数说明统一看文档：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。
注意：API Key 属于敏感凭证，不要硬编码进提交到仓库的代码里，建议用环境变量或本地配置文件加载，并在 .gitignore 中排除。
3. 可复制配置：settings.json 与 config.toml 接入骨架
统一接入层的关键，是让所有 Agent 都从同一份配置读取模型信息。下面给两套骨架，Python 生态用 config.toml，Node/通用场景用 settings.json，按你的技术栈选一套即可。
3.1 config.toml：Python 多 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:guest@localhost: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(path="config.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_url=cfg["llm"]["base_url"], api_key=cfg["llm"]["api_key"], timeout=cfg["llm"]["timeout"], max_retries=cfg["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:guest@localhost: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[f"step_{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(target=consumer, 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 Unauthorized | Key 未加载或环境变量为空 | 检查 TAOTOKEN_API_KEY 是否导出，配置里是否做了变量替换 |
| 404 Not Found | base_url 多写或少写了 /v1 | 统一用 https://taotoken.net/api，路径由 SDK 拼接 |
| 连接超时 | 网络或 timeout 设置过短 | 把 timeout 调到 60s，重试次数设 3 |
| Worker 卡死不退出 | 终止信号未发送或未消费 | 确认哨兵值 None 被每个消费者读到 |
| LangGraph 无限循环 | 条件边判断条件写反 | 检查 current_step 与 plan 长度的比较逻辑 |
| 结果冲突 | 多 Agent 结论矛盾 | 引入 Critic 节点做仲裁，或加权投票 |
提示：多 Agent 系统排查问题时，给每次调用打一个贯穿全链路的 trace_id，日志里带上 Agent 角色名，定位效率会高很多。
6. 下一步：把统一接入沉淀成团队规范
搭到这里，你已经有了一个能跑的多 Agent 协作骨架：配置统一、编排清晰、通信解耦、验证可复现。真正决定它能不能上商用的，往往不是算法多花哨，而是这些工程细节是否稳定。我的建议是把统一 Key 接入这件事固化成团队规范——所有 Agent 只认一个 base_url，模型差异通过配置区分，新增角色不改核心代码。想继续深入接入细节，可以翻接入文档：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ；需要管理多个 Key 或做配额隔离，去控制台：https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ；长期跑编码和 Agent 任务，Coding Plan 的入口在：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite 。先把最小闭环跑通，再按业务反馈逐个扩展 Agent 能力，比一上来追求“万能系统”靠谱得多。
