1. 为什么你的多智能体系统需要一个专职 RouterLangGraph Router 是多智能体系统里的“分流器”它不负责干活只负责判断用户问题该交给哪个 Agent然后把结果汇总成统一答案。适合谁适合正在用 Langchain Agent 搭建企业知识问答、代码检索、文档助手并且已经踩过“一个 Agent 塞太多工具导致 Prompt 爆炸”这个坑的开发者。我试过把 GitHub、Notion、Slack 三类工具全塞进一个 Agent结果工具选择准确率肉眼可见地下降调试时根本不知道是哪一步出的错。Router 模式的核心价值在于把“决定去哪查”这件事从 Agent 行为中剥离出来变成一个明确的结构。整个系统分成三层——Router 路由层负责分类和拆解问题Specialized Agents 专用智能体各自持有独立 Prompt 和工具集Synthesize 汇总层把多个结果去重、消解冲突后输出统一视角的答案。Router 有三个本质特征它会拆问题、判问题可以调用 0 个、1 个或多个 Agent结果一定要“合成”而不是简单拼接。不太适合 Router 的场景也要说清楚强多轮对话需要记住“刚刚说过什么”、Agent 之间频繁交接状态的更适合 Subagents 或 Handoffs 模式。Router 更适合知识天然分领域、工具差异明显、以单轮或弱多轮问答为主的场景。下面我按“状态定义 → 工具定义 → Agent 创建 → 路由编排 → 验证排错”的顺序把可复制的骨架交给你。2. TaoToken 前置把模型调用这层先稳住在写 Router 之前先把模型接入这层固定下来否则后面排错时你分不清是路由逻辑错了还是模型调用挂了。TaoToken 提供统一的 API 入口兼容 OpenAI 风格的调用方式模型对话、Coding Plan、API Keys 管理都在一个控制台里完成。对于 Router 这种需要频繁调用分类模型 多个子 Agent 模型的场景统一入口能省掉不少环境变量管理的心智负担。你需要先拿到 API Key然后配置到环境变量里。官网入口在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 基址是 https://taotoken.net/api 注意这个不加 UTM 参数。如果你要长期跑编码类 Agent可以看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 只是想先验证模型通不通用模型对话页面最快https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 。环境变量这样配后面代码里统一读export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api注意不要把 Key 硬编码进代码再提交到仓库用环境变量或 .env 文件并加进 .gitignore。3. 可复制配置State、工具、Agent 与 Router 骨架3.1 定义贯穿全流程的 RouterStateLangGraph 里所有节点共享同一个 State它决定了系统“知道哪些东西”。如果 State 说不清楚说明你还没想清楚系统结构或者在不同阶段隐式依赖了某些信息。对 Router 这种并行结构来说这一步尤其重要。from typing import Annotated, Literal, TypedDict import operator class AgentInput(TypedDict): 每个子智能体的简单输入状态。 query: str class AgentOutput(TypedDict): 每个子智能体的输出。 source: str result: str class Classification(TypedDict): 单条路由决策调用哪个 Agent传什么子问题。 source: Literal[github, notion, slack] query: str class RouterState(TypedDict): query: str classifications: list[Classification] results: Annotated[list[AgentOutput], operator.add] # Reducer 收集并行结果 final_answer: str这里最 LangGraph 的地方是results字段Annotated[list[AgentOutput], operator.add]表达的不是“值”而是规则——多个并行节点都会往 results 里拼接内容Reducer 自动合并。3.2 为每个垂直领域定义工具工具先用假实现跑通流程后续替换成真实实现即可。三个平台各代表工程的一个维度GitHub 回答“在哪、怎么写”Notion 回答“为什么这么做”Slack 回答“什么时候改的”。from langchain.tools import tool tool def search_code(query: str, repo: str main) - str: Search code in GitHub repositories. return fFound code matching {query} in {repo}: authentication middleware in src/auth.py tool def search_issues(query: str) - str: Search GitHub issues and pull requests. return fFound 3 issues matching {query}: #142, #89, #203 tool def search_prs(query: str) - str: Search pull requests for implementation details. return fPR #156 added JWT authentication, PR #178 updated OAuth scopes tool def search_notion(query: str) - str: Search Notion workspace for documentation. return fFound documentation: API Authentication Guide - covers OAuth2 flow tool def get_page(page_id: str) - str: Get a specific Notion page by ID. return Page content: Step-by-step authentication setup instructions tool def search_slack(query: str) - str: Search Slack messages and threads. return fFound discussion in #engineering: Use Bearer tokens for API auth tool def get_thread(thread_id: str) - str: Get a specific Slack thread. return Thread discusses best practices for API key rotation3.3 创建三个专用 Agent每个 Agent 由模型、工具、系统提示词三部分组成。系统提示词不是人设而是视角锁定 能力边界声明 工具优先约束。from langchain.agents import create_agent from langchain_openai import ChatOpenAI import os model ChatOpenAI( api_keyos.environ.get(TAOTOKEN_API_KEY), base_urlos.environ.get(TAOTOKEN_BASE_URL), modelgpt-4o-mini, ) github_agent create_agent( model, tools[search_code, search_issues, search_prs], system_prompt( You are a GitHub expert. Answer questions about code, API references, and implementation details by searching repositories, issues, and pull requests. ), ) notion_agent create_agent( model, tools[search_notion, get_page], system_prompt( You are a Notion expert. Answer questions about internal processes, policies, and team documentation by searching the organizations Notion workspace. ), ) slack_agent create_agent( model, tools[search_slack, get_thread], system_prompt( You are a Slack expert. Answer questions by searching relevant threads and discussions where team members have shared knowledge and solutions. ), )3.4 分类器与结构化输出Router 本质是做分类 拆解。用.with_structured_output让模型按我们定义的结构输出避免解析自由文本。from pydantic import BaseModel, Field class ClassificationResult(BaseModel): 把用户问题分类到各 Agent 的子问题列表。 classifications: list[Classification] Field( descriptionList of agents to invoke with their targeted sub-questions ) def classify_query(state: RouterState) - dict: structured_llm model.with_structured_output(ClassificationResult) result structured_llm.invoke([ { role: system, content: Analyze this query and determine which knowledge bases to consult. For each relevant source, generate a targeted sub-question optimized for that source. Available sources: - github: Code, API references, implementation details, issues, pull requests - notion: Internal documentation, processes, policies, team wikis - slack: Team discussions, informal knowledge sharing, recent conversations Return ONLY the sources that are relevant to the query. }, {role: user, content: state[query]}, ]) return {classifications: result.classifications}3.5 并行分发与结果汇总Send是 LangGraph 实现 fan-out 的关键它把 state_patch 送到指定节点多个 Send 会并行执行。from langgraph.types import Send 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: result github_agent.invoke({messages: [{role: user, content: state[query]}]}) return {results: [{source: github, result: result[messages][-1].content}]} def query_notion(state: AgentInput) - dict: result notion_agent.invoke({messages: [{role: user, content: state[query]}]}) return {results: [{source: notion, result: result[messages][-1].content}]} def query_slack(state: AgentInput) - dict: result slack_agent.invoke({messages: [{role: user, content: state[query]}]}) return {results: [{source: slack, result: result[messages][-1].content}]} def synthesize_results(state: RouterState) - dict: if not state[results]: return {final_answer: No results found from any knowledge source.} formatted [f**From {r[source].title()}:**\n{r[result]} for r in state[results]] synthesis_response model.invoke([ { role: system, content: fSynthesize these search results to answer the original question: {state[query]} - Combine information from multiple sources without redundancy - Highlight the most relevant and actionable information - Note any discrepancies between sources - Keep the response concise and well-organized }, {role: user, content: \n\n.join(formatted)}, ]) return {final_answer: synthesis_response.content}3.6 编译工作流from langgraph.graph import StateGraph, START, END 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() )4. 验证请求跑通一次完整路由result workflow.invoke({query: How do I authenticate API requests?}) print(Original query:, result[query]) print(\nClassifications:) for c in result[classifications]: print(f {c[source]}: {c[query]}) print(\nFinal Answer:) print(result[final_answer])预期输出大致是classifications 里出现 github 和 notion 两条slack 被正确省略final_answer 里把 JWT、OAuth2、API Keys 三种方式合并成一段连贯回答并标注了来源。如果 classifications 只出现一条说明分类器把问题判成了单领域这本身不算错但你可以用“同时涉及代码和文档”的问题再测一次确认 fan-out 生效。5. 本篇常见错排查清单5.1 报错InvalidUpdateError: Expected dict, got list原因节点返回的 results 没有用 Reducer 包裹LangGraph 不知道多个并行结果该怎么合并。检查RouterState里 results 是否写成Annotated[list[AgentOutput], operator.add]漏掉operator.add就会报这个。5.2 并行节点只跑了一个原因add_conditional_edges的第三个参数没列全目标节点或者route_to_agents返回的 Send 列表里 source 拼写和节点名不一致。节点名是githubSend 里也必须是小写github大小写不匹配会静默丢弃。5.3 分类器输出解析失败原因模型返回了结构外的字段或者ClassificationResult的字段描述不够清晰。把Field(description...)写具体并在系统提示词里强调“Return ONLY the sources that are relevant”。5.4 final_answer 为空原因synthesize_results里state[results]为空通常是上游 Agent 调用抛异常被吞掉。在query_github等节点里加 try/except 并打印异常先确认单个 Agent 能独立跑通。5.5 模型调用 401 / 连接失败原因TAOTOKEN_API_KEY或TAOTOKEN_BASE_URL没读到。在 Python 里print(os.environ.get(TAOTOKEN_API_KEY))确认非空base_url 结尾不要多加/v1直接用https://taotoken.net/api。Key 管理在控制台https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 接入细节看文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。5.6 想加记忆但上下文爆炸把整个 workflow 包成一个 tool 再挂到带 checkpointer 的 Agent 上即可但只保存输入输出中间子 Agent 的调用结果不要全量存否则上下文很快撑爆。from langgraph.checkpoint.memory import InMemorySaver tool def search_knowledge_base(query: str) - str: Search across multiple knowledge sources (GitHub, Notion, Slack). result workflow.invoke({query: query}) return result[final_answer] conversational_agent create_agent( model, tools[search_knowledge_base], system_promptYou are a helpful assistant. Use search_knowledge_base to find information., checkpointerInMemorySaver(), ) config {configurable: {thread_id: user-123}} r1 conversational_agent.invoke( {messages: [{role: user, content: How do I authenticate API requests?}]}, config) print(r1[messages][-1].content) r2 conversational_agent.invoke( {messages: [{role: user, content: What about rate limiting for those endpoints?}]}, config) print(r2[messages][-1].content)6. 继续往下走把 Router 接进你的真实工程跑通上面这套骨架后下一步是把假工具替换成真实实现并观察分类准确率。我的经验是分类器的系统提示词里 few-shot 示例比长篇规则更有效给一两个“该省略哪个来源”的反例模型判得明显更稳。另外如果你要长期跑编码类 AgentCoding Plan 那条线更适合持续调用https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 只是想快速验证某个模型在分类任务上的表现直接去模型对话页面手动试几轮比改代码快得多https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 。接入过程中遇到 401、超时、结构化输出解析失败先回 API Keys 和文档两处核对https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 、https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。