构建带路由的多源知识库

概述

路由器模式是一种多智能体架构，其中路由步骤对输入进行分类并将其定向给专业化的智能体，然后将结果综合成一个组合响应。当您的组织知识分散在不同的垂直领域（即独立的知识域，每个都需要自己的、配备专业工具和提示词的智能体）时，这种模式尤为出色。在本教程中，您将构建一个多源知识库路由器，通过一个实际的企业场景来展示这些优势。该系统将协调三个专业智能体：

GitHub 智能体：搜索代码、议题和拉取请求。
Notion 智能体：搜索内部文档和维基。
Slack 智能体：搜索相关的线程和讨论。

当用户提问”如何认证 API 请求？“时，路由器会将查询分解为针对特定数据源的子问题，并行路由给相关的智能体，然后将结果综合成一个连贯的答案。

为什么使用路由器？

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

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

核心概念

我们将涵盖以下概念：

多智能体系统
用于工作流编排的 StateGraph
用于并行执行的 Send API

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

环境设置

安装

本教程需要 langchain 和 langgraph 包：

pip install langchain langgraph

uv add langchain langgraph

conda install langchain langgraph -c conda-forge

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

LangSmith

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

export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="..."

import getpass
import os

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

选择一个 LLM

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

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

pip install -U "langchain[openai]"

import os
from langchain.chat_models import init_chat_model

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

model = init_chat_model("gpt-5.5")

import os
from langchain_openai import ChatOpenAI

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

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

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

pip install -U "langchain[anthropic]"

import os
from langchain.chat_models import init_chat_model

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

model = init_chat_model("claude-sonnet-4-6")

import os
from langchain_anthropic import ChatAnthropic

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

model = ChatAnthropic(model="claude-sonnet-4-6")

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

pip install -U "langchain[openai]"

import os
from langchain.chat_models import init_chat_model

os.environ["AZURE_OPENAI_API_KEY"] = "..."
os.environ["AZURE_OPENAI_ENDPOINT"] = "..."
os.environ["OPENAI_API_VERSION"] = "2025-03-01-preview"

model = init_chat_model(
    "azure_openai:gpt-5.5",
    azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
)

import os
from langchain_openai import AzureChatOpenAI

os.environ["AZURE_OPENAI_API_KEY"] = "..."
os.environ["AZURE_OPENAI_ENDPOINT"] = "..."
os.environ["OPENAI_API_VERSION"] = "2025-03-01-preview"

model = AzureChatOpenAI(
    model="gpt-5.5",
    azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"]
)

👉 阅读 Google GenAI 聊天模型集成文档

pip install -U "langchain[google-genai]"

import os
from langchain.chat_models import init_chat_model

os.environ["GOOGLE_API_KEY"] = "..."

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

import os
from langchain_google_genai import ChatGoogleGenerativeAI

os.environ["GOOGLE_API_KEY"] = "..."

model = ChatGoogleGenerativeAI(model="gemini-2.5-flash-lite")

👉 阅读 AWS Bedrock 聊天模型集成文档

pip install -U "langchain[aws]"

from langchain.chat_models import init_chat_model

# 请按照此页面的步骤配置您的凭证：
# https://docs.aws.amazon.com/bedrock/latest/userguide/getting-started.html

model = init_chat_model(
    "us.anthropic.claude-sonnet-4-6",
    model_provider="bedrock_converse",
)

from langchain_aws import ChatBedrock

model = ChatBedrock(model="us.anthropic.claude-sonnet-4-6")

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

pip install -U "langchain[huggingface]"

import os
from langchain.chat_models import init_chat_model

os.environ["HUGGINGFACEHUB_API_TOKEN"] = "hf_..."

model = init_chat_model(
    "microsoft/Phi-3-mini-4k-instruct",
    model_provider="huggingface",
    temperature=0.7,
    max_tokens=1024,
)

import os
from langchain_huggingface import ChatHuggingFace, HuggingFaceEndpoint

os.environ["HUGGINGFACEHUB_API_TOKEN"] = "hf_..."

llm = HuggingFaceEndpoint(
    repo_id="microsoft/Phi-3-mini-4k-instruct",
    temperature=0.7,
    max_length=1024,
)
model = ChatHuggingFace(llm=llm)

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

pip install -U "langchain-openrouter"

import os
from langchain.chat_models import init_chat_model

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

