在路由器架构中，路由步骤对输入进行分类，并将其定向到专门的 智能体。当你拥有不同的垂直领域（各自需要独立智能体的独立知识域）时，这种方法很有用。

关键特征

• 路由器对查询进行分解
• 并行调用零个或多个专门的智能体
• 结果被综合成连贯的响应

何时使用

当你拥有不同的垂直领域（各自需要独立智能体的独立知识域）、需要并行查询多个数据源，并希望将结果综合为组合响应时，请使用路由器模式。

基本实现

路由器对查询进行分类，并将其定向到合适的智能体。对于单智能体路由，使用 Command；对于并行分发到多个智能体，使用 Send。

单个智能体

使用 Command 路由到单个专门的智能体：

from langgraph.types import Command

def classify_query(query: str) -> str:
    """Use LLM to classify query and determine the appropriate agent."""
    # 此处为分类逻辑
    ...

def route_query(state: State) -> Command:
    """Route to the appropriate agent based on query classification."""
    active_agent = classify_query(state["query"])

    # 路由到选定的智能体
    return Command(goto=active_agent)

import { z } from "zod";
import { Command } from "@langchain/langgraph";

const ClassificationResult = z.object({
  query: z.string(),
  agent: z.string(),
});

function classifyQuery(query: string): z.infer<typeof ClassificationResult> {
  // 使用 LLM 对查询进行分类并确定合适的智能体
  // 此处为分类逻辑
  ...
}

function routeQuery(state: z.infer<typeof ClassificationResult>) {
  const classification = classifyQuery(state.query);

  // 路由到选定的智能体
  return new Command({ goto: classification.agent });
}

多个智能体（并行）

使用 Send 并行分发到多个专门的智能体：

from typing import TypedDict
from langgraph.types import Send

class ClassificationResult(TypedDict):
    query: str
    agent: str

def classify_query(query: str) -> list[ClassificationResult]:
    """Use LLM to classify query and determine which agents to invoke."""
    # 此处为分类逻辑
    ...

def route_query(state: State):
    """Route to relevant agents based on query classification."""
    classifications = classify_query(state["query"])

    # 并行分发到选定的智能体
    return [
        Send(c["agent"], {"query": c["query"]})
        for c in classifications
    ]

import { z } from "zod";
import { Command } from "@langchain/langgraph";

const ClassificationResult = z.object({
  query: z.string(),
  agent: z.string(),
});

function classifyQuery(query: string): z.infer<typeof ClassificationResult>[] {
  // 使用 LLM 对查询进行分类并确定合适的智能体
  // 此处为分类逻辑
  ...
}

function routeQuery(state: typeof State.State) {
  const classifications = classifyQuery(state.query);

  // 并行分发到选定的智能体
  return classifications.map(
    (c) => new Send(c.agent, { query: c.query })
  );
}

完整的实现请参见下面的教程。

• 教程：使用路由构建多源知识库 — 构建一个并行查询 GitHub、Notion 和 Slack 的路由器，然后将结果综合为连贯的答案。涵盖状态定义、专门智能体、使用 Send 的并行执行以及结果综合。

无状态与有状态

两种方法：

• 无状态路由器 独立处理每个请求
• 有状态路由器 跨请求维护对话历史

无状态

每个请求都被独立路由——调用之间没有记忆。对于多轮对话，请参阅 有状态路由器。

TIP

路由器 vs. 子智能体：两种模式都可以将工作分派给多个智能体，但它们的路由决策方式不同：

• 路由器：一个专门的路由步骤（通常是一次 LLM 调用或基于规则的逻辑），对输入进行分类并分派给智能体。路由器本身通常不维护对话历史或进行多轮编排——它是一个预处理步骤。
• 子智能体：一个主监督智能体在持续对话中动态决定调用哪些 子智能体。主智能体维护上下文，可以跨轮次调用多个子智能体，并编排复杂的多步骤工作流。

当你具有清晰的输入类别并希望进行确定性或轻量级分类时，请使用路由器。当你需要灵活的、感知对话的编排，由 LLM 根据不断演变的上下文决定下一步做什么时，请使用监督智能体。

有状态

对于多轮对话，你需要在多次调用之间维护上下文。

工具包装

最简单的方法：将无状态路由器包装为对话智能体可以调用的工具。对话智能体负责记忆和上下文；路由器保持无状态。这避免了跨多个并行智能体管理对话历史的复杂性。

@tool
def search_docs(query: str) -> str:
    """Search across multiple documentation sources."""
    result = workflow.invoke({"query": query})  
    return result["final_answer"]

# 对话智能体将路由器用作工具
conversational_agent = create_agent(
    model,
    tools=[search_docs],
    prompt="You are a helpful assistant. Use search_docs to answer questions."
)

const searchDocs = tool(
  async ({ query }) => {
    const result = await workflow.invoke({ query }); 
    return result.finalAnswer;
  },
  {
    name: "search_docs",
    description: "Search across multiple documentation sources",
    schema: z.object({
      query: z.string().describe("The search query"),
    }),
  }
);

// 对话智能体将路由器用作工具
const conversationalAgent = createAgent({
  model,
  tools: [searchDocs],
  systemPrompt: "You are a helpful assistant. Use search_docs to answer questions.",
});

完整持久化

如果你需要路由器本身维护状态，请使用 持久化 来存储消息历史。路由到智能体时，从状态中获取先前的消息，并选择性地将其包含在智能体的上下文中——这是 上下文工程 的一个杠杆。

WARNING

有状态路由器需要自定义历史管理。 如果路由器在轮次之间切换智能体，当各智能体具有不同的语气或提示词时，对话对最终用户而言可能不够流畅。对于并行调用，你需要在路由器级别维护历史（输入和综合后的输出），并在路由逻辑中利用该历史。请考虑改用 交接模式 或 子智能体模式——两者都为多轮对话提供了更清晰的语义。