企业官网建设流程全解析

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/api，LangChain 的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-out，structured_output对应分类器用 Pydantic 模型约束输出。

读取这份配置的代码可以这样写：

import json import os from langchain_openai import ChatOpenAI with open("settings.json", "r", encoding="utf-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( model=m["model"], temperature=m["temperature"], max_tokens=m["max_tokens"], base_url=provider["base_url"], api_key=api_key, timeout=provider["timeout"], max_retries=provider["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 和 Notion，Slack 可能被跳过。

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 始终是同一个，排查范围始终收敛在一处。这就是统一模型出口对多智能体项目最实际的意义。