LangChain 中文文档

主题

构建一个具备路由功能的多源知识库

概述 ​

路由模式是一种多智能体架构，其中路由步骤对输入进行分类并将其定向到专门的智能体，然后将结果合成为一个组合响应。当您组织的知识存在于不同的垂直领域（每个领域都需要自己的、配备专用工具和提示词的智能体）时，这种模式表现出色。

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

  • 一个GitHub智能体，用于搜索代码、问题和拉取请求。
  • 一个Notion智能体，用于搜索内部文档和维基。
  • 一个Slack智能体，用于搜索相关线程和讨论。

当用户询问“如何验证API请求？”时，路由器将查询分解为特定于源的子问题，将它们并行路由到相关智能体，并将结果合成为一个连贯的答案。

mermaid
graph LR
    A([Query]) --> B[Classify]
    B --> C[GitHub agent]
    B --> D[Notion agent]
    B --> E[Slack agent]
    C --> F[Synthesize]
    D --> F
    E --> F
    F --> G([Combined answer])

为什么使用路由器？ ​

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

  • 并行执行：同时查询多个源，与顺序方法相比减少了延迟。
  • 专用智能体：每个垂直领域都有针对其领域优化的专用工具和提示词。
  • 选择性路由：并非每个查询都需要每个源——路由器会智能地选择相关的垂直领域。
  • 针对性子问题：每个智能体收到一个针对其领域定制的问题，提高了结果质量。
  • 清晰合成：来自多个源的结果被组合成一个单一的、连贯的响应。

概念 ​

我们将涵盖以下概念：

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

设置 ​

安装 ​

本教程需要 langchain 和 langgraph 包：

bash
pip install langchain langgraph
bash
uv add langchain langgraph
bash
conda install langchain langgraph -c conda-forge

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

LangSmith ​

设置 LangSmith 以检查您的智能体内部发生的情况。然后设置以下环境变量：

bash
export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="..."
python
import getpass
import os

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

选择LLM ​

从LangChain的集成套件中选择一个聊天模型：

OpenAI
Anthropic
Azure
Google Gemini
AWS Bedrock
HuggingFace

👉 阅读 OpenAI 聊天模型集成文档

shell
pip install -U "langchain[openai]"
python
import os
from langchain.chat_models import init_chat_model

os.environ["OPENAI_API_KEY"] = "sk-..."

model = init_chat_model("gpt-4.1")
python
import os
from langchain_openai import ChatOpenAI

os.environ["OPENAI_API_KEY"] = "sk-..."

model = ChatOpenAI(model="gpt-4.1")

1. 定义状态 ​

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

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

class AgentInput(TypedDict):
    """Simple input state for each subagent."""
    query: str

class AgentOutput(TypedDict):
    """Output from each subagent."""
    source: str
    result: str

class Classification(TypedDict):
    """A single routing decision: which agent to call with what query."""
    source: Literal["github", "notion", "slack"]
    query: str

class RouterState(TypedDict):
    query: str
    classifications: list[Classification]
    results: Annotated[list[AgentOutput], operator.add]  # Reducer collects parallel results
    final_answer: str

results 字段使用一个归约器（Python中的 operator.add，JS中的concat函数）将并行智能体执行的输出收集到一个列表中。

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

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

expandable
python
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 (API auth docs), #89 (OAuth flow), #203 (token refresh)"

@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, API keys, and JWT tokens"

@tool
def get_page(page_id: str) -> str:
    """Get a specific Notion page by ID."""
    return f"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, see docs for refresh flow'"

@tool
def get_thread(thread_id: str) -> str:
    """Get a specific Slack thread."""
    return f"Thread discusses best practices for API key rotation"

3. 创建专用智能体 ​

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

expandable
python
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4o")

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."
    ),
)

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-4o-mini")

# Define structured output schema for the classifier
class ClassificationResult(BaseModel):  
    """Result of classifying a user query into agent-specific sub-questions."""
    classifications: list[Classification] = Field(
        description="List of agents to invoke with their targeted sub-questions"
    )

def classify_query(state: RouterState) -> dict:
    """Classify query and determine which agents to invoke."""
    structured_llm = router_llm.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. Each source should have
a targeted sub-question optimized for that specific knowledge domain.

Example for "How do I authenticate API requests?":
- github: "What authentication code exists? Search for auth middleware, JWT handling"
- notion: "What authentication documentation exists? Look for API auth guides"
(slack omitted because it's not relevant for this technical question)"""
        },
        {"role": "user", "content": state["query"]}
    ])

    return {"classifications": result.classifications}

def route_to_agents(state: RouterState) -> list[Send]:
    """Fan out to agents based on classifications."""
    return [
        Send(c["source"], {"query": c["query"]})  
        for c in state["classifications"]
    ]

def query_github(state: AgentInput) -> dict:
    """Query the GitHub agent."""
    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:
    """Query the Notion agent."""
    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:
    """Query the Slack agent."""
    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:
    """Combine results from all agents into a coherent answer."""
    if not state["results"]:
        return {"final_answer": "No results found from any knowledge source."}

    # Format results for synthesis
    formatted = [
        f"**From {r['source'].title()}:**\n{r['result']}"
        for r in state["results"]
    ]

    synthesis_response = router_llm.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}

5. 编译工作流 ​

现在通过用边连接节点来组装工作流。关键是使用带有路由函数的 add_conditional_edges 来启用并行执行：

python
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 函数将分类节点连接到智能体节点。当 route_to_agents 返回多个 Send 对象时，这些节点会并行执行。

6. 使用路由器 ​

使用跨越多个知识领域的查询测试您的路由器：

