LangGraph 的条件边详解

一、条件边是什么

条件边（Conditional Edge）是LangGraph中实现动态「走哪条路」的核心工具。与add_edge这种固定路径不同，条件边在节点执行后调用一个路由函数，由该函数根据当前状态的计算结果，动态决定下一步进入哪个节点。逻辑关系如下：

普通边:  节点A ─(固定路径)─→ 节点B
条件边:  节点A ─(路由函数基于状态计算)─→ 节点B 或 节点C 或 节点D...

二、基础用法：add_conditional_edges

2.1 核心参数

graph.add_conditional_edges(
    source: str,              # 起始节点名称
    path: Callable,           # 路由函数，返回目标节点名称或名称列表
    path_map: dict = None     # 可选，返回值到节点名的映射
)

path 路由函数接收当前状态作为唯一参数，返回一个字符串（路由到单个节点）或字符串列表（并行路由到多个节点）。如果返回"__end__"（或其常量END），图将停止执行。

path_map 为可选字典，当路由函数的返回值需要映射到不同的节点名时使用；若省略该参数，则路由函数的返回值必须直接是目标节点名。

2.2 第一个条件边实例：Python问答分类

以下实现一个简单的问题分类器——根据用户问题类型，路由到对应的专业处理节点：

from typing import Literal
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

# 1. 定义状态结构
class State(TypedDict):
    question: str          # 用户问题
    answer: str            # 最终答案
    category: str          # 分类标签

# 2. 定义路由函数
def route_question(state: State) -> Literal["handle_math", "handle_other"]:
    """根据问题内容判断应进入哪个处理节点"""
    if "计算" in state["question"] or "数学" in state["question"]:
        return "handle_math"
    return "handle_other"

# 3. 构建图
graph = StateGraph(State)

graph.add_node("handle_math", lambda state: {"answer": f"数学计算结果: {len(state['question'])}"})
graph.add_node("handle_other", lambda state: {"answer": f"常规回答: {state['question']}"})

graph.add_edge(START, "router")                     # START → 路由节点
graph.add_conditional_edges("router", route_question)  # 条件边分支
graph.add_edge("handle_math", END)
graph.add_edge("handle_other", END)

三、进阶一：基于复杂状态的条件边

实际应用中，路由判断往往需要综合分析多个字段（意图、错误计数、历史记录等）。以下演示如何依据复杂状态进行路由决策：

from typing import List, Literal
from pydantic import BaseModel
from langgraph.graph import StateGraph, END

class AgentStatus(str, Enum):
    RUNNING = "running"
    SUCCESS = "success"
    ERROR = "error"
    NEED_TOOL = "need_tool"

class ComplexState(BaseModel):
    messages: List[dict] = []
    current_input: str = ""
    status: AgentStatus = AgentStatus.RUNNING
    error_count: int = 0

def route_by_status(state: ComplexState) -> Literal["process", "retry", "error", "end"]:
    """复杂状态路由：综合判断错误计数和运行状态"""
    if state.status == AgentStatus.SUCCESS:
        return "end"
    elif state.status == AgentStatus.ERROR:
        if state.error_count >= 3:          # 超过3次错误 → 进入错误处理
            return "error"
        return "retry"                       # 少于3次 → 重试
    elif state.status == AgentStatus.NEED_TOOL:
        return "process"
    return "process"

# 带 path_map 的条件边（将返回值映射到节点名）
graph.add_conditional_edges(
    "check_status",
    route_by_status,
    path_map={
        "process": "execute_tool",
        "retry": "retry_handler",
        "error": "error_handler",
        "end": END
    }
)

如果需要并行执行多个节点（如同时调用多个工具），路由函数可以返回一个字符串列表：

def route_to_parallel_tools(state: State) -> list:
    """返回需要并行执行的节点列表"""
    required_tools = identify_required_tools(state)  # 返回 ["search", "weather", "crypto"]
    return required_tools      # 这些节点将并行执行

graph.add_conditional_edges("analyze", route_to_parallel_tools)
# 注意：需要为每个目标节点添加边，将它们的结果汇总到 END

四、进阶二：通过条件边调用不同的 Agent

4.1 典型架构：路由器节点 + 多个子Agent

条件边最典型的应用场景之一是多Agent路由——一个中央路由器节点评估用户输入，然后动态决定将任务分发给最适合的子Agent。

4.2 完整代码：意图驱动的多Agent路由

以下演示一个客服场景：路由器根据用户意图（booker/info/unclear）将请求路由到对应的Agent节点。

Step 1: 定义状态结构

from typing import List, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from typing_extensions import TypedDict, Annotated

class MultiAgentState(TypedDict):
    messages: Annotated[List, add_messages]   # 对话历史（自动增量）
    intent: str                               # 意图分类结果
    final_response: str                       # 最终回答
    booking_details: dict                     # 预订专用数据

Step 2: 实现路由器节点（基于LLM的意图分类）

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

