1. 多 Agent 框架选型Router 路由器到底解决什么问题多 Agent 框架最容易踩的坑不是模型不够强而是把「谁来决定、谁来执行」这件事搞混了。我见过不少项目一上来就堆四五个 Agent结果每个 Agent 都在抢着回答最后输出一段四不像的内容。Router路由器模式的核心价值就是把「分诊」和「执行」拆开Router 本身不干活它只做意图识别、并行派活、结果汇总。用大白话讲Router 就像医院分诊台。你进门说「我肚子疼还发烧」分诊护士不会给你开刀她判断这属于消化内科 发热门诊把你派给对应医生两个医生各自检查最后护士把两份结论拼成一份给你。Router 就是这个护士识别问题类型、调用多个 Agent 并行执行、合成结果返回。它自己绝不执行业务逻辑。这种模式适合三类场景。第一类是问题分类明确、需要不同专家处理的比如客服入口里「查物流」和「申请退款」必须走不同流程不能让订单专员去操作退款通道。第二类是每个 Agent 有专属工具和知识库的订单 Agent 能查物流系统退款 Agent 能调退款接口Router 清楚谁擅长什么不会乱派活。第三类是一次提问要多个结果合并的比如你同时想知道「订单发没发货」和「优惠券能不能退」Router 可以并行调用两个 Agent省掉你问两次的时间。对比 LangChain 生态里常见的四种多 Agent 实现方式职责划分差异很明显。Skills 模式是当前 AI 全程主控只把具体环节当工具调用控制权不转移适合能力复用、流程固定的任务。Subagent 模式是主 AI 定方案拆任务子 AI 执行细分事项最终仍由主 AI 统筹适合复杂任务拆解。Router 模式是路由 AI 先判断任务类型再匹配给对口领域 AI分配完成后控制权转移给对应 Agent。Handoff 模式是当前 AI 发现处理不了主动把整段对话交接给专业 AI原 AI 彻底退出适合多轮对话中切换专业角色。这四种模式不是互斥的实际项目里经常组合使用入口用 Router 拦截分流分流后的 Agent 内部用 Subagent 拆解任务执行中需要专家角色时用 Handoff 切换需要复用能力时加载 Skills。理解这个组合关系比死记四种定义有用得多。接下来我会给出可复制的 Router 配置片段和多 Agent 编排示例并演示如何通过 TaoToken 统一 API 通道接入模型服务让多个 Agent 共用一套 Key 和 Base URL省掉每个 Agent 单独配密钥的麻烦。2. TaoToken 前置准备统一 Key 与 API 通道接入多 Agent多 Agent 项目最先遇到的工程问题往往不是编排逻辑而是密钥管理。四个 Agent 各配一套 Key、各写一个 Base URL改起来容易漏排查 401 时得逐个翻配置文件。TaoToken 的思路是提供统一 API 通道多个 Agent 共用同一个 Key 和同一个 Base URL模型 ID 按需切换。这样你的 Router、Subagent、Handoff 各角色可以指向不同模型但接入层只有一套配置。先明确三个要素后面所有配置都围绕它们展开Base URL 是https://taotoken.net/apiAPI Key 在控制台创建Model ID 按你实际调用的模型填写。这三个要素在 Claude Code、Cline MCP、Codex 的 auth.json 里出现时必须写全缺一个就会报错。创建 Key 的路径是进入控制台后找到 API Keys 页面新建一个密钥并复制保存。注意 Key 只在创建时完整显示一次关掉页面就看不到了建议直接存进项目的.env文件而不是硬编码在代码里。控制台地址是https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleAPI Keys 页面是https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keys。环境变量建议这样组织把统一通道和模型 ID 分开管理# .env TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_API_KEYsk-你的密钥 ROUTER_MODELgpt-4o-mini AGENT_ORDER_MODELgpt-4o AGENT_REFUND_MODELgpt-4o这里的设计意图是Router 只需要做意图识别和分流用便宜快的小模型就够订单 Agent 和退款 Agent 要处理具体业务用能力更强的模型。统一通道的好处是这些模型 ID 可以随时替换Base URL 和 Key 不用动。如果你用的是 LangChain 的ChatOpenAI封装接入方式很直接因为 TaoToken 的 API 兼容 OpenAI 协议格式import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() def build_llm(model_env_key: str, temperature: float 0.2): return ChatOpenAI( modelos.getenv(model_env_key), api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL), temperaturetemperature, timeout60, max_retries2, ) router_llm build_llm(ROUTER_MODEL, 0.0) order_llm build_llm(AGENT_ORDER_MODEL, 0.2) refund_llm build_llm(AGENT_REFUND_MODEL, 0.2)这段代码的关键点是base_url指向统一通道api_key从环境变量读取。temperature0.0给 Router 用是因为分流判断需要稳定输出不能这次判成订单、下次判成退款。max_retries2是应对偶发网络抖动多 Agent 并行调用时单个请求失败不该拖垮整个流程。如果你用 Claude Code 做开发辅助配置方式是在项目根目录的 settings 里写全三件套。Claude Code 的配置文件通常放在.claude/settings.json内容形如{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_AUTH_TOKEN: sk-你的密钥, ANTHROPIC_MODEL: claude-sonnet-4-20250514 } }注意这里的变量名是 Anthropic 协议对应的Base URL 仍然是统一通道Key 用同一个。如果你同时用 Cline 的 MCP 配置MCP server 的配置里也要写全 Base URL、Key、Model ID 三项缺 Model ID 时部分客户端会回退到默认模型导致行为不一致。Codex 的auth.json同理字段名不同但三要素不变。统一通道还有一个实际好处多 Agent 并行调用时你只需要在一个地方看用量和错误。如果四个 Agent 各走各的通道排查「为什么退款 Agent 超时」得登录四个后台。现在所有请求都经过同一个入口日志和配额集中排障效率高很多。接入文档在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdoc里面有各客户端的完整字段说明。3. 可复制配置Router 路由片段与多 Agent 编排示例这一节给出能直接跑的配置和代码。先看 Router 的核心逻辑它接收用户问题输出一个结构化分流结果包含要调用的 Agent 列表和每个 Agent 的子任务。这里用 JSON 输出约束避免模型返回自由文本导致解析失败。Router 的提示词和结构化输出配置如下from pydantic import BaseModel, Field from typing import List class RouteDecision(BaseModel): agents: List[str] Field(description需要调用的 Agent 名称列表) subtasks: List[str] Field(description与 agents 一一对应的子任务描述) reason: str Field(description分流理由一句话) ROUTER_PROMPT 你是任务分诊器不执行任何业务操作。 可用 Agent - order_agent查询订单状态、物流信息 - refund_agent处理退款申请、优惠券退还 - faq_agent回答通用政策问题 根据用户问题判断需要哪些 Agent并为每个 Agent 写清子任务。 只输出 JSON不要解释。 def route_question(question: str) - RouteDecision: structured_llm router_llm.with_structured_output(RouteDecision) messages [ {role: system, content: ROUTER_PROMPT}, {role: user, content: question}, ] return structured_llm.invoke(messages)with_structured_output是 LangChain 提供的结构化输出封装底层会约束模型返回符合RouteDecision的 JSON。实测下来给 Router 加一句「只输出 JSON不要解释」能显著降低解析失败率因为部分模型喜欢在 JSON 前后加客套话。接下来是并行调用多个 Agent 并汇总结果。这里用asyncio.gather实现并行避免串行等待import asyncio from langchain_core.messages import HumanMessage, SystemMessage AGENT_PROMPTS { order_agent: 你是订单专员只回答订单和物流相关问题。, refund_agent: 你是退款专员只处理退款和优惠券退还。, faq_agent: 你是政策问答专员回答通用规则问题。, } AGENT_LLMS { order_agent: order_llm, refund_agent: refund_llm, faq_agent: router_llm, } async def run_single_agent(agent_name: str, subtask: str) - str: llm AGENT_LLMS[agent_name] messages [ SystemMessage(contentAGENT_PROMPTS[agent_name]), HumanMessage(contentsubtask), ] resp await llm.ainvoke(messages) return f[{agent_name}] {resp.content} async def run_router_flow(question: str) - str: decision route_question(question) tasks [ run_single_agent(name, subtask) for name, subtask in zip(decision.agents, decision.subtasks) ] results await asyncio.gather(*tasks, return_exceptionsTrue) merged [] for r in results: if isinstance(r, Exception): merged.append(f[执行失败] {type(r).__name__}: {r}) else: merged.append(r) return \n\n.join(merged)这段编排的关键设计有三点。第一route_question是同步调用因为分流只需要一次模型请求没必要异步化增加复杂度。第二asyncio.gather带return_exceptionsTrue单个 Agent 失败不会中断其他 Agent失败信息会作为结果的一部分返回方便定位是哪个 Agent 出了问题。第三每个 Agent 的提示词里明确限定职责范围防止订单 Agent 越界回答退款问题。如果你用 LangGraph 实现Router 对应的是条件边conditional edge。核心结构是入口节点做分流根据分流结果路由到不同 Agent 节点最后汇聚到合成节点。LangGraph 的配置片段如下from langgraph.graph import StateGraph, END from typing import TypedDict, List class GraphState(TypedDict): question: str agents: List[str] subtasks: List[str] results: List[str] answer: str def router_node(state: GraphState) - GraphState: decision route_question(state[question]) return {**state, agents: decision.agents, subtasks: decision.subtasks} def make_agent_node(agent_name: str): def node(state: GraphState) - GraphState: idx state[agents].index(agent_name) subtask state[subtasks][idx] llm AGENT_LLMS[agent_name] resp llm.invoke([ SystemMessage(contentAGENT_PROMPTS[agent_name]), HumanMessage(contentsubtask), ]) return {**state, results: state[results] [resp.content]} return node def route_selector(state: GraphState) - List[str]: return state[agents] builder StateGraph(GraphState) builder.add_node(router, router_node) for name in AGENT_PROMPTS: builder.add_node(name, make_agent_node(name)) builder.set_entry_point(router) builder.add_conditional_edges(router, route_selector, {n: n for n in AGENT_PROMPTS}) for name in AGENT_PROMPTS: builder.add_edge(name, END) graph builder.compile()LangGraph 版本的优势是分流路径可视化调试时能看到请求走了哪条边。注意add_conditional_edges的映射表必须覆盖所有可能的 Agent 名称漏写会导致运行时找不到目标节点。4. 验证请求与成功结果从单 Agent 到完整链路配置写完后必须逐层验证不要一上来就跑完整链路。我的习惯是先验证统一通道连通性再验证 Router 分流最后验证并行汇总。这样出错时能快速定位是哪一层的问题。第一步验证 Base URL 和 Key 是否可用。用 curl 发一个最小请求curl -s 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: 回复 OK 两个字母}], max_tokens: 10 }成功时返回的 JSON 里choices[0].message.content应该是OK或类似内容。如果返回 401说明 Key 有问题如果返回 404检查 Base URL 是否漏了/api后缀如果返回model not found说明 Model ID 写错了去控制台确认可用模型列表。第二步验证 Router 分流是否准确。单独调用route_question打印结构化结果decision route_question(我上周买的耳机还没发货另外优惠券能退吗) print(decision.agents) print(decision.subtasks)预期输出类似[order_agent, refund_agent]子任务分别是查物流和退优惠券。如果 Router 只返回了一个 Agent说明提示词里的 Agent 描述不够清晰或者模型能力不足可以换更强的模型做分流。这一步的验证标准是多意图问题必须分流到多个 Agent单意图问题不能过度分流。第三步跑完整链路并检查汇总结果result asyncio.run(run_router_flow(我上周买的耳机还没发货另外优惠券能退吗)) print(result)成功的结果应该包含两段带 Agent 标签的内容分别回答物流和退款问题。如果某段显示[执行失败]根据异常类型排查TimeoutError是超时调大timeout参数AuthenticationError是 Key 问题RateLimitError是并发太高降低并行数或加重试。第四步如果你用 LangGraph验证图结构是否正确out graph.invoke({question: 订单没发货优惠券能退吗, agents: [], subtasks: [], results: [], answer: }) print(out[results])LangGraph 的常见问题是条件边映射不全报错信息通常是Node not found或Invalid edge。对照add_conditional_edges的映射表确认每个 Agent 名称都有对应节点。验证通过后建议把这条链路固化成冒烟测试每次改配置后跑一遍。多 Agent 系统的配置项多人工回归容易漏一个 30 秒的脚本能省掉很多排查时间。模型对话页面可以用来快速对比不同模型的分流效果地址是https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentchat把同样的多意图问题丢进去看不同模型返回的分流 JSON 是否稳定。5. 常见错误排查401、local proxy failed 与 reading choices多 Agent 接入统一通道时报错集中在几个固定位置。这一节按真实报错信息对照排查每条都给出触发原因和修复动作。401 Unauthorized。最常见的原因是 Key 没读到。检查三处.env文件是否被load_dotenv()正确加载环境变量名是否和代码里一致Key 是否带了多余空格。有个隐蔽情况是 Key 复制时带了换行符os.getenv读出来末尾有\n请求头里就变成非法字符。修复方式是在读取后加.strip()。另一个原因是 Key 被删除或过期去 API Keys 页面确认状态。local proxy failed / connection refused。这个报错通常出现在客户端配置了本地代理但代理没启动。如果你在 Claude Code 或 Cline 里看到这个检查 settings 里是否残留了HTTP_PROXY或HTTPS_PROXY环境变量。统一通道是直连的不需要额外代理配置把相关环境变量清掉即可。还有一种情况是 Base URL 写成了https://taotoken.net/api/带尾斜杠部分客户端拼接路径时会产生双斜杠导致连接异常去掉尾斜杠。reading choices of undefined。这个报错说明响应体里没有choices字段通常是请求根本没成功但代码没检查状态码。触发原因有三类Base URL 写错导致返回了 HTML 错误页Model ID 不存在导致返回错误 JSON请求体格式不对被服务端拒绝。修复方式是先在代码里打印完整响应resp llm.invoke(messages) print(resp.response_metadata)response_metadata里能看到实际的状态码和原始响应。确认是 404 就改 Base URL是 400 就检查请求体字段是 401 就回到上一条排查 Key。OAuth 相关报错。如果你在 Claude Code 里看到 OAuth 认证失败说明客户端走了 OAuth 流程而不是 API Key 流程。统一通道用的是 Key 认证需要在 settings 里显式配置ANTHROPIC_AUTH_TOKEN并且不要同时配置 OAuth 相关字段。两者冲突时客户端优先走 OAuth导致 Key 不生效。清理掉 OAuth 配置只保留 Base URL、Key、Model ID 三件套。结构化输出解析失败。Router 用with_structured_output时偶尔报 JSON 解析错误原因是模型在 JSON 前后加了说明文字。修复方式是在提示词末尾强调「只输出 JSON」并在代码里加一次重试from langchain_core.exceptions import OutputParserException def route_with_retry(question: str, retries: int 2) - RouteDecision: for i in range(retries): try: return route_question(question) except OutputParserException: if i retries - 1: raise raise RuntimeError(unreachable)并行调用部分失败。asyncio.gather带return_exceptionsTrue时失败信息混在结果里容易被忽略。建议在汇总前先过滤异常并打日志明确哪个 Agent 失败、失败原因是什么。如果失败集中在某个 Agent检查该 Agent 的模型 ID 是否可用、提示词是否触发了内容过滤。排查顺序建议固定为先 curl 验证通道再单测 Router再单测单个 Agent最后跑完整链路。这个顺序能把问题范围逐层缩小避免在完整链路里大海捞针。接入文档里有各客户端完整的字段对照表遇到字段名不确定时优先查文档而不是猜。6. 长期编码与 Agent 项目落地建议多 Agent 项目从 demo 到稳定运行差距主要在配置管理和模型选型上。Router 用便宜模型做分流、业务 Agent 用强模型做执行这个分层能显著控制成本。统一通道的价值在于模型 ID 可以随时替换你今天用 A 模型做分流效果不好改成 B 模型只需要改一个环境变量Base URL 和 Key 不动。如果你打算长期做 Agent 开发建议把 Coding Plan 用起来地址是https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-plan。它适合需要持续调用模型做编码辅助和 Agent 调试的场景配额和通道统一管理不用每次调试都担心额度。配合 API Keys 页面做密钥轮换开发环境和生产环境用不同的 Key出问题能快速隔离。最后给一个实操建议把 Router 的分流结果落日志记录每次请求的agents、subtasks、reason和耗时。跑一周后回看日志你会发现哪些意图经常被误分流针对性优化提示词比盲目换模型有效得多。多 Agent 系统的调优是数据驱动的没有日志就只能凭感觉。
1. 多 Agent 框架选型Router 路由器到底解决什么问题多 Agent 框架最容易踩的坑不是模型不够强而是把「谁来决定、谁来执行」这件事搞混了。我见过不少项目一上来就堆四五个 Agent结果每个 Agent 都在抢着回答最后输出一段四不像的内容。Router路由器模式的核心价值就是把「分诊」和「执行」拆开Router 本身不干活它只做意图识别、并行派活、结果汇总。用大白话讲Router 就像医院分诊台。你进门说「我肚子疼还发烧」分诊护士不会给你开刀她判断这属于消化内科 发热门诊把你派给对应医生两个医生各自检查最后护士把两份结论拼成一份给你。Router 就是这个护士识别问题类型、调用多个 Agent 并行执行、合成结果返回。它自己绝不执行业务逻辑。这种模式适合三类场景。第一类是问题分类明确、需要不同专家处理的比如客服入口里「查物流」和「申请退款」必须走不同流程不能让订单专员去操作退款通道。第二类是每个 Agent 有专属工具和知识库的订单 Agent 能查物流系统退款 Agent 能调退款接口Router 清楚谁擅长什么不会乱派活。第三类是一次提问要多个结果合并的比如你同时想知道「订单发没发货」和「优惠券能不能退」Router 可以并行调用两个 Agent省掉你问两次的时间。对比 LangChain 生态里常见的四种多 Agent 实现方式职责划分差异很明显。Skills 模式是当前 AI 全程主控只把具体环节当工具调用控制权不转移适合能力复用、流程固定的任务。Subagent 模式是主 AI 定方案拆任务子 AI 执行细分事项最终仍由主 AI 统筹适合复杂任务拆解。Router 模式是路由 AI 先判断任务类型再匹配给对口领域 AI分配完成后控制权转移给对应 Agent。Handoff 模式是当前 AI 发现处理不了主动把整段对话交接给专业 AI原 AI 彻底退出适合多轮对话中切换专业角色。这四种模式不是互斥的实际项目里经常组合使用入口用 Router 拦截分流分流后的 Agent 内部用 Subagent 拆解任务执行中需要专家角色时用 Handoff 切换需要复用能力时加载 Skills。理解这个组合关系比死记四种定义有用得多。接下来我会给出可复制的 Router 配置片段和多 Agent 编排示例并演示如何通过 TaoToken 统一 API 通道接入模型服务让多个 Agent 共用一套 Key 和 Base URL省掉每个 Agent 单独配密钥的麻烦。2. TaoToken 前置准备统一 Key 与 API 通道接入多 Agent多 Agent 项目最先遇到的工程问题往往不是编排逻辑而是密钥管理。四个 Agent 各配一套 Key、各写一个 Base URL改起来容易漏排查 401 时得逐个翻配置文件。TaoToken 的思路是提供统一 API 通道多个 Agent 共用同一个 Key 和同一个 Base URL模型 ID 按需切换。这样你的 Router、Subagent、Handoff 各角色可以指向不同模型但接入层只有一套配置。先明确三个要素后面所有配置都围绕它们展开Base URL 是https://taotoken.net/apiAPI Key 在控制台创建Model ID 按你实际调用的模型填写。这三个要素在 Claude Code、Cline MCP、Codex 的 auth.json 里出现时必须写全缺一个就会报错。创建 Key 的路径是进入控制台后找到 API Keys 页面新建一个密钥并复制保存。注意 Key 只在创建时完整显示一次关掉页面就看不到了建议直接存进项目的.env文件而不是硬编码在代码里。控制台地址是https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleAPI Keys 页面是https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keys。环境变量建议这样组织把统一通道和模型 ID 分开管理# .env TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_API_KEYsk-你的密钥 ROUTER_MODELgpt-4o-mini AGENT_ORDER_MODELgpt-4o AGENT_REFUND_MODELgpt-4o这里的设计意图是Router 只需要做意图识别和分流用便宜快的小模型就够订单 Agent 和退款 Agent 要处理具体业务用能力更强的模型。统一通道的好处是这些模型 ID 可以随时替换Base URL 和 Key 不用动。如果你用的是 LangChain 的ChatOpenAI封装接入方式很直接因为 TaoToken 的 API 兼容 OpenAI 协议格式import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() def build_llm(model_env_key: str, temperature: float 0.2): return ChatOpenAI( modelos.getenv(model_env_key), api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL), temperaturetemperature, timeout60, max_retries2, ) router_llm build_llm(ROUTER_MODEL, 0.0) order_llm build_llm(AGENT_ORDER_MODEL, 0.2) refund_llm build_llm(AGENT_REFUND_MODEL, 0.2)这段代码的关键点是base_url指向统一通道api_key从环境变量读取。temperature0.0给 Router 用是因为分流判断需要稳定输出不能这次判成订单、下次判成退款。max_retries2是应对偶发网络抖动多 Agent 并行调用时单个请求失败不该拖垮整个流程。如果你用 Claude Code 做开发辅助配置方式是在项目根目录的 settings 里写全三件套。Claude Code 的配置文件通常放在.claude/settings.json内容形如{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_AUTH_TOKEN: sk-你的密钥, ANTHROPIC_MODEL: claude-sonnet-4-20250514 } }注意这里的变量名是 Anthropic 协议对应的Base URL 仍然是统一通道Key 用同一个。如果你同时用 Cline 的 MCP 配置MCP server 的配置里也要写全 Base URL、Key、Model ID 三项缺 Model ID 时部分客户端会回退到默认模型导致行为不一致。Codex 的auth.json同理字段名不同但三要素不变。统一通道还有一个实际好处多 Agent 并行调用时你只需要在一个地方看用量和错误。如果四个 Agent 各走各的通道排查「为什么退款 Agent 超时」得登录四个后台。现在所有请求都经过同一个入口日志和配额集中排障效率高很多。接入文档在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdoc里面有各客户端的完整字段说明。3. 可复制配置Router 路由片段与多 Agent 编排示例这一节给出能直接跑的配置和代码。先看 Router 的核心逻辑它接收用户问题输出一个结构化分流结果包含要调用的 Agent 列表和每个 Agent 的子任务。这里用 JSON 输出约束避免模型返回自由文本导致解析失败。Router 的提示词和结构化输出配置如下from pydantic import BaseModel, Field from typing import List class RouteDecision(BaseModel): agents: List[str] Field(description需要调用的 Agent 名称列表) subtasks: List[str] Field(description与 agents 一一对应的子任务描述) reason: str Field(description分流理由一句话) ROUTER_PROMPT 你是任务分诊器不执行任何业务操作。 可用 Agent - order_agent查询订单状态、物流信息 - refund_agent处理退款申请、优惠券退还 - faq_agent回答通用政策问题 根据用户问题判断需要哪些 Agent并为每个 Agent 写清子任务。 只输出 JSON不要解释。 def route_question(question: str) - RouteDecision: structured_llm router_llm.with_structured_output(RouteDecision) messages [ {role: system, content: ROUTER_PROMPT}, {role: user, content: question}, ] return structured_llm.invoke(messages)with_structured_output是 LangChain 提供的结构化输出封装底层会约束模型返回符合RouteDecision的 JSON。实测下来给 Router 加一句「只输出 JSON不要解释」能显著降低解析失败率因为部分模型喜欢在 JSON 前后加客套话。接下来是并行调用多个 Agent 并汇总结果。这里用asyncio.gather实现并行避免串行等待import asyncio from langchain_core.messages import HumanMessage, SystemMessage AGENT_PROMPTS { order_agent: 你是订单专员只回答订单和物流相关问题。, refund_agent: 你是退款专员只处理退款和优惠券退还。, faq_agent: 你是政策问答专员回答通用规则问题。, } AGENT_LLMS { order_agent: order_llm, refund_agent: refund_llm, faq_agent: router_llm, } async def run_single_agent(agent_name: str, subtask: str) - str: llm AGENT_LLMS[agent_name] messages [ SystemMessage(contentAGENT_PROMPTS[agent_name]), HumanMessage(contentsubtask), ] resp await llm.ainvoke(messages) return f[{agent_name}] {resp.content} async def run_router_flow(question: str) - str: decision route_question(question) tasks [ run_single_agent(name, subtask) for name, subtask in zip(decision.agents, decision.subtasks) ] results await asyncio.gather(*tasks, return_exceptionsTrue) merged [] for r in results: if isinstance(r, Exception): merged.append(f[执行失败] {type(r).__name__}: {r}) else: merged.append(r) return \n\n.join(merged)这段编排的关键设计有三点。第一route_question是同步调用因为分流只需要一次模型请求没必要异步化增加复杂度。第二asyncio.gather带return_exceptionsTrue单个 Agent 失败不会中断其他 Agent失败信息会作为结果的一部分返回方便定位是哪个 Agent 出了问题。第三每个 Agent 的提示词里明确限定职责范围防止订单 Agent 越界回答退款问题。如果你用 LangGraph 实现Router 对应的是条件边conditional edge。核心结构是入口节点做分流根据分流结果路由到不同 Agent 节点最后汇聚到合成节点。LangGraph 的配置片段如下from langgraph.graph import StateGraph, END from typing import TypedDict, List class GraphState(TypedDict): question: str agents: List[str] subtasks: List[str] results: List[str] answer: str def router_node(state: GraphState) - GraphState: decision route_question(state[question]) return {**state, agents: decision.agents, subtasks: decision.subtasks} def make_agent_node(agent_name: str): def node(state: GraphState) - GraphState: idx state[agents].index(agent_name) subtask state[subtasks][idx] llm AGENT_LLMS[agent_name] resp llm.invoke([ SystemMessage(contentAGENT_PROMPTS[agent_name]), HumanMessage(contentsubtask), ]) return {**state, results: state[results] [resp.content]} return node def route_selector(state: GraphState) - List[str]: return state[agents] builder StateGraph(GraphState) builder.add_node(router, router_node) for name in AGENT_PROMPTS: builder.add_node(name, make_agent_node(name)) builder.set_entry_point(router) builder.add_conditional_edges(router, route_selector, {n: n for n in AGENT_PROMPTS}) for name in AGENT_PROMPTS: builder.add_edge(name, END) graph builder.compile()LangGraph 版本的优势是分流路径可视化调试时能看到请求走了哪条边。注意add_conditional_edges的映射表必须覆盖所有可能的 Agent 名称漏写会导致运行时找不到目标节点。4. 验证请求与成功结果从单 Agent 到完整链路配置写完后必须逐层验证不要一上来就跑完整链路。我的习惯是先验证统一通道连通性再验证 Router 分流最后验证并行汇总。这样出错时能快速定位是哪一层的问题。第一步验证 Base URL 和 Key 是否可用。用 curl 发一个最小请求curl -s 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: 回复 OK 两个字母}], max_tokens: 10 }成功时返回的 JSON 里choices[0].message.content应该是OK或类似内容。如果返回 401说明 Key 有问题如果返回 404检查 Base URL 是否漏了/api后缀如果返回model not found说明 Model ID 写错了去控制台确认可用模型列表。第二步验证 Router 分流是否准确。单独调用route_question打印结构化结果decision route_question(我上周买的耳机还没发货另外优惠券能退吗) print(decision.agents) print(decision.subtasks)预期输出类似[order_agent, refund_agent]子任务分别是查物流和退优惠券。如果 Router 只返回了一个 Agent说明提示词里的 Agent 描述不够清晰或者模型能力不足可以换更强的模型做分流。这一步的验证标准是多意图问题必须分流到多个 Agent单意图问题不能过度分流。第三步跑完整链路并检查汇总结果result asyncio.run(run_router_flow(我上周买的耳机还没发货另外优惠券能退吗)) print(result)成功的结果应该包含两段带 Agent 标签的内容分别回答物流和退款问题。如果某段显示[执行失败]根据异常类型排查TimeoutError是超时调大timeout参数AuthenticationError是 Key 问题RateLimitError是并发太高降低并行数或加重试。第四步如果你用 LangGraph验证图结构是否正确out graph.invoke({question: 订单没发货优惠券能退吗, agents: [], subtasks: [], results: [], answer: }) print(out[results])LangGraph 的常见问题是条件边映射不全报错信息通常是Node not found或Invalid edge。对照add_conditional_edges的映射表确认每个 Agent 名称都有对应节点。验证通过后建议把这条链路固化成冒烟测试每次改配置后跑一遍。多 Agent 系统的配置项多人工回归容易漏一个 30 秒的脚本能省掉很多排查时间。模型对话页面可以用来快速对比不同模型的分流效果地址是https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentchat把同样的多意图问题丢进去看不同模型返回的分流 JSON 是否稳定。5. 常见错误排查401、local proxy failed 与 reading choices多 Agent 接入统一通道时报错集中在几个固定位置。这一节按真实报错信息对照排查每条都给出触发原因和修复动作。401 Unauthorized。最常见的原因是 Key 没读到。检查三处.env文件是否被load_dotenv()正确加载环境变量名是否和代码里一致Key 是否带了多余空格。有个隐蔽情况是 Key 复制时带了换行符os.getenv读出来末尾有\n请求头里就变成非法字符。修复方式是在读取后加.strip()。另一个原因是 Key 被删除或过期去 API Keys 页面确认状态。local proxy failed / connection refused。这个报错通常出现在客户端配置了本地代理但代理没启动。如果你在 Claude Code 或 Cline 里看到这个检查 settings 里是否残留了HTTP_PROXY或HTTPS_PROXY环境变量。统一通道是直连的不需要额外代理配置把相关环境变量清掉即可。还有一种情况是 Base URL 写成了https://taotoken.net/api/带尾斜杠部分客户端拼接路径时会产生双斜杠导致连接异常去掉尾斜杠。reading choices of undefined。这个报错说明响应体里没有choices字段通常是请求根本没成功但代码没检查状态码。触发原因有三类Base URL 写错导致返回了 HTML 错误页Model ID 不存在导致返回错误 JSON请求体格式不对被服务端拒绝。修复方式是先在代码里打印完整响应resp llm.invoke(messages) print(resp.response_metadata)response_metadata里能看到实际的状态码和原始响应。确认是 404 就改 Base URL是 400 就检查请求体字段是 401 就回到上一条排查 Key。OAuth 相关报错。如果你在 Claude Code 里看到 OAuth 认证失败说明客户端走了 OAuth 流程而不是 API Key 流程。统一通道用的是 Key 认证需要在 settings 里显式配置ANTHROPIC_AUTH_TOKEN并且不要同时配置 OAuth 相关字段。两者冲突时客户端优先走 OAuth导致 Key 不生效。清理掉 OAuth 配置只保留 Base URL、Key、Model ID 三件套。结构化输出解析失败。Router 用with_structured_output时偶尔报 JSON 解析错误原因是模型在 JSON 前后加了说明文字。修复方式是在提示词末尾强调「只输出 JSON」并在代码里加一次重试from langchain_core.exceptions import OutputParserException def route_with_retry(question: str, retries: int 2) - RouteDecision: for i in range(retries): try: return route_question(question) except OutputParserException: if i retries - 1: raise raise RuntimeError(unreachable)并行调用部分失败。asyncio.gather带return_exceptionsTrue时失败信息混在结果里容易被忽略。建议在汇总前先过滤异常并打日志明确哪个 Agent 失败、失败原因是什么。如果失败集中在某个 Agent检查该 Agent 的模型 ID 是否可用、提示词是否触发了内容过滤。排查顺序建议固定为先 curl 验证通道再单测 Router再单测单个 Agent最后跑完整链路。这个顺序能把问题范围逐层缩小避免在完整链路里大海捞针。接入文档里有各客户端完整的字段对照表遇到字段名不确定时优先查文档而不是猜。6. 长期编码与 Agent 项目落地建议多 Agent 项目从 demo 到稳定运行差距主要在配置管理和模型选型上。Router 用便宜模型做分流、业务 Agent 用强模型做执行这个分层能显著控制成本。统一通道的价值在于模型 ID 可以随时替换你今天用 A 模型做分流效果不好改成 B 模型只需要改一个环境变量Base URL 和 Key 不动。如果你打算长期做 Agent 开发建议把 Coding Plan 用起来地址是https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-plan。它适合需要持续调用模型做编码辅助和 Agent 调试的场景配额和通道统一管理不用每次调试都担心额度。配合 API Keys 页面做密钥轮换开发环境和生产环境用不同的 Key出问题能快速隔离。最后给一个实操建议把 Router 的分流结果落日志记录每次请求的agents、subtasks、reason和耗时。跑一周后回看日志你会发现哪些意图经常被误分流针对性优化提示词比盲目换模型有效得多。多 Agent 系统的调优是数据驱动的没有日志就只能凭感觉。
