16. 预构建组件与多分支路由
在第 15 章中，我们手动组装了一个代理图——一个模型节点、一个工具节点，以及一条决定继续循环还是停止的条件边。这本质上就是工具调用代理的标准结构，因此 LangChain 和 LangGraph 将其作为预构建组件提供，你可以直接使用，而不必每次都从头编写相同的脚手架代码。
在本章的前半部分，我们将使用预构建组件重建第 15 章的代理。我们会用 ToolNode 和 tools_condition 替换工具执行节点和路由函数，最后用一次 create_agent 调用替换整个图组装过程。你会看到其行为与第 15 章完全一致，而代码却大幅缩减。
在后半部分，我们将结合预构建组件与第 15 章的手动图组装方法，构建一个更复杂的代理。我们要构建的代理会将每个请求路由到不同的处理器——复杂咨询交给高性能模型，而简单问题则由更便宜、更小的模型回答。这是一种多分支结构，处理路径会根据请求类型而分叉。
16.1) 预构建组件与 create_agent
在本节中，我们将用预构建组件 ToolNode 和 tools_condition 替换第 15 章图中的 tool_node 函数和 should_continue 函数。之后，我们将完全跳过手动组装，用一次 create_agent 调用创建整个图。每一步都要留意的是：代码越来越短，而代理的行为与第 15 章保持一致。
16.1.1) ToolNode 与 tools_condition
ToolNode 是一个为你处理工具执行的预构建节点。当 State 中的最后一条消息（LLM 返回的 AIMessage）包含 tool_calls 时，它会运行所请求的工具，并将结果作为 ToolMessage 对象添加到 messages 中。它做的工作与我们在第 15 章编写的 tool_node 函数相同。此外，当 LLM 一次请求多个工具时，它会并行运行这些工具。
它还支持工具执行期间的异常处理。如果你设置 ToolNode(tools, handle_tool_errors=True)，即使某个工具抛出异常，图也不会崩溃。异常会被转换为携带错误详情的 ToolMessage，并传递给 LLM，让它能够看到失败情况并使用修正后的参数重试。
你通过传入一个工具列表来创建 ToolNode。我们将使用第 15 章中相同的两个工具。
from langgraph.prebuilt import ToolNode
from langchain.tools import tool
@tool
def get_weather(city: str) -> str:
"""获取某个城市的当前天气。"""
fake_data = {"Tokyo": "18°C, cloudy", "Cairo": "31°C, sunny"}
return fake_data.get(city, f"No weather data for {city}.")
@tool
def calculate(expression: str) -> str:
"""计算一个简单的算术表达式。示例: '3 * 21'。"""
return str(eval(expression)) # 警告: eval() 存在安全风险。请勿在生产环境中使用。
tools = [get_weather, calculate]
# 第 15 章的 tool_map + tool_node 函数被这一行替换
tool_node = ToolNode(tools)ToolNode 会根据工具列表构建一个内部的名称到工具的映射，就像第 15 章的 tool_map 一样。在运行时，它会根据 LLM 请求的名称查找每个工具并调用它。换句话说，tool_map 字典、遍历 tool_calls 的 for 循环，以及构建和收集 ToolMessage 对象的代码——所有这些现在都存在于 ToolNode 内部。
tools_condition 是用于替换第 15 章 should_continue 函数的预构建路由函数。我们再来看看第 15 章的 should_continue。
def should_continue(state: AgentState) -> Literal["tool_node", "__end__"]:
"""决定是执行工具还是结束图。"""
last_message = state["messages"][-1]
if last_message.tool_calls:
return "tool_node"
return END当最后一条消息带有 tool_calls 时，它返回 "tool_node"（我们工具节点的注册名称），否则返回 END。tools_condition 的工作方式完全相同，只有它返回的名称存在一处差异。should_continue 被写为返回 "tool_node"——我们在图中注册的名称——而 tools_condition 被硬编码为返回 "tools"。
正如我们在 15.2.4 节所学，路由函数返回的值是下一个要执行节点的名称。如果图中不存在具有该名称的节点，路由就会失败。因此，使用 tools_condition 时，工具节点必须以 "tools" 这个名称注册。
from langgraph.prebuilt import ToolNode, tools_condition
builder.add_node("tools", ToolNode(tools)) # 以名称 "tools" 注册
builder.add_conditional_edges("llm_call", tools_condition) # 路由到 "tools" 或 END这样，当 tools_condition 返回 "tools" 时，它会准确地连接到我们刚刚注册的 ToolNode。
现在，让我们重新组装第 15 章的完整图，包括所有其余部分。工具、State 和 llm_call 节点与第 15 章保持不变。
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain.tools import tool
@tool
def get_weather(city: str) -> str:
"""获取某个城市的当前天气。"""
fake_data = {"Tokyo": "18°C, cloudy", "Cairo": "31°C, sunny"}
return fake_data.get(city, f"No weather data for {city}.")
@tool
def calculate(expression: str) -> str:
"""计算一个简单的算术表达式。示例: '3 * 21'。"""
return str(eval(expression)) # 警告: eval() 存在安全风险。请勿在生产环境中使用。
tools = [get_weather, calculate]
class AgentState(MessagesState):
llm_calls: int
llm = ChatOpenAI(model="gpt-5-mini")
model_with_tools = llm.bind_tools(tools)
def llm_call(state: AgentState):
"""调用 LLM 并返回其响应。"""
response = model_with_tools.invoke(state["messages"])
return {
"messages": [response],
"llm_calls": state.get("llm_calls", 0) + 1,
}
builder = StateGraph(AgentState)
builder.add_node("llm_call", llm_call)
builder.add_node("tools", ToolNode(tools)) # 使用 ToolNode 而非第 15 章的 tool_node 函数
builder.add_edge(START, "llm_call")
builder.add_conditional_edges("llm_call", tools_condition) # 使用 tools_condition 而非第 15 章的 should_continue
builder.add_edge("tools", "llm_call")
agent = builder.compile()将它与第 15 章的代码对比一下。tool_map 字典、tool_node 函数和 should_continue 函数都消失了。它们所做的工作现在由 ToolNode(tools) 和 tools_condition 处理。工具节点注册为 "tools"，以匹配 tools_condition 路由到的名称。让我们用第 15 章中相同的问题来运行它。
result = agent.invoke({
"messages": [HumanMessage(content="获取开罗的气温，然后将这个数字乘以 3。")],
"llm_calls": 0,
})
print(result["messages"][-1].content)
print(f"\nTotal LLM calls: {result['llm_calls']}")输出:
开罗当前气温：31°C。乘以 3 = 93。
Total LLM calls: 3结果与第 15 章完全一致。代理查询天气、执行计算并生成最终答案——行为完全保留下来，而我们需要编写和维护的代码却缩减了。
如果你想以 "tools" 以外的名称注册工具节点该怎么办？这种情况下，向 add_conditional_edges 传入一个映射字典作为第三个参数，指定 tools_condition 的每个返回值应该连接到哪个节点。由于 tools_condition 返回 "tools" 或 END，你就用它们作为键，并将它们映射到目标节点。例如，如果你将工具节点注册为 "run_tools"：
builder.add_node("run_tools", ToolNode(tools))
builder.add_conditional_edges(
"llm_call",
tools_condition,
{"tools": "run_tools", END: END} # "tools" 返回 → run_tools 节点，END 返回 → 终止
)ToolNode 和 tools_condition 替换了图中的各个部分——那些繁琐的部分——但添加节点并将它们连接起来仍然由我们负责。我们能把这个组装工作也交出去吗？这正是 create_agent 所做的。
16.1.2) create_agent
create_agent 是一个 LangChain 工厂函数，它为工具调用代理处理整个图组装过程。向它传入一个模型和一个工具列表，它就会构建一个与我们在 16.1.1 中组装的结构相同的图，且已经编译好、可以运行。让我们试试看。
from langchain.agents import create_agent
from langchain.tools import tool
from langchain_core.messages import HumanMessage
@tool
def get_weather(city: str) -> str:
"""获取某个城市的当前天气。"""
fake_data = {"Tokyo": "18°C, cloudy", "Cairo": "31°C, sunny"}
return fake_data.get(city, f"No weather data for {city}.")
@tool
def calculate(expression: str) -> str:
"""计算一个简单的算术表达式。示例: '3 * 21'。"""
return str(eval(expression)) # 警告: eval() 存在安全风险。请勿在生产环境中使用。
agent = create_agent(
model="openai:gpt-5-mini",
tools=[get_weather, calculate],
system_prompt="你是一个有帮助的助手。",
)
result = agent.invoke({
"messages": [HumanMessage(content="获取开罗的气温，然后将这个数字乘以 3。")],
})
print(result["messages"][-1].content)输出:
开罗当前气温：31°C。乘以 3 = 93。没有 State 定义，没有节点函数，没有 add_node 或 add_edge。一次 create_agent 调用完成了所有这些，其结果与 16.1.1 完全一致。
内部发生的事情正是我们已经了解的。create_agent 会从你传入的模型创建一个 LLM 调用节点，从工具列表构建一个 ToolNode，然后用一条 tools_condition 边和一条回环边将它们连接起来。其结果是一个与我们在 16.1.1 中组装的完全相同的循环结构图。
让我们来看看这些参数。create_agent 对我们来说其实并不陌生——我们在第 11 章构建对话式 RAG 时简要使用过它，但没有详细介绍参数。让我们逐个过一遍。
model：代理将使用的 LLM。最简单的方式是传入一个提供商字符串，如"openai:gpt-5-mini"。如果你需要直接配置模型参数，可以传入一个已初始化的模型实例，如ChatOpenAI(model="gpt-5-mini")。在内部，LLM 调用节点使用这个模型。tools：代理可以使用的工具列表。内部会从这些工具构建一个ToolNode。system_prompt：代理的行为指令。它会在每次 LLM 调用时作为系统消息添加到消息列表的开头。checkpointer：保存对话状态，使代理能够记住之前的轮次。这与我们在第 11 章使用InMemorySaver()和thread_id来实现多轮对话的参数相同。我们将在第 17 章详细介绍它的工作原理。response_format：当你想将代理的最终答案作为结构化输出时使用它。传入一个 Pydantic 模型（与第 7 章相同的概念），经过验证的对象将在result["structured_response"]中可用。middleware：注册在代理执行循环的特定节点运行的函数。这是我们在第 11 章用来注册trim_old_messages的参数。我们将在下面详细解释它。
让我们看看 response_format 在实践中是如何工作的。
from pydantic import BaseModel
from langchain.agents import create_agent
class WeatherReport(BaseModel):
city: str
temperature: str
condition: str
agent = create_agent(
model="openai:gpt-5-mini",
tools=[get_weather, calculate],
response_format=WeatherReport,
)
result = agent.invoke({
"messages": [HumanMessage(content="东京的天气怎么样？")],
})
print(result["structured_response"])输出:
city='Tokyo' temperature='18°C' condition='cloudy'代理调用了 get_weather 工具，然后将信息整理成一个与模式匹配的 WeatherReport 对象。
middleware
代理循环有不同的阶段。它调用 LLM、执行工具、再次调用 LLM——这些阶段不断重复。中间件(middleware) 让你可以在这些阶段之前或之后插入你自己的函数。你用装饰器指定时机：@before_model 表示在 LLM 调用之前，@after_model 表示在 LLM 响应之后。将函数注册到 middleware 参数中，它就会每次在指定的节点运行。
我们在第 11 章已经使用过中间件。我们用 @before_model 装饰了一个 trim_old_messages 函数并注册它——由于它在每次 LLM 调用之前运行，所以它每次都能修剪消息列表。
让我们构建一个中间件，在每次 LLM 调用之前打印消息数量，这样我们就能准确地看到它何时运行。
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_model
@before_model
def log_llm_call(state: AgentState, runtime) -> None:
"""在每次 LLM 调用之前打印消息数量。"""
print(f"[before_model] About to call LLM, current messages: {len(state['messages'])}")
agent = create_agent(
model="openai:gpt-5-mini",
tools=[get_weather, calculate],
middleware=[log_llm_call],
)
result = agent.invoke({
"messages": [HumanMessage(content="获取开罗的气温，然后将这个数字乘以 3。")],
})
print(result["messages"][-1].content)输出:
[before_model] About to call LLM, current messages: 1
[before_model] About to call LLM, current messages: 3
[before_model] About to call LLM, current messages: 5
开罗当前气温：31°C。乘以 3 = 93。log_llm_call 中间件运行了三次。在处理用户请求的过程中，LLM 被调用了三次，而中间件在每次调用之前运行。消息数量告诉了我们每个时刻的 State：在第一次调用之前，只有用户的问题（HumanMessage）——1 条消息。每次循环迭代后，请求工具调用的 AIMessage 和带有结果的 ToolMessage 被添加进来，增长到 3 条，然后是 5 条。
有一点需要了解：在 LangChain 1.0 之前，这个角色由 LangGraph 一侧一个名为
create_react_agent的函数承担，现在它已被弃用。如果你在较旧的教程或博客文章中看到from langgraph.prebuilt import create_react_agent，请理解它是你现在正在学习的create_agent的先前版本。
我们已经用 ToolNode、tools_condition 和 create_agent 简洁地重建了第 15 章的代理。在下一节中，我们将结合这些预构建组件与手动图组装来构建一个更复杂的代理。
16.2) 构建多分支代理
我们将在本节中构建的多分支代理(multi-branch agent) 首先判断它面对的是哪种请求，然后用不同的模型或不同的工具集处理每种请求。我们将使用第 15 章的方法手动组装整体图，并对需要工具调用循环的部分使用 create_agent。
16.2.1) 需求与设计
我们将构建本章引言中提到的客户支持代理。以下是需求：
- 简单咨询（"你们的营业时间是什么？"）→ 由低成本小模型直接回答。
- 复杂咨询（"我的订单到货时损坏了——我该换货还是退款？"）→ 由高性能模型回答。
- 订单查询（"订单 #12345 的配送状态是什么？"）→ 由带有订单查询工具的代理处理。
每种咨询类型都需要不同的模型和工具配置，因此我们需要一个先对每个请求进行分类、然后将其路由到正确处理器的图。结构如下：
当一个请求进来时，classify 节点判断它属于哪种类型的咨询，并将结果记录在 State 中。然后一条条件边从 State 中读取记录的类型，并路由到相应的节点。simple_handler 处理简单咨询，complex_handler 处理复杂咨询。order_agent 使用订单查询工具来检查配送状态并响应。由于 order_agent 是一个标准的工具调用节点，我们将用 create_agent 构建它。现在让我们依次构建每个部分。
16.2.2) 分类节点与路由函数
首先，让我们定义 State。我们将添加一个 intent 字段来存储分类结果。这三种类型将由值 "simple"、"complex" 和 "order" 表示。
from langgraph.graph import MessagesState
class State(MessagesState):
intent: str # 分类结果: "simple"、"complex"、"order"接下来是分类节点。它使用 LLM 来判断用户的请求属于三种类型中的哪一种，并将结果记录在 intent 中。我们使用第 7 章学过的结构化输出来接收分类结果。当我们用 Literal 类型声明模式的 intent 字段时，LLM 的响应会被限制为所声明的值之一。
from typing import Literal
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
class IntentRoute(BaseModel):
"""客户咨询的分类结果。"""
intent: Literal["simple", "complex", "order"] = Field(
description=(
"simple: 一般性问题，如营业时间或问候。 "
"complex: 需要仔细推理的咨询，如纠纷或退款。 "
"order: 查询特定订单的请求。"
)
)
classifier_llm = ChatOpenAI(model="gpt-5.4-nano").with_structured_output(IntentRoute)
def classify(state: State):
"""对客户咨询的类型进行分类。"""
question = state["messages"][-1].content
result = classifier_llm.invoke(
f"对客户的请求进行分类。\n\n请求: {question}"
)
return {"intent": result.intent}我们用了最小的模型（gpt-5.4-nano）来进行分类。判断"这是哪种类型的咨询？"是一项不需要高性能模型的简单任务。而且由于分类节点是每个请求都会经过的入口，因此便宜又快速的模型更为可取。
接下来是路由函数。它只是返回存储在 State 中的分类结果。
def route_by_intent(state: State) -> Literal["simple", "complex", "order"]:
"""根据分类结果确定下一个节点。"""
return state["intent"]classify 节点已经决定了接下来应该运行哪个节点，并将其记录在 intent 中，所以路由函数只是原样返回该值。
16.2.3) 按类型划分的处理器节点
现在让我们为三种咨询类型分别构建处理器。
简单咨询处理器调用一次小模型。在真实的客户支持代理中，你会应用 RAG 来搜索内部文档以获取答案，但我们让处理器保持简单，以便专注于本章的主题。
simple_llm = ChatOpenAI(model="gpt-5.4-mini")
def simple_handler(state: State):
"""用小模型回答简单咨询。"""
response = simple_llm.invoke(state["messages"])
return {"messages": [response]}复杂咨询处理器使用高性能模型。出于与简单咨询处理器相同的原因，我们让它保持简单——只是生成一个响应。
complex_llm = ChatOpenAI(model="gpt-5.4")
def complex_handler(state: State):
"""用高性能模型回答复杂咨询。"""
response = complex_llm.invoke(state["messages"])
return {"messages": [response]}订单查询处理器需要使用订单查询工具，这意味着它需要一个工具调用循环。由于它的结构与标准工具调用代理相同，我们将用 create_agent 构建它。
from langchain.tools import tool
from langchain.agents import create_agent
@tool
def get_order_status(order_id: str) -> str:
"""根据订单号查询订单的配送状态。"""
fake_data = {"12345": "In transit, expected tomorrow", "67890": "Delivered"}
return fake_data.get(order_id, f"Order {order_id} not found.")
order_agent = create_agent(
model="openai:gpt-5.4-mini",
tools=[get_order_status],
)16.2.4) 组装并运行图
让我们将构建好的所有节点连接成一个图。我们把 classify 放在起点，通过一条条件边将它连接到三个处理器，并将每个处理器配置为在完成后终止。
from langgraph.graph import StateGraph, START, END
builder = StateGraph(State)
builder.add_node("classify", classify)
builder.add_node("simple", simple_handler)
builder.add_node("complex", complex_handler)
builder.add_node("order", order_agent) # 将 create_agent 图注册为一个节点
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route_by_intent)
builder.add_edge("simple", END)
builder.add_edge("complex", END)
builder.add_edge("order", END)
agent = builder.compile()让我们为每种类型运行一个咨询，看看是哪个处理器在处理它。
from langchain_core.messages import HumanMessage
for question in [
"你们的营业时间是什么？",
"我的订单到货时损坏了。我该换货还是退款？",
"订单 12345 的配送状态是什么？",
]:
result = agent.invoke({"messages": [HumanMessage(content=question)]})
print(f"Q: {question}")
print(f"[{result['intent']}] A: {result['messages'][-1].content}\n")输出:
Q: 你们的营业时间是什么？
[simple] A: 我没有固定的营业时间——我全天候 24/7 可用。
...
Q: 我的订单到货时损坏了。我该换货还是退款？
[complex] A: 如果您的订单到货时损坏了，通常您应该有权获得**替换/换货或全额退款**。
...
Q: 订单 12345 的配送状态是什么？
[order] A: 订单 12345 正在**运输途中**，**预计明天**送达。每种咨询都通过不同的路径处理。simple_handler 用小模型回答了简单咨询，complex_handler 用高性能模型回答了复杂咨询，而 order_agent 调用 get_order_status 工具回答了订单查询。