LangChain 中文学习手册

Appearance

路由器：知识库 ​

概述 ​

路由器模式是一种多代理架构，其中路由步骤对输入进行分类并将其引导至专门的代理，然后将结果综合为统一的响应。当组织的知识存储在不同的垂直领域（各自需要具有专门工具和提示词的独立代理的知识域）中时，这种模式尤为出色。

在本教程中，您将构建一个多源知识库路由器，通过一个真实的企业场景展示这些优势。该系统将协调三个专家代理：

  • GitHub 代理 — 搜索代码、问题和拉取请求。
  • Notion 代理 — 搜索内部文档和维基。
  • Slack 代理 — 搜索相关的线程和讨论。

当用户问"如何认证 API 请求？"时，路由器将查询分解为特定于源的子问题，并行路由到相关代理，并将结果综合为连贯的答案。

mermaid
graph LR
    A([查询]) --> B[分类]
    B --> C[GitHub 代理]
    B --> D[Notion 代理]
    B --> E[Slack 代理]
    C --> F[综合]
    D --> F
    E --> F
    F --> G([组合答案])

为什么要使用路由器？ ​

路由器模式提供了几个优势：

  • 并行执行：同时查询多个源，减少延迟。
  • 专门化代理：每个垂直领域都有聚焦的工具和针对其领域优化的提示词。
  • 选择性路由：不是每个查询都需要所有源——路由器智能地选择相关的垂直领域。
  • 定向子问题：每个代理收到针对其领域量身定制的问题，提高结果质量。
  • 清晰的综合：将来自多个源的结果合并为单一、连贯的响应。

概念 ​

我们将涵盖以下概念：

路由器 vs. 子代理：子代理模式也可以路由到多个代理。当您需要专门的预处理、自定义路由逻辑或想要显式控制并行执行时，使用路由器模式。当您希望 LLM 动态决定调用哪些代理时，使用子代理模式。

设置 ​

安装 ​

本教程需要 langchain 和 langgraph 包：

bash
pip install langchain langgraph

更多详情请参阅我们的安装指南。

LangSmith ​

设置 LangSmith：

python
import getpass
import os

os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = getpass.getpass()

选择模型 ​

选择一个聊天模型：

python
from langchain.chat_models import init_chat_model

# OpenAI
model = init_chat_model("openai:gpt-5.4")

# Anthropic
model = init_chat_model("anthropic:claude-sonnet-4-6")

# Google Gemini
model = init_chat_model("google_genai:gemini-2.5-flash-lite")

1. 定义状态 ​

首先定义状态模式。我们使用三种类型：

  • AgentInput：传递给每个子代理的简单状态（仅查询）
  • AgentOutput：每个子代理返回的结果（源名称 + 结果）
  • RouterState：主工作流状态，跟踪查询、分类、结果和最终答案
python
from typing import Annotated, Literal, TypedDict
import operator


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]  # reducer 收集并行结果
    final_answer: str

results 字段使用 reducer（Python 中的 operator.add）来收集来自并行代理执行的结果到单个列表中。

2. 为每个垂直领域定义工具 ​

为每个知识领域创建工具。在生产系统中，这些会调用实际的 API。在本教程中，我们使用桩代码实现返回模拟数据。我们在 3 个垂直领域中定义了 7 个工具：GitHub（搜索代码、问题、PR）、Notion（搜索文档、获取页面）、Slack（搜索消息、获取线程）。

python
from langchain.tools import tool


@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 问题和拉取请求。"""
    return f"找到 3 个匹配 '{query}' 的问题：#142（API 认证文档）、#89（OAuth 流程）、#203（令牌刷新）"


@tool
def search_prs(query: str) -> str:
    """搜索拉取请求以获取实现细节。"""
    return "PR #156 添加了 JWT 认证，PR #178 更新了 OAuth 作用域"


@tool
def search_notion(query: str) -> str:
    """在 Notion 工作空间中搜索文档。"""
    return "找到文档：'API 认证指南' - 涵盖 OAuth2 流程、API 密钥和 JWT 令牌"


@tool
def get_page(page_id: str) -> str:
    """通过 ID 获取特定的 Notion 页面。"""
    return "页面内容：逐步认证设置说明"


@tool
def search_slack(query: str) -> str:
    """搜索 Slack 消息和线程。"""
    return "在 #engineering 中找到讨论：'使用 Bearer 令牌进行 API 认证，参见刷新流程文档'"


@tool
def get_thread(thread_id: str) -> str:
    """获取特定的 Slack 线程。"""
    return "线程讨论了 API 密钥轮换的最佳实践"

3. 创建专门化的代理 ​

为每个垂直领域创建一个代理。每个代理都有领域特定的工具和针对其知识源优化的提示词。三者遵循相同的模式——只有工具和系统提示不同。

python
from langchain.agents import create_agent

github_agent = create_agent(
    model,
    tools=[search_code, search_issues, search_prs],
    system_prompt=(
        "你是一个 GitHub 专家。通过搜索仓库、"
        "问题和拉取请求来回答关于代码、"
        "API 参考和实现细节的问题。"
    ),
)

notion_agent = create_agent(
    model,
    tools=[search_notion, get_page],
    system_prompt=(
        "你是一个 Notion 专家。通过搜索组织的"
        "Notion 工作空间来回答关于内部流程、"
        "政策和团队文档的问题。"
    ),
)

slack_agent = create_agent(
    model,
    tools=[search_slack, get_thread],
    system_prompt=(
        "你是一个 Slack 专家。通过搜索团队成员"
        "分享知识和解决方案的相关线程和讨论"
        "来回答问题。"
    ),
)

4. 构建路由器工作流 ​

现在使用 StateGraph 构建路由器工作流。工作流有四个主要步骤：

  1. 分类：分析查询并决定调用哪些代理以及使用什么子问题。
  2. 路由：使用 Send 并行分发到选定的代理。
  3. 查询代理：每个代理接收一个简单的 AgentInput 并返回一个 AgentOutput。
  4. 综合：将收集的结果合并为连贯的响应。
python
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send

router_llm = init_chat_model("openai:gpt-5.4-mini")


# 为分类器定义结构化输出模式
class ClassificationResult(BaseModel):
    """将用户查询分类为代理特定的子问题的结果。"""
    classifications: list[Classification] = Field(
        description="要调用的代理列表及其定向的子问题"
    )


def classify_query(state: RouterState) -> dict:
    """分类查询并确定要调用哪些代理。"""
    structured_llm = router_llm.with_structured_output(ClassificationResult)

    result = structured_llm.invoke([
        {
            "role": "system",
            "content": """分析此查询并确定要查阅哪些知识库。
