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