def router_node(state: MultiAgentState) -> dict:
    """分析用户输入，识别意图并记录分类结果"""
    prompt = f"""
    请将以下用户输入分类为以下三类之一:
    - "booker": 用户想预订、预约、确认座位/房间
    - "info": 用户想询问信息、了解产品、咨询介绍
    - "unclear": 意图不明确或不属于以上两类

    用户输入: {state['messages'][-1].content}

    只返回类别名称(booker/info/unclear)：
    """
    intent = llm.invoke(prompt).content.strip().lower()
    print(f"[Router] 识别的意图: {intent}")
    return {"intent": intent}

Step 3: 实现各Agent节点

def booking_agent(state: MultiAgentState) -> dict:
    """预订Agent: 处理座位/房间预订"""
    # 这里可以调用预订系统API
    return {
        "final_response": "✅ 预订已完成！您的座位号为A12，确认码已发送。",
        "booking_details": {"seat": "A12", "status": "confirmed"}
    }

def info_agent(state: MultiAgentState) -> dict:
    """信息Agent: 回答产品咨询"""
    return {
        "final_response": "📖 我们产品支持20种语言，月费$9.99，含全功能试用7天。"
    }

def unclear_agent(state: MultiAgentState) -> dict:
    """兜底Agent: 意图不明确时的处理"""
    return {
        "final_response": "❓ 抱歉我没理解您的意思。请重新表达，或选择:\n"
                         "1️⃣ 预订服务  2️⃣ 信息咨询"
    }

Step 4: 构建条件边 + 图结构

def route_by_intent(state: MultiAgentState) -> Literal["booking_agent", "info_agent", "unclear_agent"]:
    """根据意图字段路由到对应的子Agent"""
    intent = state.get("intent", "unclear")
    if intent == "booker":
        return "booking_agent"
    elif intent == "info":
        return "info_agent"
    return "unclear_agent"

# 构建图
graph = StateGraph(MultiAgentState)

graph.add_node("router", router_node)
graph.add_node("booking_agent", booking_agent)
graph.add_node("info_agent", info_agent)
graph.add_node("unclear_agent", unclear_agent)

graph.add_edge(START, "router")
graph.add_conditional_edges("router", route_by_intent)  # 核心条件边
graph.add_edge("booking_agent", END)
graph.add_edge("info_agent", END)
graph.add_edge("unclear_agent", END)

五、高级模式选项

1. 预置工具条件边 toolsCondition

对于Agent需要决定是否调用工具的常见场景，LangGraph提供了预置条件边函数toolsCondition，它检查状态中最后一条消息是否存在tool_calls：

from langgraph.prebuilt import tools_condition

# 工具条件边：如果有工具调用 → "tools"节点 → 工具执行后回到Agent；否则 → END
graph.add_conditional_edges("agent", tools_condition, {"tools": "tools", END: END})

该函数返回"tools"（若有工具调用）或END（若无）。

2. Command 路由

使用Command可以在单个节点内部直接指定下一步，无需单独的路由函数：

def intelligent_node(state: State):
    if state["needs_tool"]:
        return Command(goto="tool_executor", update={"status": "routing_to_tool"})
    else:
        return Command(goto=END, update={"final_answer": state.get("answer", "")})

3. Send 并行扇出

当需要将同一个输入并行派发给多个Agent时，在条件边路由函数中返回Send列表即可实现并行扇出：

from langgraph.types import Send

def dispatch_to_agents(state: State) -> list[Send]:
    """将用户输入并行分发给多个Agent"""
    return [
        Send("research_agent", {"query": state["question"]}),
        Send("summary_agent", {"query": state["question"]}),
        Send("translation_agent", {"query": state["question"]})
    ]

六、常见陷阱与最佳实践

陷阱1：路由函数返回了节点映射中没有的键

# ❌ 错误示例
graph.add_conditional_edges("router", route_func, mapping={"math": "math_node"})
# route_func可能返回 "other"，但 mapping 中没有 "other" 的映射 → KeyError

解决方案：确保path_map中覆盖了所有返回值，或让路由函数直接返回节点名称。

陷阱2：path_map中字典内误写注释导致键被污染

# ❌ 错误：多行注释字符串被当作字典键
graph.add_conditional_edges("decision", condition, mapping={
    """如果返回tools → 进入工具节点""",
    "tools": "Retrieve",
    END: END
})

Python解释器会将注释字符串当作字典的第一个键，导致KeyError。

解决方案：将注释移到字典外部。

# ✅ 正确写法
graph.add_conditional_edges(
    "decision", condition,
    mapping={
        "tools": "Retrieve",  # 工具调用分支
        END: END              # 结束分支
    }
)

陷阱3：忘记处理 END 分支导致无限循环

条件边必须提供明确的终止路径，否则图可能永不结束，耗尽资源。

# ❌ 危险：没有终止条件，可能无限循环
def route(state): return "next_step"   # 永远不回 END

# ✅ 安全：总是提供 END 作为可选输出
def route(state):
    if state["completed"]:
        return END
    return "next_step"

陷阱4：在异步环境中错误使用同步路由函数

# ❌ 会阻塞事件循环
async def some_async_node(state):
    return route_function(state)   # 在异步节点内调用了同步路由函数