为每个相关的源生成一个针对该源优化的定向子问题。

可用源：
- github：代码、API 参考、实现细节、问题、拉取请求
- notion：内部文档、流程、策略、团队维基
- slack：团队讨论、非正式知识分享、近期对话

只返回与查询相关的源。每个源应有一个针对该特定知识领域优化的定向子问题。

例如，对于"如何认证 API 请求？"：
- github："存在哪些认证代码？搜索 auth 中间件、JWT 处理"
- notion："存在哪些认证文档？查找 API 认证指南"
（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:
    """查询 GitHub 代理。"""
    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:
    """查询 Notion 代理。"""
    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:
    """查询 Slack 代理。"""
    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:
    """将来自所有代理的结果综合为单个连贯的答案。"""
    synthesizer = init_chat_model("openai:gpt-5.4")

    formatted_results = "\n\n".join(
        f"## {r['source'].upper()} 的结果\n{r['result']}"
        for r in state["results"]
    )

    response = synthesizer.invoke([
        {
            "role": "system",
            "content": f"""你是一个知识综合助手。将来自多个来源的信息合并为
一个连贯的答案。原始用户查询是：{state['query']}

以下是来自不同来源的结果：
{formatted_results}

提供一个综合性的答案，清晰说明来自每个源的信息。"""
        },
    ])

    return {"final_answer": response.content}

5. 组装图 ​

现在将所有步骤连接成一个工作流图：

python
# 创建图
builder = StateGraph(RouterState)

# 添加节点
builder.add_node("classify", classify_query)
builder.add_node("github", query_github)
builder.add_node("notion", query_notion)
builder.add_node("slack", query_slack)
builder.add_node("synthesize", synthesize_results)

# 添加边
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route_to_agents, ["github", "notion", "slack"])
builder.add_edge("github", "synthesize")
builder.add_edge("notion", "synthesize")
builder.add_edge("slack", "synthesize")
builder.add_edge("synthesize", END)

# 编译图
router_app = builder.compile()

6. 运行路由器 ​

测试完整的路由器工作流：

python
query = "我们如何认证 API 请求？请在代码、文档和团队讨论中查找"

for chunk in router_app.stream({"query": query}):
    for node_name, output in chunk.items():
        if node_name == "classify":
            print(f"\n=== 分类阶段 ===")
            print(f"查询: {query}")
            print(f"分类: {output}")
        elif node_name in ("github", "notion", "slack"):
            print(f"\n=== {node_name} 代理 ===")
            print(output)
        elif node_name == "synthesize":
            print(f"\n=== 综合答案 ===")
            print(output["final_answer"])

路由器会将查询分解为源特定的子问题，并行查询相关代理，然后综合结果。例如：

  • GitHub 代理：搜索认证中间件的代码、相关的 PR 和 issue
  • Notion 代理：查找 API 认证指南的内部文档
  • Slack 代理：搜索关于令牌轮换和认证最佳实践的讨论

最终的综合答案将结合来自所有三个源的信息，提供全面而连贯的回复。

概念回顾 ​

在本教程中，您学到了：

  • 路由器模式 — 分类查询并并行路由到专门化的代理
  • 结构化输出 — 使用 Pydantic 的 with_structured_output 获取可靠的路由决策
  • Send API — 使用 LangGraph 的 Send 进行循环内并行扇出
  • 综合 — 将来自多个专用代理的结果合并为连贯的答案
  • StateGraph — 具有 reducer 的工作流编排，用于收集并行结果

下一步 ​

本站为非官方中文学习站点，不代表 LangChain 官方。部分内容参考官方文档并重新整理为中文学习笔记。

Copyright © 2026 LangChain 中文学习手册