LangGraph Router 详细解读
📚 概述
本文档详细解读 LangGraph 中的 Router(路由器) 模式。这是 Agent 系统的基础概念,通过 LLM 智能决定执行路径:直接响应用户或调用工具。Router 是理解 LangGraph 条件边和动态控制流的关键入门案例。
🎯 核心概念
什么是 Router?
Router 是一种设计模式,让 LLM 充当"交通指挥官":
- 接收输入 - 用户发送查询
- 智能判断 - LLM 分析是否需要工具
- 动态路由 - 自动选择执行路径:
- 路径 A:直接回答(无需工具)
- 路径 B:调用工具获取信息再回答
为什么 Router 重要?
这是 Agent 的本质:
- Agent ≠ 简单的函数调用
- Agent = LLM 自主决策 + 动态执行路径
- Router 是最简单但最核心的 Agent 模式
经典应用场景
- 智能客服:简单问题直接回答,复杂问题查询数据库
- 数学助手:文字问题直接答,计算问题调用计算器
- 信息查询:基础知识直接答,实时信息调用搜索 API
- 多功能 Bot:根据意图自动选择功能模块
🎭 实战案例:智能数学助手
我们将构建一个会"思考"的数学助手,演示 Router 的完整流程:
需求:
- 用户问"你好" → LLM 直接回答,无需工具
- 用户问"2 乘以 3 是多少?" → LLM 调用
multiply
工具计算
系统架构图
用户输入消息
↓
[tool_calling_llm] ← LLM 分析并决策
↓
(条件判断)
↓
/ \
/ \
↓ ↓
直接回答 [tools] 调用工具
↓ ↓
END END
关键设计:
- 一个节点:
tool_calling_llm
(绑定了工具的 LLM) - 一个条件边:
tools_condition
(自动判断是否需要工具) - 一个工具节点:
tools
(执行实际计算)
🔧 代码实现详解
1. 环境准备
# 安装依赖
%pip install --quiet -U langchain_openai langchain_core langgraph langgraph-prebuilt
# 设置 API Key
import os, getpass
def _set_env(var: str):
if not os.environ.get(var):
os.environ[var] = getpass.getpass(f"{var}: ")
_set_env("OPENAI_API_KEY")
说明:
langgraph-prebuilt
:包含ToolNode
和tools_condition
等预构建组件getpass
:安全地输入敏感信息(API Key 不会显示在屏幕上)
2. 定义工具和 LLM
from langchain_openai import ChatOpenAI
def multiply(a: int, b: int) -> int:
"""Multiply a and b.
Args:
a: first int
b: second int
"""
return a * b
llm = ChatOpenAI(model="gpt-4o")
llm_with_tools = llm.bind_tools([multiply])
Python 知识点:函数作为工具
在 LangChain 中,普通 Python 函数可以直接转换为 LLM 工具:
关键要素:
- 函数名:LLM 用来识别工具(如
multiply
) - Docstring:LLM 用来理解工具的用途(非常重要!)
- 类型提示:LLM 用来理解参数类型(
a: int, b: int
) - 返回类型:LLM 用来理解输出类型(
-> int
)
示例:
def multiply(a: int, b: int) -> int:
"""Multiply a and b.""" # ← LLM 看到这个描述
return a * b
# LLM 内部理解为:
# {
# "name": "multiply",
# "description": "Multiply a and b.",
# "parameters": {
# "a": {"type": "integer"},
# "b": {"type": "integer"}
# }
# }
LangChain 知识点:bind_tools
llm_with_tools = llm.bind_tools([multiply])
作用:
- 将工具"绑定"到 LLM,让 LLM 知道可以调用这些工具
- LLM 不会自动执行工具,只会返回"我想调用这个工具"的指令
- 实际执行由
ToolNode
完成
工作流程:
用户: "2 乘以 3 是多少?"
↓
llm_with_tools.invoke(...)
↓
LLM 返回:AIMessage(tool_calls=[{
"name": "multiply",
"args": {"a": 2, "b": 3},
"id": "call_123"
}])
↓
ToolNode 执行 multiply(2, 3)
↓
返回:ToolMessage(content="6", tool_call_id="call_123")
3. 定义状态
from langgraph.graph import MessagesState
# MessagesState 是预定义的状态类
# 等价于:
# class MessagesState(TypedDict):
# messages: Annotated[list[BaseMessage], add_messages]
LangGraph 知识点:MessagesState
MessagesState
是 LangGraph 提供的标准消息状态:
特性:
messages
字段:存储对话历史- 自动使用
add_messages
reducer:智能合并新消息
add_messages 的智能之处:
# 场景 1:追加新消息
state = {"messages": [HumanMessage("你好")]}
update = {"messages": [AIMessage("你好!")]}
# 结果:[HumanMessage("你好"), AIMessage("你好!")]
# 场景 2:更新现有消息(通过 ID)
state = {"messages": [AIMessage("思考中...", id="msg1")]}
update = {"messages": [AIMessage("2 * 3 = 6", id="msg1")]}
# 结果:[AIMessage("2 * 3 = 6", id="msg1")] # 替换而不是追加
4. 构建核心节点
def tool_calling_llm(state: MessagesState):
return {"messages": [llm_with_tools.invoke(state["messages"])]}
功能详解:
- 输入:
state["messages"]
- 当前对话历史 - 处理:
llm_with_tools.invoke(...)
- LLM 分析并决策 - 输出:
{"messages": [...]}
- 返回 LLM 的响应
可能的输出类型:
情况 A:直接回答(无工具调用)
AIMessage(content="你好!我能帮你什么?")
情况 B:调用工具
AIMessage(
content="", # 内容为空
tool_calls=[{
"name": "multiply",
"args": {"a": 2, "b": 3},
"id": "call_abc123"
}]
)
5. 使用预构建组件:ToolNode 和 tools_condition
from langgraph.prebuilt import ToolNode, tools_condition
# 创建工具节点
tool_node = ToolNode([multiply])
# tools_condition 是预构建的条件函数
LangGraph 知识点:ToolNode
ToolNode 的作用:
- 自动执行 LLM 请求的工具调用
- 接收带有
tool_calls
的AIMessage
- 返回
ToolMessage
包含执行结果
内部实现逻辑:
# ToolNode 内部类似这样:
def tool_node(state: MessagesState):
results = []
last_message = state["messages"][-1]
for tool_call in last_message.tool_calls:
# 找到对应的工具函数
tool = find_tool(tool_call["name"]) # 如 multiply
# 执行工具
result = tool(**tool_call["args"]) # multiply(a=2, b=3)
# 创建 ToolMessage
results.append(ToolMessage(
content=str(result), # "6"
tool_call_id=tool_call["id"]
))
return {"messages": results}
LangGraph 知识点:tools_condition
tools_condition 的作用:
- 自动检查最后一条消息是否包含
tool_calls
- 如果有工具调用 → 路由到工具节点
- 如果无工具调用 → 路由到 END
内部实现逻辑:
# tools_condition 内部类似这样:
def tools_condition(state: MessagesState):
last_message = state["messages"][-1]
# 检查是否有工具调用
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools" # 路由到工具节点
else:
return END # 直接结束
6. 构建图
from langgraph.graph import StateGraph, START, END
# 创建图
builder = StateGraph(MessagesState)
# 添加节点
builder.add_node("tool_calling_llm", tool_calling_llm)
builder.add_node("tools", ToolNode([multiply]))
# 添加边
builder.add_edge(START, "tool_calling_llm")
# 添加条件边
builder.add_conditional_edges(
"tool_calling_llm",
tools_condition,
)
builder.add_edge("tools", END)
# 编译
graph = builder.compile()
LangGraph 知识点:条件边详解
builder.add_conditional_edges(
"tool_calling_llm", # 源节点
tools_condition, # 条件函数
)
工作原理:
tool_calling_llm
节点执行完毕- 调用
tools_condition(state)
判断路由 tools_condition
返回值决定下一步:- 返回
"tools"
→ 进入tools
节点 - 返回
END
→ 结束执行
- 返回
完整执行流程:
场景 A:简单问候
用户: "你好"
↓
[tool_calling_llm] → AIMessage(content="你好!")
↓
tools_condition 检测:无 tool_calls
↓
路由到 END
场景 B:数学计算
用户: "2 乘以 3 是多少?"
↓
[tool_calling_llm] → AIMessage(tool_calls=[multiply(2,3)])
↓
tools_condition 检测:有 tool_calls
↓
路由到 [tools]
↓
[tools] 执行 multiply(2,3) → ToolMessage(content="6")
↓
END
图结构可视化:
START
↓
[tool_calling_llm]
↓
(条件判断)
/ \
/ \
↓ ↓
END [tools]
↓
END
7. 执行图
测试场景 1:工具调用
from langchain_core.messages import HumanMessage
messages = [HumanMessage(content="Hello, what is 2 multiplied by 2?")]
result = graph.invoke({"messages": messages})
for m in result['messages']:
m.pretty_print()
输出:
================================ Human Message =================================
Hello, what is 2 multiplied by 2?
================================== Ai Message ==================================
Tool Calls:
multiply (call_abc123)
Call ID: call_abc123
Args:
a: 2
b: 2
================================= Tool Message =================================
Name: multiply
4
执行流程分析:
- HumanMessage 进入系统
- AIMessage 返回工具调用请求
- ToolMessage 返回计算结果
4
测试场景 2:直接回答
messages = [HumanMessage(content="Hello world.")]
result = graph.invoke({"messages": messages})
for m in result['messages']:
m.pretty_print()
输出:
================================ Human Message =================================
Hello world.
================================== Ai Message ==================================
Hello! How can I assist you today?
执行流程分析:
- HumanMessage 进入系统
- AIMessage 直接回答,无工具调用
- 直接到达 END
🎓 核心知识点总结
LangGraph 特有概念
1. Router 模式
定义: LLM 作为决策中心,动态选择执行路径
实现方式:
- 节点:
tool_calling_llm
(绑定工具的 LLM) - 条件边:
tools_condition
(自动路由) - 工具节点:
ToolNode
(执行工具)
对比传统流程:
特性 | 传统流程 | Router 模式 |
---|---|---|
路径决策 | 硬编码规则 | LLM 智能判断 |
灵活性 | 低 | 高 |
可扩展性 | 差(需修改代码) | 强(增加工具即可) |
适应性 | 固定场景 | 动态适应 |
2. 条件边(Conditional Edge)
builder.add_conditional_edges(
source_node, # 源节点
condition_func, # 条件函数:state → 目标节点名称
)
条件函数的返回值:
- 字符串(如
"tools"
):路由到指定节点 END
:结束执行- 列表(高级用法):多个可能的目标
示例:自定义条件函数
def my_condition(state: MessagesState):
last_msg = state["messages"][-1]
if last_msg.content == "结束":
return END
elif "计算" in last_msg.content:
return "calculator"
else:
return "chat"
builder.add_conditional_edges("start", my_condition)
3. ToolNode
作用: 自动执行工具并返回结果
关键特性:
- 接收
AIMessage
中的tool_calls
- 自动匹配和执行对应工具
- 返回
ToolMessage
包含结果 - 支持多个工具并发执行
手动实现 ToolNode 的等价代码:
def manual_tool_node(state: MessagesState):
last_message = state["messages"][-1]
tool_calls = last_message.tool_calls
tool_messages = []
for call in tool_calls:
if call["name"] == "multiply":
result = multiply(**call["args"])
tool_messages.append(ToolMessage(
content=str(result),
tool_call_id=call["id"]
))
return {"messages": tool_messages}
4. tools_condition
作用: 标准化的工具调用检测函数
等价实现:
def tools_condition(state: MessagesState):
last_msg = state["messages"][-1]
if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
return "tools"
return END
为什么使用预构建函数?
- 经过充分测试
- 处理边缘情况(如空消息、格式错误等)
- 保持代码简洁
Python 特有知识点
1. 类型提示在工具中的作用
def multiply(a: int, b: int) -> int:
"""Multiply a and b."""
return a * b
类型提示的重要性:
- LLM 根据类型提示生成正确的参数
- 运行时验证(配合 Pydantic)
- IDE 代码补全和检查
对比:
# ❌ 不好的实践
def multiply(a, b): # 没有类型提示
return a * b
# ✅ 好的实践
def multiply(a: int, b: int) -> int:
"""Multiply a and b."""
return a * b
2. Docstring 的关键作用
Docstring 是 LLM 理解工具的唯一途径!
def multiply(a: int, b: int) -> int:
"""Multiply a and b.
Args:
a: first int
b: second int
"""
return a * b
LLM 会读取:
- 第一行:工具的简短描述("Multiply a and b.")
- Args 部分:参数的详细说明
示例:好的 vs 不好的 Docstring
# ❌ 不好
def process_data(x, y):
"""Process data.""" # 太模糊
return x + y
# ✅ 好
def calculate_total_price(unit_price: float, quantity: int) -> float:
"""Calculate the total price by multiplying unit price and quantity.
Args:
unit_price: Price per unit in USD
quantity: Number of units to purchase
Returns:
Total price in USD
"""
return unit_price * quantity
3. getpass 模块
import getpass
os.environ["OPENAI_API_KEY"] = getpass.getpass("OPENAI_API_KEY: ")
作用:
- 安全地输入敏感信息
- 输入时不会显示在屏幕上(类似输入密码)
- 防止 API Key 被意外泄露
💡 最佳实践
1. 设计良好的工具
原则 1:单一职责
# ✅ 好的设计 - 每个工具只做一件事
def multiply(a: int, b: int) -> int:
"""Multiply two numbers."""
return a * b
def add(a: int, b: int) -> int:
"""Add two numbers."""
return a + b
# ❌ 不好的设计 - 一个工具做多件事
def calculate(operation: str, a: int, b: int) -> int:
"""Perform various calculations."""
if operation == "multiply":
return a * b
elif operation == "add":
return a + b
为什么? LLM 更容易理解和选择专注的工具。
原则 2:清晰的命名和文档
# ✅ 好
def get_current_weather(location: str, unit: str = "celsius") -> dict:
"""Get the current weather for a specific location.
Args:
location: City name or zip code
unit: Temperature unit ('celsius' or 'fahrenheit')
Returns:
Dictionary with temperature, humidity, and conditions
"""
pass
# ❌ 不好
def get_data(loc: str, u: str = "c") -> dict:
"""Get data."""
pass
原则 3:返回结构化数据
# ✅ 好 - 返回结构化数据
def multiply(a: int, b: int) -> int:
return a * b # 返回数字
# 或者
def get_user_info(user_id: str) -> dict:
return {
"name": "Alice",
"age": 30,
"email": "alice@example.com"
}
# ❌ 不好 - 返回非结构化字符串
def get_user_info(user_id: str) -> str:
return "Name: Alice, Age: 30, Email: alice@example.com"
2. 条件边的设计技巧
技巧 1:清晰的条件逻辑
# ✅ 好 - 逻辑清晰
def route_decision(state: MessagesState):
last_msg = state["messages"][-1]
# 明确的条件检查
if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
return "tools"
return END
# ❌ 不好 - 逻辑复杂
def route_decision(state: MessagesState):
return "tools" if state["messages"][-1].tool_calls if hasattr(state["messages"][-1], "tool_calls") else False else END
技巧 2:处理边缘情况
def safe_route_decision(state: MessagesState):
# 检查消息列表是否为空
if not state["messages"]:
return END
last_msg = state["messages"][-1]
# 检查是否为 AI 消息
if not isinstance(last_msg, AIMessage):
return END
# 检查是否有工具调用
if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
return "tools"
return END
3. 状态管理最佳实践
实践 1:保持消息历史完整
# ✅ 好 - 保留完整对话历史
def tool_calling_llm(state: MessagesState):
# 传递所有历史消息
return {"messages": [llm_with_tools.invoke(state["messages"])]}
# ❌ 不好 - 只使用最后一条消息
def tool_calling_llm(state: MessagesState):
# 丢失了上下文
return {"messages": [llm_with_tools.invoke([state["messages"][-1]])]}
为什么重要? LLM 需要完整上下文才能做出正确决策。
实践 2:正确返回消息列表
# ✅ 正确 - 返回列表
def my_node(state: MessagesState):
response = llm.invoke(...)
return {"messages": [response]} # 注意是列表
# ❌ 错误 - 返回单个消息
def my_node(state: MessagesState):
response = llm.invoke(...)
return {"messages": response} # 会出错
🚀 进阶技巧
1. 多工具 Router
扩展到多个工具:
def add(a: int, b: int) -> int:
"""Add two numbers."""
return a + b
def multiply(a: int, b: int) -> int:
"""Multiply two numbers."""
return a * b
def divide(a: float, b: float) -> float:
"""Divide a by b."""
if b == 0:
raise ValueError("Cannot divide by zero")
return a / b
# 绑定多个工具
llm_with_tools = llm.bind_tools([add, multiply, divide])
# ToolNode 自动处理所有工具
tool_node = ToolNode([add, multiply, divide])
LLM 自动选择正确工具:
- "2 加 3" → 调用
add
- "4 乘以 5" → 调用
multiply
- "10 除以 2" → 调用
divide
2. 工具链式调用
LLM 可以连续调用多个工具:
# 用户:"先计算 2*3,然后把结果加 5"
# 第一轮
# LLM → multiply(2, 3)
# Tool → 6
# 第二轮(LLM 看到结果 6)
# LLM → add(6, 5)
# Tool → 11
# 第三轮(LLM 看到结果 11)
# LLM → "结果是 11"
实现方式: 添加循环边
# 修改图结构
builder.add_edge("tools", "tool_calling_llm") # 工具执行后回到 LLM
# 现在流程变成:
# START → tool_calling_llm → tools → tool_calling_llm → tools → ... → END
3. 带错误处理的 Router
class SafeToolNode:
def __init__(self, tools):
self.tool_node = ToolNode(tools)
def __call__(self, state: MessagesState):
try:
return self.tool_node(state)
except Exception as e:
# 返回错误消息
return {"messages": [ToolMessage(
content=f"Error executing tool: {str(e)}",
tool_call_id=state["messages"][-1].tool_calls[0]["id"]
)]}
# 使用
builder.add_node("tools", SafeToolNode([multiply]))
4. 自定义工具执行逻辑
def custom_tool_node(state: MessagesState):
last_message = state["messages"][-1]
tool_calls = last_message.tool_calls
results = []
for call in tool_calls:
# 记录工具调用
print(f"Calling tool: {call['name']} with args: {call['args']}")
# 执行工具
if call["name"] == "multiply":
result = multiply(**call["args"])
# 添加额外信息
results.append(ToolMessage(
content=f"Result: {result} (computed at {datetime.now()})",
tool_call_id=call["id"]
))
return {"messages": results}
📊 Router 模式对比
Router vs 简单 LLM 调用
特性 | 简单 LLM 调用 | Router 模式 |
---|---|---|
工具支持 | 无 | 有 |
动态决策 | 无 | 有 |
可扩展性 | 低 | 高 |
复杂度 | 低 | 中 |
适用场景 | 纯对话 | 需要工具的场景 |
Router vs ReAct Agent
特性 | Router | ReAct Agent |
---|---|---|
循环执行 | 无(单次) | 有(多轮推理) |
复杂度 | 简单 | 复杂 |
适用任务 | 单步工具调用 | 多步推理任务 |
状态管理 | 简单 | 复杂 |
Router 是 ReAct 的基础:
- ReAct = Router + 循环 + 推理状态
🎯 实际应用案例
案例 1:智能客服
def search_knowledge_base(query: str) -> str:
"""Search internal knowledge base for answers."""
# 实现知识库搜索
pass
def create_ticket(issue: str, priority: str) -> str:
"""Create a support ticket."""
# 实现工单创建
pass
llm_with_tools = llm.bind_tools([search_knowledge_base, create_ticket])
# 用户问题:
# "如何重置密码?" → search_knowledge_base
# "我的账号被锁定了,请帮我处理" → create_ticket
# "你好" → 直接回答
案例 2:数据分析助手
def query_database(sql: str) -> list:
"""Execute SQL query on the database."""
pass
def generate_chart(data: list, chart_type: str) -> str:
"""Generate a chart from data."""
pass
def calculate_statistics(data: list) -> dict:
"""Calculate basic statistics."""
pass
llm_with_tools = llm.bind_tools([query_database, generate_chart, calculate_statistics])
# 用户请求:
# "查询上个月的销售数据" → query_database
# "用这些数据生成柱状图" → generate_chart
# "计算平均值和标准差" → calculate_statistics
案例 3:多语言翻译工具
def translate_text(text: str, target_language: str) -> str:
"""Translate text to target language."""
pass
def detect_language(text: str) -> str:
"""Detect the language of the text."""
pass
llm_with_tools = llm.bind_tools([translate_text, detect_language])
# 用户输入:
# "Translate 'hello' to Chinese" → translate_text
# "What language is 'Bonjour'?" → detect_language
# "How do I say goodbye in Spanish?" → 直接回答(LLM 知识)
🔍 常见问题
Q1: 为什么要使用 bind_tools 而不是直接调用函数?
答: bind_tools
让 LLM 知道可用的工具,但不会自动执行:
# 错误理解 ❌
llm_with_tools.invoke(...) # LLM 不会自动执行 multiply
# 正确理解 ✅
llm_with_tools.invoke(...) # LLM 返回"我想调用 multiply"
# 然后由 ToolNode 实际执行
这样设计的好处:
- LLM 负责决策(调用哪个工具、传什么参数)
- 系统负责执行(安全、可控)
- 清晰的责任分离
Q2: tools_condition 如何知道路由到哪里?
答: 它检查最后一条消息:
def tools_condition(state):
last_msg = state["messages"][-1]
# 如果有 tool_calls 属性且不为空
if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
return "tools" # 字符串 "tools" 匹配节点名称
return END
关键点:
- 返回值
"tools"
必须匹配某个已添加的节点名称 - 如果返回不存在的节点名,会报错
Q3: 如果 LLM 调用了不存在的工具会怎样?
答: ToolNode 会抛出错误:
# 如果只绑定了 multiply
llm_with_tools = llm.bind_tools([multiply])
# 但 LLM 尝试调用 divide
AIMessage(tool_calls=[{"name": "divide", ...}])
# ToolNode 会报错:Tool 'divide' not found
解决方案:
- 确保绑定的工具和 ToolNode 中的工具一致
- 使用错误处理包装 ToolNode
Q4: MessagesState 中的消息会一直增长吗?
答: 是的,默认情况下消息会持续累积:
# 第 1 轮
messages = [HumanMessage("你好")]
# 第 2 轮
messages = [
HumanMessage("你好"),
AIMessage("你好!"),
HumanMessage("2*3=?"),
]
# 第 3 轮
messages = [
HumanMessage("你好"),
AIMessage("你好!"),
HumanMessage("2*3=?"),
AIMessage(tool_calls=[...]),
ToolMessage("6"),
]
管理消息历史:
# 方法 1:限制消息数量
def trim_messages(state: MessagesState):
# 只保留最近 10 条消息
return {"messages": state["messages"][-10:]}
# 方法 2:使用 LangChain 的 trim_messages
from langchain_core.messages import trim_messages
trimmed = trim_messages(
messages,
max_tokens=1000,
strategy="last",
token_counter=llm
)
📖 扩展阅读
🎓 学习路径
掌握 Router 后,建议按以下顺序深入学习:
当前阶段:Router ✅
- 理解条件边
- 掌握工具调用
- 熟悉 MessagesState
下一步:Agent with Memory
- 添加记忆功能
- 多轮对话
- 状态持久化
进阶:ReAct Agent
- 循环推理
- 多步规划
- 复杂任务分解
高级:Multi-Agent System
- 多个 Agent 协作
- 任务分配
- 并行执行
总结:Router 是 LangGraph Agent 系统的基石。通过 LLM 智能决策 + 条件边动态路由,我们可以构建能够自主选择工具的智能系统。这是从简单聊天机器人到强大 AI Agent 的关键一步!