1. 为什么 Router 场景下要统一模型通道LangChain 的多智能体 Router 架构本质上是让一个分类器先判断用户问题该交给哪些垂直领域的子智能体再并行分发、最后合成答案。这套流程里模型调用点比普通单链应用多得多分类器要调一次结构化输出模型每个子智能体内部可能还要多轮工具调用合成阶段又要再调一次。如果每个节点各自配置不同的 Key、不同的 base_url本地调试时你会陷入「到底哪个节点超时了」的泥潭。我试过在一个 GitHub Notion Slack 三源知识库 Router 里把分类器、三个子智能体、合成器分别指向不同供应商结果一次查询里出现了三种不同的限流报错排查成本极高。后来改成统一走 TaoToken 的 OpenAI 兼容通道所有节点共用一个 Key 和一个 base_url问题立刻收敛成「一个通道是否可用」这一件事。TaoToken 在这里扮演的角色是「统一模型出口」它提供 OpenAI 兼容的/v1/chat/completions接口LangChain 的ChatOpenAI只要改base_url和api_key就能接上不需要改任何 Router 的图结构。适合谁适合正在本地跑多智能体项目、想让 Router 链路稳定可复现、又不想在多个供应商之间来回切换的开发者。这篇会给你一份可直接复制的settings.json骨架把 Router 里所有模型调用点收敛到 TaoToken然后跑通一次「分类 → 并行分发 → 合成」的完整验证。2. TaoToken 前置准备Key 与通道确认在写配置之前先把通道准备好。你需要一个 TaoToken 的 API Key以及确认要用的模型名。整个 Router 里我会用两个模型一个便宜快速的做分类器比如gpt-4.1-mini这类一个能力更强的做子智能体和合成比如gpt-4.1。你也可以全用一个模型配置更简单。获取 Key 的入口在控制台的 API Keys 页面登录后新建一个即可。拿到形如sk-...的字符串后先别急着写进代码用一条 curl 确认通道本身是通的curl https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -H Content-Type: application/json \ -d { model: gpt-4.1-mini, messages: [{role: user, content: ping}], max_tokens: 16 }返回里能看到choices[0].message.content就说明通道正常。这一步很关键因为后面 Router 报错时你要能区分是「通道不通」还是「图配置错」。注意base_url 用https://taotoken.net/apiLangChain 的ChatOpenAI会自动拼接/v1/chat/completions所以你在配置里写https://taotoken.net/api即可不要重复加/v1。如果你更想先在网页里验证模型是否可用可以直接打开模型对话页面发一条消息确认返回正常后再进入配置环节。3. settings.json 骨架把 Router 所有模型调用点收敛本地多智能体项目我习惯用一个settings.json集中管理模型通道代码里只读配置不硬编码。下面这份骨架覆盖了 Router 的三个调用层router分类器、agents子智能体、synthesis合成器它们共用同一个provider块。{ provider: { name: taotoken, base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, timeout: 60, max_retries: 2 }, models: { router: { model: gpt-4.1-mini, temperature: 0, max_tokens: 512 }, agents: { model: gpt-4.1, temperature: 0.2, max_tokens: 1024 }, synthesis: { model: gpt-4.1, temperature: 0.3, max_tokens: 1500 } }, router: { sources: [github, notion, slack], parallel: true, structured_output: true } }几个设计点说明一下。api_key_env指向环境变量而不是明文写 Key避免误提交。router.temperature设成 0因为分类需要稳定可复现同样的查询应该路由到同样的智能体。agents和synthesis温度略高让子智能体的检索式回答和最终合成更自然。router.parallel对应 LangGraph 里用Send做 fan-outstructured_output对应分类器用 Pydantic 模型约束输出。读取这份配置的代码可以这样写import json import os from langchain_openai import ChatOpenAI with open(settings.json, r, encodingutf-8) as f: cfg json.load(f) provider cfg[provider] api_key os.environ[provider[api_key_env]] def build_model(section: str) - ChatOpenAI: m cfg[models][section] return ChatOpenAI( modelm[model], temperaturem[temperature], max_tokensm[max_tokens], base_urlprovider[base_url], api_keyapi_key, timeoutprovider[timeout], max_retriesprovider[max_retries], ) router_llm build_model(router) agent_llm build_model(agents) synth_llm build_model(synthesis)这样 Router 图里所有节点都从这三个实例取模型通道统一改一处配置全链路生效。4. 可复制配置把 Router 图接到统一通道有了模型实例接下来把 Router 的图搭起来。核心是分类节点用router_llm做结构化输出子智能体节点用agent_llm合成节点用synth_llm。下面这段可以直接跑工具用模拟实现你替换成真实 API 即可。import operator from typing import Annotated, Literal, TypedDict from langchain.agents import create_agent from langchain.tools import tool from langgraph.graph import StateGraph, START, END from langgraph.types import Send from pydantic import BaseModel, Field class AgentInput(TypedDict): query: str class AgentOutput(TypedDict): source: str result: str class Classification(TypedDict): source: Literal[github, notion, slack] query: str class RouterState(TypedDict): query: str classifications: list[Classification] results: Annotated[list[AgentOutput], operator.add] final_answer: str class ClassificationResult(BaseModel): classifications: list[Classification] Field( description需要调用的智能体列表每个附带一个针对性的子问题 ) tool def search_code(query: str, repo: str main) - str: 在 GitHub 仓库中搜索代码。 return f在 {repo} 中找到与 {query} 匹配的代码src/auth.py 中的身份验证中间件 tool def search_issues(query: str) - str: 搜索 GitHub issue 和 PR。 return f找到 3 个与 {query} 匹配的 issue#142、#89、#203 tool def search_notion(query: str) - str: 在 Notion 工作区中搜索文档。 return f找到文档《API 身份验证指南》——涵盖 OAuth2、API 密钥和 JWT tool def search_slack(query: str) - str: 搜索 Slack 消息和讨论线程。 return f在 #engineering 发现讨论API 认证请使用 Bearer 令牌 github_agent create_agent( agent_llm, tools[search_code, search_issues], system_prompt你是 GitHub 专家回答代码、API 文档和实现细节问题。, ) notion_agent create_agent( agent_llm, tools[search_notion], system_prompt你是 Notion 专家回答内部流程、政策和团队文档问题。, ) slack_agent create_agent( agent_llm, tools[search_slack], system_prompt你是 Slack 专家回答团队讨论和非正式知识分享问题。, ) def classify_query(state: RouterState) - dict: structured_llm router_llm.with_structured_output(ClassificationResult) result structured_llm.invoke([ {role: system, content: ( 分析查询判断应咨询哪些知识源并为每个源生成优化过的子问题。 可用源github代码/issue/PR、notion文档/流程、slack讨论。 仅返回相关源。 )}, {role: user, content: state[query]}, ]) return {classifications: result.classifications} def route_to_agents(state: RouterState) - list[Send]: return [Send(c[source], {query: c[query]}) for c in state[classifications]] def query_github(state: AgentInput) - dict: r github_agent.invoke({messages: [{role: user, content: state[query]}]}) return {results: [{source: github, result: r[messages][-1].content}]} def query_notion(state: AgentInput) - dict: r notion_agent.invoke({messages: [{role: user, content: state[query]}]}) return {results: [{source: notion, result: r[messages][-1].content}]} def query_slack(state: AgentInput) - dict: r slack_agent.invoke({messages: [{role: user, content: state[query]}]}) return {results: [{source: slack, result: r[messages][-1].content}]} def synthesize_results(state: RouterState) - dict: if not state[results]: return {final_answer: 未从任何知识源找到结果。} formatted [f**来自 {r[source].title()}**\n{r[result]} for r in state[results]] resp synth_llm.invoke([ {role: system, content: f综合以下结果回答{state[query]}融合多源、避免重复、条理清晰。}, {role: user, content: \n\n.join(formatted)}, ]) return {final_answer: resp.content} workflow ( StateGraph(RouterState) .add_node(classify, classify_query) .add_node(github, query_github) .add_node(notion, query_notion) .add_node(slack, query_slack) .add_node(synthesize, synthesize_results) .add_edge(START, classify) .add_conditional_edges(classify, route_to_agents, [github, notion, slack]) .add_edge(github, synthesize) .add_edge(notion, synthesize) .add_edge(slack, synthesize) .add_edge(synthesize, END) .compile() )注意add_conditional_edges的第三个参数列出了所有可能的目标节点route_to_agents返回的Send列表决定实际并行执行哪几个。这就是 Router 的 fan-out 核心。5. 验证请求跑通一次多智能体分发配置写好后用一条跨领域查询验证整条链路。查询「如何对 API 请求进行身份验证」应该同时命中 GitHub 和 NotionSlack 可能被跳过。if __name__ __main__: result workflow.invoke({query: 如何对 API 请求进行身份验证}) print(原始查询:, result[query]) print(\n路由分类结果:) for c in result[classifications]: print(f {c[source]}: {c[query]}) print(\n最终回答:) print(result[final_answer])预期输出里分类结果应该出现github和notion两条各自带一个针对该源优化的子问题比如 github 那条是「搜索 auth 中间件、JWT 处理」notion 那条是「查找 API 认证指南」。最终回答会把两个源的结果融合成一段带编号的说明。如果分类结果为空说明分类器没解析出任何源检查structured_output是否生效。如果只有一条说明分类器判断该问题只涉及一个领域可以换一条更跨领域的查询再试比如「API 认证的代码实现和内部文档分别在哪」。验证通过后你还可以打开模型对话页面用同样的查询对比单模型直答和 Router 合成的差异直观感受多源融合的价值。6. 本篇常见错排查报错一AuthenticationError或 401。先确认TAOTOKEN_API_KEY环境变量已导出再确认base_url是https://taotoken.net/api而不是带/v1的完整路径。LangChain 的ChatOpenAI会自己拼/v1/chat/completions重复拼接会导致 404 或 401。报错二分类器返回空列表。多半是with_structured_output没拿到合法 JSON。把router.temperature设为 0并在 system prompt 里明确「仅返回相关源不要编造」。如果还不行检查ClassificationResult的字段名和Classification是否一致。报错三并行节点只跑了一个。检查add_conditional_edges的第三个参数是否列出了全部三个目标节点。如果只列了两个第三个Send会被丢弃。另外确认route_to_agents返回的是list[Send]而不是单个Send。报错四合成阶段拿不到结果。results字段必须用Annotated[list[AgentOutput], operator.add]做 reducer否则并行分支的返回值会互相覆盖最后只剩一个。这是 Router 并行收集结果最容易踩的坑。报错五超时。三个子智能体并行时如果某个源响应慢整体会被拖住。在settings.json里把timeout调到 60 秒max_retries设 2给慢源留出重试空间。如果某个源长期慢考虑在分类阶段就把它排除。排障时如果怀疑是 Key 或通道问题直接去 API Keys 页面重新生成一个 Key 对比测试接入细节可以对照接入文档逐项核对 base_url 和请求头格式。7. 下一步把 Router 用到长期编码与 Agent 场景跑通这次验证后你会发现 Router 的价值在于「按领域分流 并行 合成」这套结构同样适合长期运行的编码助手和 Agent 工作流。如果你打算把 Router 接到日常编码任务里比如让不同子智能体分别处理代码检索、文档查询、历史讨论可以考虑用 Coding Plan 把模型调用额度固定下来避免调试期频繁触发限流。配置层面你只需要维护好那份settings.json新增子智能体时在models.agents下加一个 section在router.sources里加一个源名图里加一个节点和一条边即可。通道始终是同一个Key 始终是同一个排查范围始终收敛在一处。这就是统一模型出口对多智能体项目最实际的意义。
LangChain多智能体Router配置TaoToken：settings.json骨架与路由验证
1. 为什么 Router 场景下要统一模型通道LangChain 的多智能体 Router 架构本质上是让一个分类器先判断用户问题该交给哪些垂直领域的子智能体再并行分发、最后合成答案。这套流程里模型调用点比普通单链应用多得多分类器要调一次结构化输出模型每个子智能体内部可能还要多轮工具调用合成阶段又要再调一次。如果每个节点各自配置不同的 Key、不同的 base_url本地调试时你会陷入「到底哪个节点超时了」的泥潭。我试过在一个 GitHub Notion Slack 三源知识库 Router 里把分类器、三个子智能体、合成器分别指向不同供应商结果一次查询里出现了三种不同的限流报错排查成本极高。后来改成统一走 TaoToken 的 OpenAI 兼容通道所有节点共用一个 Key 和一个 base_url问题立刻收敛成「一个通道是否可用」这一件事。TaoToken 在这里扮演的角色是「统一模型出口」它提供 OpenAI 兼容的/v1/chat/completions接口LangChain 的ChatOpenAI只要改base_url和api_key就能接上不需要改任何 Router 的图结构。适合谁适合正在本地跑多智能体项目、想让 Router 链路稳定可复现、又不想在多个供应商之间来回切换的开发者。这篇会给你一份可直接复制的settings.json骨架把 Router 里所有模型调用点收敛到 TaoToken然后跑通一次「分类 → 并行分发 → 合成」的完整验证。2. TaoToken 前置准备Key 与通道确认在写配置之前先把通道准备好。你需要一个 TaoToken 的 API Key以及确认要用的模型名。整个 Router 里我会用两个模型一个便宜快速的做分类器比如gpt-4.1-mini这类一个能力更强的做子智能体和合成比如gpt-4.1。你也可以全用一个模型配置更简单。获取 Key 的入口在控制台的 API Keys 页面登录后新建一个即可。拿到形如sk-...的字符串后先别急着写进代码用一条 curl 确认通道本身是通的curl https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -H Content-Type: application/json \ -d { model: gpt-4.1-mini, messages: [{role: user, content: ping}], max_tokens: 16 }返回里能看到choices[0].message.content就说明通道正常。这一步很关键因为后面 Router 报错时你要能区分是「通道不通」还是「图配置错」。注意base_url 用https://taotoken.net/apiLangChain 的ChatOpenAI会自动拼接/v1/chat/completions所以你在配置里写https://taotoken.net/api即可不要重复加/v1。如果你更想先在网页里验证模型是否可用可以直接打开模型对话页面发一条消息确认返回正常后再进入配置环节。3. settings.json 骨架把 Router 所有模型调用点收敛本地多智能体项目我习惯用一个settings.json集中管理模型通道代码里只读配置不硬编码。下面这份骨架覆盖了 Router 的三个调用层router分类器、agents子智能体、synthesis合成器它们共用同一个provider块。{ provider: { name: taotoken, base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, timeout: 60, max_retries: 2 }, models: { router: { model: gpt-4.1-mini, temperature: 0, max_tokens: 512 }, agents: { model: gpt-4.1, temperature: 0.2, max_tokens: 1024 }, synthesis: { model: gpt-4.1, temperature: 0.3, max_tokens: 1500 } }, router: { sources: [github, notion, slack], parallel: true, structured_output: true } }几个设计点说明一下。api_key_env指向环境变量而不是明文写 Key避免误提交。router.temperature设成 0因为分类需要稳定可复现同样的查询应该路由到同样的智能体。agents和synthesis温度略高让子智能体的检索式回答和最终合成更自然。router.parallel对应 LangGraph 里用Send做 fan-outstructured_output对应分类器用 Pydantic 模型约束输出。读取这份配置的代码可以这样写import json import os from langchain_openai import ChatOpenAI with open(settings.json, r, encodingutf-8) as f: cfg json.load(f) provider cfg[provider] api_key os.environ[provider[api_key_env]] def build_model(section: str) - ChatOpenAI: m cfg[models][section] return ChatOpenAI( modelm[model], temperaturem[temperature], max_tokensm[max_tokens], base_urlprovider[base_url], api_keyapi_key, timeoutprovider[timeout], max_retriesprovider[max_retries], ) router_llm build_model(router) agent_llm build_model(agents) synth_llm build_model(synthesis)这样 Router 图里所有节点都从这三个实例取模型通道统一改一处配置全链路生效。4. 可复制配置把 Router 图接到统一通道有了模型实例接下来把 Router 的图搭起来。核心是分类节点用router_llm做结构化输出子智能体节点用agent_llm合成节点用synth_llm。下面这段可以直接跑工具用模拟实现你替换成真实 API 即可。import operator from typing import Annotated, Literal, TypedDict from langchain.agents import create_agent from langchain.tools import tool from langgraph.graph import StateGraph, START, END from langgraph.types import Send from pydantic import BaseModel, Field class AgentInput(TypedDict): query: str class AgentOutput(TypedDict): source: str result: str class Classification(TypedDict): source: Literal[github, notion, slack] query: str class RouterState(TypedDict): query: str classifications: list[Classification] results: Annotated[list[AgentOutput], operator.add] final_answer: str class ClassificationResult(BaseModel): classifications: list[Classification] Field( description需要调用的智能体列表每个附带一个针对性的子问题 ) tool def search_code(query: str, repo: str main) - str: 在 GitHub 仓库中搜索代码。 return f在 {repo} 中找到与 {query} 匹配的代码src/auth.py 中的身份验证中间件 tool def search_issues(query: str) - str: 搜索 GitHub issue 和 PR。 return f找到 3 个与 {query} 匹配的 issue#142、#89、#203 tool def search_notion(query: str) - str: 在 Notion 工作区中搜索文档。 return f找到文档《API 身份验证指南》——涵盖 OAuth2、API 密钥和 JWT tool def search_slack(query: str) - str: 搜索 Slack 消息和讨论线程。 return f在 #engineering 发现讨论API 认证请使用 Bearer 令牌 github_agent create_agent( agent_llm, tools[search_code, search_issues], system_prompt你是 GitHub 专家回答代码、API 文档和实现细节问题。, ) notion_agent create_agent( agent_llm, tools[search_notion], system_prompt你是 Notion 专家回答内部流程、政策和团队文档问题。, ) slack_agent create_agent( agent_llm, tools[search_slack], system_prompt你是 Slack 专家回答团队讨论和非正式知识分享问题。, ) def classify_query(state: RouterState) - dict: structured_llm router_llm.with_structured_output(ClassificationResult) result structured_llm.invoke([ {role: system, content: ( 分析查询判断应咨询哪些知识源并为每个源生成优化过的子问题。 可用源github代码/issue/PR、notion文档/流程、slack讨论。 仅返回相关源。 )}, {role: user, content: state[query]}, ]) return {classifications: result.classifications} def route_to_agents(state: RouterState) - list[Send]: return [Send(c[source], {query: c[query]}) for c in state[classifications]] def query_github(state: AgentInput) - dict: r github_agent.invoke({messages: [{role: user, content: state[query]}]}) return {results: [{source: github, result: r[messages][-1].content}]} def query_notion(state: AgentInput) - dict: r notion_agent.invoke({messages: [{role: user, content: state[query]}]}) return {results: [{source: notion, result: r[messages][-1].content}]} def query_slack(state: AgentInput) - dict: r slack_agent.invoke({messages: [{role: user, content: state[query]}]}) return {results: [{source: slack, result: r[messages][-1].content}]} def synthesize_results(state: RouterState) - dict: if not state[results]: return {final_answer: 未从任何知识源找到结果。} formatted [f**来自 {r[source].title()}**\n{r[result]} for r in state[results]] resp synth_llm.invoke([ {role: system, content: f综合以下结果回答{state[query]}融合多源、避免重复、条理清晰。}, {role: user, content: \n\n.join(formatted)}, ]) return {final_answer: resp.content} workflow ( StateGraph(RouterState) .add_node(classify, classify_query) .add_node(github, query_github) .add_node(notion, query_notion) .add_node(slack, query_slack) .add_node(synthesize, synthesize_results) .add_edge(START, classify) .add_conditional_edges(classify, route_to_agents, [github, notion, slack]) .add_edge(github, synthesize) .add_edge(notion, synthesize) .add_edge(slack, synthesize) .add_edge(synthesize, END) .compile() )注意add_conditional_edges的第三个参数列出了所有可能的目标节点route_to_agents返回的Send列表决定实际并行执行哪几个。这就是 Router 的 fan-out 核心。5. 验证请求跑通一次多智能体分发配置写好后用一条跨领域查询验证整条链路。查询「如何对 API 请求进行身份验证」应该同时命中 GitHub 和 NotionSlack 可能被跳过。if __name__ __main__: result workflow.invoke({query: 如何对 API 请求进行身份验证}) print(原始查询:, result[query]) print(\n路由分类结果:) for c in result[classifications]: print(f {c[source]}: {c[query]}) print(\n最终回答:) print(result[final_answer])预期输出里分类结果应该出现github和notion两条各自带一个针对该源优化的子问题比如 github 那条是「搜索 auth 中间件、JWT 处理」notion 那条是「查找 API 认证指南」。最终回答会把两个源的结果融合成一段带编号的说明。如果分类结果为空说明分类器没解析出任何源检查structured_output是否生效。如果只有一条说明分类器判断该问题只涉及一个领域可以换一条更跨领域的查询再试比如「API 认证的代码实现和内部文档分别在哪」。验证通过后你还可以打开模型对话页面用同样的查询对比单模型直答和 Router 合成的差异直观感受多源融合的价值。6. 本篇常见错排查报错一AuthenticationError或 401。先确认TAOTOKEN_API_KEY环境变量已导出再确认base_url是https://taotoken.net/api而不是带/v1的完整路径。LangChain 的ChatOpenAI会自己拼/v1/chat/completions重复拼接会导致 404 或 401。报错二分类器返回空列表。多半是with_structured_output没拿到合法 JSON。把router.temperature设为 0并在 system prompt 里明确「仅返回相关源不要编造」。如果还不行检查ClassificationResult的字段名和Classification是否一致。报错三并行节点只跑了一个。检查add_conditional_edges的第三个参数是否列出了全部三个目标节点。如果只列了两个第三个Send会被丢弃。另外确认route_to_agents返回的是list[Send]而不是单个Send。报错四合成阶段拿不到结果。results字段必须用Annotated[list[AgentOutput], operator.add]做 reducer否则并行分支的返回值会互相覆盖最后只剩一个。这是 Router 并行收集结果最容易踩的坑。报错五超时。三个子智能体并行时如果某个源响应慢整体会被拖住。在settings.json里把timeout调到 60 秒max_retries设 2给慢源留出重试空间。如果某个源长期慢考虑在分类阶段就把它排除。排障时如果怀疑是 Key 或通道问题直接去 API Keys 页面重新生成一个 Key 对比测试接入细节可以对照接入文档逐项核对 base_url 和请求头格式。7. 下一步把 Router 用到长期编码与 Agent 场景跑通这次验证后你会发现 Router 的价值在于「按领域分流 并行 合成」这套结构同样适合长期运行的编码助手和 Agent 工作流。如果你打算把 Router 接到日常编码任务里比如让不同子智能体分别处理代码检索、文档查询、历史讨论可以考虑用 Coding Plan 把模型调用额度固定下来避免调试期频繁触发限流。配置层面你只需要维护好那份settings.json新增子智能体时在models.agents下加一个 section在router.sources里加一个源名图里加一个节点和一条边即可。通道始终是同一个Key 始终是同一个排查范围始终收敛在一处。这就是统一模型出口对多智能体项目最实际的意义。