# ✅ 使用异步版本
graph.add_conditional_edges(source="async_node", path=async_route_function)

最佳实践清单

七、总结

条件边的核心价值在于：将「走哪条路」的决策逻辑从节点内部抽离为独立的路由函数，使多Agent系统可以实现因「请求」而异的动态路径选择，而非僵硬的固定流水线。从简单路由到复杂多Agent编排，条件边都是构建健壮工作流的基石。

如需进一步针对您的具体业务场景进行路由设计或代码示例优化，欢迎提供更多需求细节，我可以为您生成更高贴合度的代码。

FAQ

Q1：条件边和普通边有什么区别？

A：
- 普通边（add_edge）是静态的：从节点 A 出发，固定走向节点 B。
- 条件边（add_conditional_edges）是动态的：从节点 A 出发，由路由函数根据当前状态（State）计算出一个目标节点（或多个目标节点），从而实现分支甚至并行执行。

Q2：路由函数可以返回多个节点吗？如何实现并行执行？

A：可以。路由函数返回字符串列表（例如 ["tool_a", "tool_b"]），LangGraph 会并行（或按顺序）执行列表中所有节点。之后需要手动将这些节点连接到汇聚节点（如 END），或使用 Send 机制更精细地控制数据分发。

示例：

def route_to_many(state):
    return ["math_agent", "weather_agent"]

graph.add_conditional_edges("router", route_to_many)

Q3：条件边中的 path_map 参数到底有什么用？什么时候必须使用？

A：path_map 用于将路由函数返回值映射到真实的节点名称。
必须使用的情形：路由函数返回的是语义键（如 "booker"、"info"），而不是节点名（如 "booking_agent"）。此时 path_map 提供映射关系，防止 KeyError。

示例：

graph.add_conditional_edges(
    "router", 
    route_intent,               # 返回 "booker"/"info"/"unclear"
    path_map={
        "booker": "booking_agent",
        "info": "info_agent",
        "unclear": "unclear_agent"
    }
)

如果路由函数直接返回节点名称（"booking_agent"），则可以省略 path_map。

Q4：如何根据复杂状态（比如错误计数、多条历史消息）进行分支判断？

A：路由函数可以访问整个 State 对象，因此能综合分析多个字段。例如：

def route_by_status(state: ComplexState):
    if state.error_count >= 3:
        return "error_handler"
    if state.status == "need_tool":
        return "tool_node"
    return "continue_node"

这种方式支持任意复杂的逻辑（计算、调用子函数、读取历史消息等）。

Q5：在没有 LLM 的情况下，条件边能实现多 Agent 路由吗？

A：完全可以。路由函数不限制必须使用 LLM——你可以用正则匹配、关键词词典、规则引擎甚至随机选择。LLM（如分类器）只是其中一种实现方式。上文示例中既有基于关键词的简单路由（"计算" → math），也有基于 LLM 的意图分类。

Q6：条件边会不会导致无限循环？如何防止？

A：会。如果路由函数永远不返回 END（或其常量），图可能永远循环。
解决方案：
- 确保至少有一条分支走向 END。
- 使用状态字段记录循环次数，超过阈值时强制结束。
- 添加日志监控，检测异常循环。

def safe_route(state):
    if state["completed"] or state["loop_count"] > 10:
        return END
    return "another_node"

Q7：tools_condition 是什么？它和自定义条件边有什么关系？

A：tools_condition 是 LangGraph 预置的一个条件边函数，专门用于 Agent 工具调用的场景。它检查状态中最后一条消息是否包含 tool_calls，若有则返回 "tools"，否则返回 END。你可以直接使用它，而不需要自己编写路由函数。

from langgraph.prebuilt import tools_condition

graph.add_conditional_edges("agent", tools_condition, {"tools": "tools", END: END})

对于更复杂的路由需求，仍然需要自定义条件边。

Q8：条件边能否在异步节点中使用？

A：可以，但需要注意：
- 如果节点是异步的（async def my_node(state)），条件边的路由函数也必须是异步的（async def route_func(state)）。
- 使用 graph.add_conditional_edges 时，path 参数会自动识别同步或异步函数。但不要在异步节点内调用同步路由函数（会阻塞事件循环）。

Q9：Command 机制可以替代条件边吗？

A：Command 更适合节点内部的局部决策（节点自己决定下一步，并通过 Command(goto=...) 返回），而条件边适合将路由逻辑集中在一个路由函数中。两者可以混用。
- 简单、线性的分流 → 条件边更清晰。
- Agent 根据单次执行结果决定去向 → Command 更简洁。

Q10：路由函数里访问状态时，应该注意什么？

A：
1. 只读访问：路由函数不应修改 State，应保持纯函数特性。
2. 安全取值：使用 state.get("key", default) 避免 KeyError。
3. 类型标注：建议用 Literal["node1", "node2", END] 标注返回值，IDE 可自动补全。
4. 日志记录：在路由函数开头打印状态关键字段，便于调试跳转路径。

def route(state: State) -> Literal["math", "other"]:
    print(f"[Route] question={state['question']}")
    return "math" if "计算" in state["question"] else "other"