model = init_chat_model(
    "auto",
    model_provider="openrouter",
)

import os
from langchain_openrouter import ChatOpenRouter

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

model = ChatOpenRouter(model="auto")

1. 定义状态

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

AgentInput：传递给每个子智能体的简单状态（只是一个查询）
AgentOutput：每个子智能体返回的结果（数据源名称 + 结果）
RouterState：主工作流状态，跟踪查询、分类、结果和最终答案

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

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

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

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

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 f"PR #156 添加了 JWT 认证，PR #178 更新了 OAuth 范围"


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


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


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


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

3. 创建专业化的智能体

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

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

model = init_chat_model("openai:gpt-5.5")

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 构建路由器工作流。工作流有四个主要步骤：

分类：分析查询并确定要调用哪些智能体以及使用什么子问题
路由：使用 Send 并行扇出到选定的智能体
查询智能体：每个智能体接收一个简单的 AgentInput 并返回一个 AgentOutput
综合：将收集到的结果组合成一个连贯的响应

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: "存在什么认证代码？搜索认证中间件、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:
    """将来自所有智能体的结果组合成一个连贯的答案。"""
    if not state["results"]:
        return {"final_answer": "未从任何知识源找到结果。"}

    # 格式化结果以便综合
    formatted = [
        f"**来自 {r['source'].title()}:**\n{r['result']}"
        for r in state["results"]
    ]

    synthesis_response = router_llm.invoke([
        {
            "role": "system",
            "content": f"""综合这些搜索结果来回答原始问题："{state['query']}"

- 组合来自多个来源的信息，避免冗余
- 突出最相关和可操作的信息
- 注意来源之间的任何差异
- 保持响应简洁且组织良好"""
        },
        {"role": "user", "content": "\n\n".join(formatted)}
    ])

    return {"final_answer": synthesis_response.content}

5. 编译工作流

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

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. 使用路由器

测试您的路由器是否能处理跨越多个知识域的查询：

result = workflow.invoke({
    "query": "如何认证 API 请求？"
})

print("原始查询:", result["query"])
print("\n分类:")
for c in result["classifications"]:
    print(f"  {c['source']}: {c['query']}")
print("\n" + "=" * 60 + "\n")
print("最终答案:")
print(result["final_answer"])

预期输出：

原始查询: 如何认证 API 请求？

分类:
  github: 存在什么认证代码？搜索认证中间件、JWT 处理
  notion: 存在什么认证文档？查找 API 认证指南

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

最终答案:
要认证 API 请求，您有几种选择：

1. **JWT 令牌**：大多数用例的推荐方法。
   实现细节在 `src/auth.py` 中（PR #156）。

2. **OAuth2 流程**：对于第三方集成，请遵循 Notion 的 'API 认证指南' 中记录的 OAuth2 流程。

3. **API 密钥**：对于服务器间通信，在 Authorization 头中使用 Bearer 令牌。

有关令牌刷新处理，请参阅议题 #203 和 PR #178 以了解最新的 OAuth 范围更新。

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

7. 理解架构

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

分类阶段

classify_query 函数使用结构化输出来分析用户查询并确定要调用哪些智能体。路由智能就在这里：

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

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

使用 Send 进行并行执行

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

# 分类: [{"source": "github", "query": "..."}, {"source": "notion", "query": "..."}]
# 变为:
[Send("github", {"query": "..."}), Send("notion", {"query": "..."})]
# 两个智能体同时执行，每个只接收它需要的查询

每个智能体节点接收一个简单的 AgentInput，只包含 query 字段——而不是完整的路由器状态。这保持了接口的清晰和显式。

使用归约器收集结果

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

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

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

综合阶段

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

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

部分结果：在本教程中，所有选定的智能体必须在综合前完成。

8. 完整的工作示例

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

Show 查看完整代码

"""
多源知识路由器示例

本示例演示了多智能体系统的路由器模式。
路由器对查询进行分类，并行路由到专业化的智能体，
并将结果综合成一个组合响应。"""

