Skip to content

LangGraph Router 详细解读

📚 概述

本文档详细解读 LangGraph 中的 Router(路由器) 模式。这是 Agent 系统的基础概念,通过 LLM 智能决定执行路径:直接响应用户或调用工具。Router 是理解 LangGraph 条件边和动态控制流的关键入门案例。

🎯 核心概念

什么是 Router?

Router 是一种设计模式,让 LLM 充当"交通指挥官":

  1. 接收输入 - 用户发送查询
  2. 智能判断 - LLM 分析是否需要工具
  3. 动态路由 - 自动选择执行路径:
    • 路径 A:直接回答(无需工具)
    • 路径 B:调用工具获取信息再回答

为什么 Router 重要?

这是 Agent 的本质:

  • Agent ≠ 简单的函数调用
  • Agent = LLM 自主决策 + 动态执行路径
  • Router 是最简单但最核心的 Agent 模式

经典应用场景

  • 智能客服:简单问题直接回答,复杂问题查询数据库
  • 数学助手:文字问题直接答,计算问题调用计算器
  • 信息查询:基础知识直接答,实时信息调用搜索 API
  • 多功能 Bot:根据意图自动选择功能模块

🎭 实战案例:智能数学助手

我们将构建一个会"思考"的数学助手,演示 Router 的完整流程:

需求:

  1. 用户问"你好" → LLM 直接回答,无需工具
  2. 用户问"2 乘以 3 是多少?" → LLM 调用 multiply 工具计算

系统架构图

用户输入消息

[tool_calling_llm]  ← LLM 分析并决策

  (条件判断)

    / \
   /   \
  ↓     ↓
直接回答  [tools] 调用工具
  ↓     ↓
  END   END

关键设计:

  • 一个节点tool_calling_llm(绑定了工具的 LLM)
  • 一个条件边tools_condition(自动判断是否需要工具)
  • 一个工具节点tools(执行实际计算)

🔧 代码实现详解

1. 环境准备

python
# 安装依赖
%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:包含 ToolNodetools_condition 等预构建组件
  • getpass:安全地输入敏感信息(API Key 不会显示在屏幕上)

2. 定义工具和 LLM

python
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 工具:

