Skip to content

条件路由

条件路由(Conditional Routing)是 LangGraph 最强大的特性之一。它让图能够根据当前状态动态决定下一步执行哪个节点——这正是 Agent 行为的基础。

基本用法

条件边通过 add_conditional_edges 方法定义:

python
from langgraph.graph import StateGraph, START, END
from typing import Literal

def route(state: State) -> Literal["node_a", "node_b"]:
    """根据状态决定路由"""
    if state["should_do_a"]:
        return "node_a"
    return "node_b"

builder.add_conditional_edges(
    "decision_node",  # 起始节点
    route,            # 路由函数
    {                 # 路由映射
        "node_a": "node_a",
        "node_b": "node_b",
    }
)

路由函数的返回值

路由函数必须返回一个字符串(或字符串列表),指向要执行的节点名称。返回值必须包含在路由映射中。

常见的路由模式

1. Tool Calling 路由

最基本的 Agent 模式:根据 LLM 是否调用了工具来决定下一步:

python
from typing import Literal

def should_continue(state: MessagesState) -> Literal["tools", END]:
    """如果 LLM 调用了工具,就执行工具节点"""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

builder.add_conditional_edges(
    "llm",
    should_continue,
    {
        "tools": "tools",
        END: END,
    }
)

2. 多分支分类路由

python
from typing import Literal

def classify_intent(state: State) -> Literal["search", "calculate", "chat"]:
    query = state["user_input"].lower()

    if "搜索" in query or "查找" in query:
        return "search"
    elif "计算" in query or "多少" in query:
        return "calculate"
    else:
        return "chat"

builder.add_conditional_edges(
    "classifier",
    classify_intent,
    {
        "search": "search_node",
        "calculate": "calculate_node",
        "chat": "chat_node",
    }
)

3. 质量检查路由(自修正)

python
def quality_check(state: State) -> Literal["regenerate", END]:
    """检查输出质量"""
    output = state["output"]

    if len(output) < 10:
        return "regenerate"  # 内容太短,重新生成
    if state.get("retry_count", 0) > 3:
        return END  # 重试太多次了,放弃

    return END

builder.add_conditional_edges(
    "generate",
    quality_check,
    {
        "regenerate": "generate",  # 回到生成节点
        END: END,
    }
)

4. 基于路由做并行

路由函数也可以返回列表,实现动态并行

python
def route_to_nodes(state: State) -> list[str]:
    """根据输入分发给对应的处理节点"""
    nodes = []
    if state.get("needs_search"):
        nodes.append("search_node")
    if state.get("needs_calc"):
        nodes.append("calculate_node")
    if state.get("needs_summarize"):
        nodes.append("summarize_node")
    return nodes

builder.add_conditional_edges(
    "router",
    route_to_nodes,
    ["search_node", "calculate_node", "summarize_node"]
)

带 Config 的路由函数

路由函数还可以接收 config 参数,访问运行配置:

python
from langchain_core.runnables import RunnableConfig
from typing import Literal

def route_with_config(state: State, config: RunnableConfig) -> Literal["a", "b"]:
    thread_id = config["configurable"]["thread_id"]
    # 根据线程状态路由
    if thread_id in active_threads:
        return "a"
    return "b"

路径映射的两种写法

字典映射(推荐)

显式映射,清晰且安全:

python
builder.add_conditional_edges(
    "classifier",
    classify_intent,
    {
        "search": "search_node",
        "calculate": "calculate_node",
        "chat": "chat_node",
    }
)

列表映射(简洁)

按路由函数返回的字符串顺序匹配:

python
builder.add_conditional_edges(
    "classifier",
    classify_intent,
    ["search_node", "calculate_node", "chat_node"]
)

注意:如果路由函数返回的是节点名称,列表映射也可以直接用节点名称字符串。但字典映射更明确,推荐使用。

条件边 + 普通边混合

python
# START → router
builder.add_edge(START, "router")

# router 根据输入做条件路由
builder.add_conditional_edges("router", route_to_task, ["task_a", "task_b"])

# 不同任务走不同路径
builder.add_edge("task_a", "qa_node")    # task_a 后加质检
builder.add_edge("task_b", "summary_node") # task_b 后直接总结

# 最终汇聚
builder.add_edge("qa_node", "output")
builder.add_edge("summary_node", "output")
builder.add_edge("output", END)

注意事项

  1. 路由函数的返回值必须能在映射中找到,否则运行时会报错
  2. 不要返回不在映射中的节点名,建议用 Literal 类型注解做静态检查
  3. 条件边可以指向自身,实现循环(如工具调用循环)
  4. 路由映射中的 value 必须实际被添加为节点,否则编译会失败

参考

本站为非官方中文学习站点,不代表 LangChain 官方。部分内容参考官方文档并重新整理为中文学习笔记。