1. 为什么你的多智能体系统需要一个专职 Router
LangGraph 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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 基址是 https://taotoken.net/api （注意这个不加 UTM 参数）。如果你要长期跑编码类 Agent，可以看 Coding Plan：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite ；只是想先验证模型通不通，用模型对话页面最快：https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=model-chat&utm_campaign=rewrite 。
环境变量这样配，后面代码里统一读：
export TAOTOKEN_API_KEY="sk-你的key" export TAOTOKEN_BASE_URL="https://taotoken.net/api"注意：不要把 Key 硬编码进代码再提交到仓库，用环境变量或 .env 文件并加进 .gitignore。
3. 可复制配置：State、工具、Agent 与 Router 骨架
3.1 定义贯穿全流程的 RouterState
LangGraph 里所有节点共享同一个 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 f"Found 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 f"Found 3 issues matching '{query}': #142, #89, #203" @tool def search_prs(query: str) -> str: """Search pull requests for implementation details.""" return f"PR #156 added JWT authentication, PR #178 updated OAuth scopes" @tool def search_notion(query: str) -> str: """Search Notion workspace for documentation.""" return f"Found 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 f"Found 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 rotation"3.3 创建三个专用 Agent
每个 Agent 由模型、工具、系统提示词三部分组成。系统提示词不是人设，而是视角锁定 + 能力边界声明 + 工具优先约束。
from langchain.agents import create_agent from langchain_openai import ChatOpenAI import os model = ChatOpenAI( api_key=os.environ.get("TAOTOKEN_API_KEY"), base_url=os.environ.get("TAOTOKEN_BASE_URL"), model="gpt-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 organization's 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( description="List 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": f"""Synthesize 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 拼写和节点名不一致。节点名是"github"，Send 里也必须是小写"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_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite ，接入细节看文档：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。
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_prompt="You are a helpful assistant. Use search_knowledge_base to find information.", checkpointer=InMemorySaver(), ) 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 示例比长篇规则更有效，给一两个“该省略哪个来源”的反例，模型判得明显更稳。另外，如果你要长期跑编码类 Agent，Coding Plan 那条线更适合持续调用：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite ；只是想快速验证某个模型在分类任务上的表现，直接去模型对话页面手动试几轮，比改代码快得多：https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=model-chat&utm_campaign=rewrite 。接入过程中遇到 401、超时、结构化输出解析失败，先回 API Keys 和文档两处核对：https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 、https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。