python
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("\n" + "=" * 60 + "\n")
print("Final Answer:")
print(result["final_answer"])

预期输出：

Original query: How do I authenticate API requests?

Classifications:
  github: What authentication code exists? Search for auth middleware, JWT handling
  notion: What authentication documentation exists? Look for API auth guides

============================================================

Final Answer:
To authenticate API requests, you have several options:

1. **JWT Tokens**: The recommended approach for most use cases.
   Implementation details are in `src/auth.py` (PR #156).

2. **OAuth2 Flow**: For third-party integrations, follow the OAuth2
   flow documented in Notion's 'API Authentication Guide'.

3. **API Keys**: For server-to-server communication, use Bearer tokens
   in the Authorization header.

For token refresh handling, see issue #203 and PR #178 for the latest
OAuth scope updates.

路由器分析了查询，对其分类以确定调用哪些智能体（对于这个技术问题，是GitHub和Notion，而不是Slack），并行查询了两个智能体，并将结果合成为一个连贯的答案。

7. 理解架构 ​

路由器工作流遵循一个清晰的模式：

分类阶段 ​

classify_query 函数使用结构化输出来分析用户的查询并确定调用哪些智能体。这是路由智能所在之处：

  • 使用Pydantic模型（Python）或Zod模式（JS）来确保输出有效
  • 返回一个 Classification 对象列表，每个对象包含一个 source 和目标 query
  • 仅包含相关源——不相关的源会被直接省略

这种结构化方法比自由格式的JSON解析更可靠，并使路由逻辑更加明确。

使用Send进行并行执行 ​

route_to_agents 函数将分类映射到 Send 对象。每个 Send 指定目标节点和要传递的状态：

python
# Classifications: [{"source": "github", "query": "..."}, {"source": "notion", "query": "..."}]
# Becomes:
[Send("github", {"query": "..."}), Send("notion", {"query": "..."})]
# Both agents execute simultaneously, each receiving only the query it needs

每个智能体节点接收一个仅包含 query 字段的简单 AgentInput——而不是完整的路由器状态。这保持了接口的简洁和明确。

使用归约器收集结果 ​

智能体结果通过归约器流回主状态。每个智能体返回：

python
{"results": [{"source": "github", "result": "..."}]}

归约器（Python中的 operator.add）连接这些列表，将所有并行结果收集到 state["results"] 中。

合成阶段 ​

在所有智能体完成后，synthesize_results 函数遍历收集到的结果：

  • 等待所有并行分支完成（LangGraph会自动处理）
  • 引用原始查询以确保答案解决了用户的问题
  • 组合所有来源的信息，避免冗余

部分结果：在本教程中，所有选定的智能体必须在合成之前完成。对于更高级的模式，例如您希望处理部分结果或超时，请参阅 map-reduce指南。

8. 完整工作示例 ​

以下是一个可运行脚本中的所有内容：

Show 查看完整代码

9. 高级：有状态路由器 ​

我们目前构建的路由器是无状态的——每个请求都是独立处理的，调用之间没有记忆。对于多轮对话，您需要一种有状态的方法。

工具包装器方法 ​

添加对话记忆的最简单方法是将无状态路由器包装为对话智能体可以调用的工具：

python
from langgraph.checkpoint.memory import InMemorySaver

@tool
def search_knowledge_base(query: str) -> str:
    """Search across multiple knowledge sources (GitHub, Notion, Slack).

    Use this to find information about code, documentation, or team discussions.
    """
    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 that answers questions about our organization. "
        "Use the search_knowledge_base tool to find information across our code, "
        "documentation, and team discussions."
    ),
    checkpointer=InMemorySaver(),
)

这种方法保持了路由器的无状态性，而对话智能体则处理记忆和上下文。用户可以进行多轮对话，智能体将在需要时调用路由器工具。

python
config = {"configurable": {"thread_id": "user-123"}}

result = conversational_agent.invoke(
    {"messages": [{"role": "user", "content": "How do I authenticate API requests?"}]},
    config
)
print(result["messages"][-1].content)

result = conversational_agent.invoke(
    {"messages": [{"role": "user", "content": "What about rate limiting for those endpoints?"}]},
    config
)
print(result["messages"][-1].content)

对于大多数用例，推荐使用工具包装器方法。它提供了清晰的分离：路由器处理多源查询，而对话智能体处理上下文和记忆。

完全持久化方法 ​

如果您需要路由器本身维护状态——例如，在路由决策中使用先前的搜索结果——请使用持久化在路由器级别存储消息历史记录。

有状态路由器增加了复杂性。 当在不同轮次中路由到不同的智能体时，如果智能体具有不同的语气或提示词，对话可能会感觉不一致。请考虑使用 交接模式 或 子智能体模式 代替——两者都为与不同智能体的多轮对话提供了更清晰的语义。

10. 关键要点 ​

路由模式在以下情况下表现出色：

  • 不同的垂直领域：每个都需要专用工具和提示词的独立知识领域
  • 并行查询需求：受益于同时查询多个源的问题
  • 合成需求：需要将来自多个源的结果组合成一个连贯的响应

该模式有三个阶段：分解（分析查询并生成有针对性的子问题）、路由（并行执行查询）和合成（组合结果）。

何时使用路由模式

当您有多个独立的知识源、需要低延迟的并行查询并且希望显式控制路由逻辑时，请使用路由模式。

对于具有动态工具选择的更简单情况，请考虑 子智能体模式。对于需要智能体与用户顺序对话的工作流，请考虑 交接模式。

后续步骤 ​

LangChain 中文文档

Copyright © 王码码