关键要素:

  1. 函数名:LLM 用来识别工具(如 multiply
  2. Docstring:LLM 用来理解工具的用途(非常重要!)
  3. 类型提示:LLM 用来理解参数类型(a: int, b: int
  4. 返回类型:LLM 用来理解输出类型(-> int

示例:

python
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

python
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. 定义状态

python
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 的智能之处:

python
# 场景 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. 构建核心节点

python
def tool_calling_llm(state: MessagesState):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

功能详解:

  1. 输入state["messages"] - 当前对话历史
  2. 处理llm_with_tools.invoke(...) - LLM 分析并决策
  3. 输出{"messages": [...]} - 返回 LLM 的响应

可能的输出类型:

情况 A:直接回答(无工具调用)

python
AIMessage(content="你好!我能帮你什么?")

情况 B:调用工具

python
AIMessage(
    content="",  # 内容为空
    tool_calls=[{
        "name": "multiply",
        "args": {"a": 2, "b": 3},
        "id": "call_abc123"
    }]
)

5. 使用预构建组件:ToolNode 和 tools_condition

python
from langgraph.prebuilt import ToolNode, tools_condition

# 创建工具节点
tool_node = ToolNode([multiply])

# tools_condition 是预构建的条件函数

LangGraph 知识点:ToolNode

ToolNode 的作用:

  • 自动执行 LLM 请求的工具调用
  • 接收带有 tool_callsAIMessage
  • 返回 ToolMessage 包含执行结果

内部实现逻辑:

python
# 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

内部实现逻辑:

python
# 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. 构建图

python
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 知识点:条件边详解

python
builder.add_conditional_edges(
    "tool_calling_llm",  # 源节点
    tools_condition,      # 条件函数
)

工作原理:

  1. tool_calling_llm 节点执行完毕
  2. 调用 tools_condition(state) 判断路由
  3. 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:工具调用

python
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

执行流程分析:

  1. HumanMessage 进入系统
  2. AIMessage 返回工具调用请求
  3. ToolMessage 返回计算结果 4

测试场景 2:直接回答

python
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?

执行流程分析:

  1. HumanMessage 进入系统
  2. AIMessage 直接回答,无工具调用
  3. 直接到达 END

🎓 核心知识点总结

LangGraph 特有概念

1. Router 模式

定义: LLM 作为决策中心,动态选择执行路径

实现方式:

  • 节点tool_calling_llm(绑定工具的 LLM)
  • 条件边tools_condition(自动路由)
  • 工具节点ToolNode(执行工具)

对比传统流程:

特性传统流程Router 模式
路径决策硬编码规则LLM 智能判断
灵活性
可扩展性差(需修改代码)强(增加工具即可)
适应性固定场景动态适应

2. 条件边(Conditional Edge)

python
builder.add_conditional_edges(
    source_node,      # 源节点
    condition_func,   # 条件函数:state → 目标节点名称
)

条件函数的返回值:

  • 字符串(如 "tools"):路由到指定节点
  • END:结束执行
  • 列表(高级用法):多个可能的目标

示例:自定义条件函数

python
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 的等价代码:

python
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

作用: 标准化的工具调用检测函数

等价实现:

python
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. 类型提示在工具中的作用

python
def multiply(a: int, b: int) -> int:
    """Multiply a and b."""
    return a * b

类型提示的重要性:

  • LLM 根据类型提示生成正确的参数
  • 运行时验证(配合 Pydantic)
  • IDE 代码补全和检查

对比:

python
# ❌ 不好的实践
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 理解工具的唯一途径!

python
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

python
# ❌ 不好
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 模块

python
import getpass
os.environ["OPENAI_API_KEY"] = getpass.getpass("OPENAI_API_KEY: ")

作用:

  • 安全地输入敏感信息
  • 输入时不会显示在屏幕上(类似输入密码)
  • 防止 API Key 被意外泄露

💡 最佳实践

1. 设计良好的工具

原则 1:单一职责

python
# ✅ 好的设计 - 每个工具只做一件事
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:清晰的命名和文档

python
# ✅ 好
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:返回结构化数据

python
# ✅ 好 - 返回结构化数据
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:清晰的条件逻辑

python
# ✅ 好 - 逻辑清晰
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:处理边缘情况

python
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:保持消息历史完整

python
# ✅ 好 - 保留完整对话历史
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:正确返回消息列表

python
# ✅ 正确 - 返回列表
def my_node(state: MessagesState):
    response = llm.invoke(...)
    return {"messages": [response]}  # 注意是列表

# ❌ 错误 - 返回单个消息
def my_node(state: MessagesState):
    response = llm.invoke(...)
    return {"messages": response}  # 会出错

🚀 进阶技巧

1. 多工具 Router

扩展到多个工具:

python
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 可以连续调用多个工具:

python
# 用户:"先计算 2*3,然后把结果加 5"

# 第一轮
# LLM → multiply(2, 3)
# Tool → 6

# 第二轮(LLM 看到结果 6)
# LLM → add(6, 5)
# Tool → 11

# 第三轮(LLM 看到结果 11)
# LLM → "结果是 11"

实现方式: 添加循环边

python
# 修改图结构
builder.add_edge("tools", "tool_calling_llm")  # 工具执行后回到 LLM

# 现在流程变成:
# START → tool_calling_llm → tools → tool_calling_llm → tools → ... → END

3. 带错误处理的 Router

python
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. 自定义工具执行逻辑

python
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

特性RouterReAct Agent
循环执行无(单次)有(多轮推理)
复杂度简单复杂
适用任务单步工具调用多步推理任务
状态管理简单复杂

Router 是 ReAct 的基础:

  • ReAct = Router + 循环 + 推理状态

🎯 实际应用案例

案例 1:智能客服

python
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:数据分析助手

python
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:多语言翻译工具

python
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 知道可用的工具,但不会自动执行:

python
# 错误理解 ❌
llm_with_tools.invoke(...)  # LLM 不会自动执行 multiply

# 正确理解 ✅
llm_with_tools.invoke(...)  # LLM 返回"我想调用 multiply"
# 然后由 ToolNode 实际执行

这样设计的好处:

  • LLM 负责决策(调用哪个工具、传什么参数)
  • 系统负责执行(安全、可控)
  • 清晰的责任分离

Q2: tools_condition 如何知道路由到哪里?

答: 它检查最后一条消息:

python
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 会抛出错误:

python
# 如果只绑定了 multiply
llm_with_tools = llm.bind_tools([multiply])

# 但 LLM 尝试调用 divide
AIMessage(tool_calls=[{"name": "divide", ...}])

# ToolNode 会报错:Tool 'divide' not found

解决方案:

  1. 确保绑定的工具和 ToolNode 中的工具一致
  2. 使用错误处理包装 ToolNode

Q4: MessagesState 中的消息会一直增长吗?

答: 是的,默认情况下消息会持续累积:

python
# 第 1 轮
messages = [HumanMessage("你好")]

# 第 2 轮
messages = [
    HumanMessage("你好"),
    AIMessage("你好!"),
    HumanMessage("2*3=?"),
]

# 第 3 轮
messages = [
    HumanMessage("你好"),
    AIMessage("你好!"),
    HumanMessage("2*3=?"),
    AIMessage(tool_calls=[...]),
    ToolMessage("6"),
]

管理消息历史:

python
# 方法 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 后,建议按以下顺序深入学习:

  1. 当前阶段:Router

    • 理解条件边
    • 掌握工具调用
    • 熟悉 MessagesState
  2. 下一步:Agent with Memory

    • 添加记忆功能
    • 多轮对话
    • 状态持久化
  3. 进阶:ReAct Agent

    • 循环推理
    • 多步规划
    • 复杂任务分解
  4. 高级:Multi-Agent System

    • 多个 Agent 协作
    • 任务分配
    • 并行执行

总结:Router 是 LangGraph Agent 系统的基石。通过 LLM 智能决策 + 条件边动态路由,我们可以构建能够自主选择工具的智能系统。这是从简单聊天机器人到强大 AI Agent 的关键一步!

基于 MIT 许可证发布。内容版权归作者所有。