import operator
from typing import Annotated, Literal, TypedDict

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
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 议题和拉取请求。"""
    return f"找到 3 个匹配 '{query}' 的议题：#142（API 认证文档），#89（OAuth 流程），#203（令牌刷新）"


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


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


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


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


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


# 模型和智能体
model = init_chat_model("openai:gpt-5.5")
router_llm = init_chat_model("openai:gpt-5.4-mini")

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 专家。通过搜索团队成员分享知识和解决方案的相关"
        "线程和讨论来回答问题。"
    ),
)


# 工作流节点
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: 团队讨论、非正式知识共享、近期对话

仅返回与查询相关的来源。"""
        },
        {"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:
    """将来自所有智能体的结果组合成一个连贯的答案。"""
    if not state["results"]:
        return {"final_answer": "未从任何知识源找到结果。"}

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

    synthesis_response = router_llm.invoke([
        {
            "role": "system",
            "content": f"""综合这些搜索结果来回答原始问题："{state['query']}"

- 组合来自多个来源的信息，避免冗余
- 突出最相关和可操作的信息
- 注意来源之间的任何差异
- 保持响应简洁且组织良好"""
        },
        {"role": "user", "content": "\n\n".join(formatted)}
    ])

    return {"final_answer": synthesis_response.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()
)


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" + "=" * 60 + "\n")
    print("最终答案:")
    print(result["final_answer"])

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

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

工具包装器方法

添加对话记忆的最简单方法是将无状态路由器包装成一个工具，供对话智能体调用：

from langgraph.checkpoint.memory import InMemorySaver


@tool
def search_knowledge_base(query: str) -> str:
    """跨多个知识源（GitHub、Notion、Slack）搜索。

    用于查找关于代码、文档或团队讨论的信息。"""
    result = workflow.invoke({"query": query})
    return result["final_answer"]


conversational_agent = create_agent(
    model,
    tools=[search_knowledge_base],
    system_prompt=(
        "你是一个有用的助手，回答关于我们组织的问题。"
        "使用 search_knowledge_base 工具在我们的代码、"
        "文档和团队讨论中查找信息。"
    ),
    checkpointer=InMemorySaver(),
)

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

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

result = conversational_agent.invoke(
    {"messages": [{"role": "user", "content": "如何认证 API 请求？"}]},
    config
)
print(result["messages"][-1].content)

result = conversational_agent.invoke(
    {"messages": [{"role": "user", "content": "那些端点的速率限制如何？"}]},
    config
)
print(result["messages"][-1].content)

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

完整持久化方法

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

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

10. 关键要点

当您具备以下条件时，路由器模式表现出色：

独立的垂直领域：需要专业化工具和提示词的独立知识域
并行查询需求：能够从同时查询多个数据源中受益的问题
综合要求：需要将来自多个数据源的结果组合成一个连贯的响应

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

何时使用路由器模式当您有多个独立的知识源、需要低延迟并行查询，并且希望明确控制路由逻辑时，请使用路由器模式。对于需要动态工具选择的简单场景，请考虑子智能体模式。对于需要按顺序与用户进行对话的工作流，请考虑切换模式。

后续步骤

了解用于智能体间对话的切换模式
探索用于集中编排的子智能体模式
阅读多智能体概述以比较不同模式
使用 LangSmith 调试和监控您的路由器

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.

Tutorials

Conceptual overviews

Additional resources

概述

为什么使用路由器？

核心概念

环境设置

安装

LangSmith

选择一个 LLM

1. 定义状态

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

3. 创建专业化的智能体

4. 构建路由器工作流

5. 编译工作流

6. 使用路由器

7. 理解架构

分类阶段

使用 Send 进行并行执行

使用归约器收集结果

综合阶段

8. 完整的工作示例

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

工具包装器方法

完整持久化方法

10. 关键要点

后续步骤

​概述

​为什么使用路由器？

​核心概念

​环境设置

​安装

​LangSmith

​选择一个 LLM

​1. 定义状态

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

​3. 创建专业化的智能体

​4. 构建路由器工作流

​5. 编译工作流

​6. 使用路由器

​7. 理解架构

​分类阶段

​使用 Send 进行并行执行

​使用归约器收集结果

​综合阶段

​8. 完整的工作示例

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

​工具包装器方法

​完整持久化方法

​10. 关键要点

​后续步骤

概述

为什么使用路由器？

核心概念

环境设置

安装

LangSmith

选择一个 LLM

1. 定义状态

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

3. 创建专业化的智能体

4. 构建路由器工作流

5. 编译工作流

6. 使用路由器

7. 理解架构

分类阶段

使用 Send 进行并行执行

使用归约器收集结果

综合阶段

8. 完整的工作示例

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

工具包装器方法

完整持久化方法

10. 关键要点

后续步骤