基于本文回答

播面 播面

刷题像听歌，多听自然懂
0
评论
请从左侧选择一个分类

如何编写一个路由函数（Router/Condition function）来决定条件边的走向？

知识点图片

编写一个用于决定条件边（Conditional Edge）走向的路由函数（Router/Condition Function），核心逻辑其实非常简单：接收当前的工作流状态（State），通过预设的逻辑判断，返回下一个需要执行的节点名称。

在目前最流行的基于图的 AI 工作流框架（如 LangGraph）中，路由函数扮演着“交通警察”的角色。

下面我将分步骤教你如何编写，并提供几种最常见的路由模式和最佳实践。

1. 路由函数的基本结构

路由函数本质上是一个普通的 Python 函数，它包含三个步骤：

  1. 读取状态：从传入的 state 中提取关键信息。
  2. 逻辑判断：使用 if-else 或其他条件语句进行判断。
  3. 返回路径：返回目标节点的名字（字符串）。
python 
def my_router(state: dict) -> str:
    # 1. 提取状态信息
    current_value = state.get("some_key")
    
    # 2. 逻辑判断
    if current_value == "condition_A":
        # 3. 返回目标节点名称
        return "node_a"
    else:
        return "node_b"

2. 常见的路由场景与代码示例

场景一：判断是否调用工具 (AI Agent 最常用的模式)

Agent 生成回复后，我们需要判断它是想直接回复用户（结束），还是想调用外部工具。

python 
from typing import Literal
from langgraph.graph import END

def should_continue(state: dict) -> Literal["tools", "__end__"]:
    """判断是否需要执行工具节点"""
    messages = state["messages"]
    last_message = messages[-1]
    
    # 如果 LLM 返回了工具调用请求 (tool_calls)
    if last_message.tool_calls:
        return "tools" # 走向名为 "tools" 的节点
    
    # 否则，结束流程
    return END # 或者返回 "__end__"

场景二：基于分类/意图的多分支路由

在客服系统中，先由一个节点判断用户意图，然后路由到专门的处理节点。

python 
def route_by_intent(state: dict) -> str:
    """根据意图分发到不同的处理流程"""
    intent = state.get("user_intent", "unknown")
    
    if intent == "billing":
        return "billing_support_node"
    elif intent == "technical":
        return "tech_support_node"
    elif intent == "refund":
        return "refund_processing_node"
    else:
        # 必须提供一个默认/兜底分支
        return "human_fallback_node"

场景三：循环控制 (如重试机制)

当某个操作失败时，决定是重试还是放弃。

python 
def retry_router(state: dict) -> str:
    error_count = state.get("error_count", 0)
    
    if error_count < 3:
        return "retry_node" # 继续重试
    else:
        return "fail_node"  # 超过最大重试次数，走向失败处理节点

3. 如何在图中配置这个路由函数？

写好路由函数后，你需要将它添加到图（Graph）的“条件边”中。以 LangGraph 为例：

python 
from langgraph.graph import StateGraph, END

# 初始化图...
builder = StateGraph(MyState)
# 添加节点...
builder.add_node("agent", agent_node)
builder.add_node("tools", tool_node)

# 【核心配置：添加条件边】
# 参数1: 起始节点名称
# 参数2: 你编写的路由函数
# 参数3(可选): 路由映射字典 (将函数的返回值映射到真实的节点名称)
builder.add_conditional_edges(
    "agent",                 # 从 agent 节点出来后
    should_continue,         # 执行 should_continue 函数
    {
        "tools": "tools",    # 如果函数返回 "tools"，就去 "tools" 节点
        END: END             # 如果函数返回 END，就结束
    }
)

注：在最新版的 LangGraph 中，如果你的路由函数直接返回的就是确切的节点名称，第三个参数（映射字典）是可以省略的。

4. 编写路由函数的 4 个“最佳实践” (避坑指南)

  1. 保持纯函数特性（Pure Function）：
    绝对不要在路由函数内部去修改 state 的值，也不要在里面进行耗时的操作（比如请求数据库、调用 LLM）。路由函数只负责“读”和“指路”。

    • 错误做法：在路由函数里调用大模型判断意图。
    • 正确做法：用一个普通节点调用大模型，把意图存入 state["intent"]，然后路由函数只读取 state["intent"] 进行 if-else 判断。

  2. 必须要有兜底分支（Fallback）：
    永远写上 else。哪怕你确信变量只有 A 和 B 两种状态，也要写一个 else 处理异常情况，防止工作流因为返回了不存在的节点名称而崩溃。

  3. 使用类型提示（Type Hinting）：
    像上面的例子一样，使用 from typing import Literal 并标注 -> Literal["node_a", "node_b"]。这不仅能让代码更清晰，IDE 还能帮你在写配置时检查拼写错误。

  4. 高级用法：动态路由（并行执行）：
    如果你使用的是 LangGraph，路由函数还可以返回一个特殊的 Send("node_name", {state_args}) 对象（或 Send 对象的列表），这被称作 Map-Reduce 模式，可以触发下游节点的并行多开执行。适用于需要同时处理多个独立任务的场景。

00:00
00:00