条件边Conditional Edge 高级模式详解

一、预置工具条件边 tools_condition

在 ReAct 风格的 Agent 中，模型会输出是否调用工具。tools_condition 是一个内置的路由函数，它检查状态中最后一条消息是否包含 tool_calls 属性，若有则返回 "tools"，否则返回 END。它常与 add_conditional_edges 配合，形成一个「Agent → 工具 → Agent」循环。

实用高级例子：带记忆和重试的工具调用循环

以下实现一个智能客服 Agent，它可以使用搜索引擎工具，并在工具调用失败时自动重试。

from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.prebuilt import tools_condition, ToolNode
from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearchRun
from typing import Literal

# 初始化工具
search_tool = DuckDuckGoSearchRun()
tools = [search_tool]
tool_node = ToolNode(tools)

# 绑定工具的 LLM
llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)

def agent(state: MessagesState):
    """Agent 节点：调用 LLM 并返回带 tool_calls 的消息"""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: MessagesState) -> Literal["tools", END]:
    """自定义条件，增加重试逻辑"""
    last_message = state["messages"][-1]
    # 若存在工具调用且未超过最大重试次数，走 tools 节点
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        # 可以在 state 中记录重试次数，这里简化
        return "tools"
    return END

# 构建图
graph = StateGraph(MessagesState)
graph.add_node("agent", agent)
graph.add_node("tools", tool_node)

graph.add_edge(START, "agent")
# 使用 tools_condition 的等效实现
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")  # 工具执行后回到 agent

app = graph.compile()

关键点：tools_condition 是一个可复用的内置函数，等价于上面 should_continue 的标准实现。你可以在 path_map 中覆盖默认映射（如将 "tools" 映射到其他节点）。

二、Command 路由

Command 是 LangGraph 0.2+ 引入的机制，允许在节点内部直接指定下一步去向并更新状态，无需单独的路由函数。当分支逻辑只依赖该节点自身的计算结果时，Command 比条件边更简洁。

实用高级例子：意图识别 + 直接跳转

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

class State(TypedDict):
    user_input: str
    response: str
    retries: int

def intelligent_router(state: State) -> Command[Literal["booking", "info", "retry", END]]:
    """根据输入和重试次数，直接返回 Command 决定下一个节点"""
    if state["retries"] >= 3:
        return Command(goto=END, update={"response": "多次失败，请稍后再试"})
    
    if "预订" in state["user_input"]:
        return Command(goto="booking", update={"response": "正在跳转预订..."})
    elif "信息" in state["user_input"]:
        return Command(goto="info", update={"response": "正在跳转信息查询..."})
    else:
        # 重试：增加重试计数，跳转到同一个节点重新分析
        return Command(goto="intelligent_router", update={"retries": state["retries"] + 1})

def booking_node(state: State):
    return {"response": "预订成功，确认码 12345"}

def info_node(state: State):
    return {"response": "我们的产品支持 20 种语言。"}

builder = StateGraph(State)
builder.add_node("intelligent_router", intelligent_router)
builder.add_node("booking", booking_node)
builder.add_node("info", info_node)

builder.add_edge(START, "intelligent_router")
builder.add_edge("booking", END)
builder.add_edge("info", END)

graph = builder.compile()

优势：无需单独定义条件边函数，状态更新和路由决策在同一个函数内完成，适合复杂逻辑内聚。

三、Send 并行扇出

当需要将一个输入并行分配给多个子 Agent（或任务）时，路由函数可以返回一个 Send 对象列表。每个 Send 包含目标节点名称和要传递给该节点的状态子集。LangGraph 会并发执行所有目标节点。

实用高级例子：文档分块并行摘要

from langgraph.graph import StateGraph, START, END, Send
from typing import List, TypedDict

class OverallState(TypedDict):
    document: str                # 原始文档
    chunks: List[str]            # 分块后的列表
    summaries: List[str]         # 各块摘要（按顺序）
    final_summary: str

def chunk_document(state: OverallState):
    """将文档切分成块"""
    chunks = [state["document"][i:i+500] for i in range(0, len(state["document"]), 500)]
    return {"chunks": chunks}

def dispatch_to_summarizers(state: OverallState) -> List[Send]:
    """为每一个 chunk 生成一个 Send 对象，并行调用 summarize_chunk 节点"""
    sends = []
    for idx, chunk in enumerate(state["chunks"]):
        # 传递 chunk 内容和索引，以便后续按序合并
        sends.append(Send("summarize_chunk", {"chunk": chunk, "index": idx}))
    return sends

def summarize_chunk(state: dict):
    """单个块的摘要逻辑（实际可调用 LLM）"""
    chunk = state["chunk"]
    # 模拟摘要
    summary = f"Summary of chunk {state['index']}: {chunk[:50]}..."
    return {"summary": summary, "index": state["index"]}

def aggregate_summaries(state: OverallState):
    """收集所有并行结果并生成最终摘要"""
    # 注意：并行节点的输出会通过状态合并，需要合理设计 reducer
    # 此处简化：假设 summaries 列表通过自定义 reducer 按 index 填充
    final = " ".join(state["summaries"])
    return {"final_summary": final}

# 构建图
builder = StateGraph(OverallState)
builder.add_node("chunk_document", chunk_document)
builder.add_node("summarize_chunk", summarize_chunk)
builder.add_node("aggregate", aggregate_summaries)

builder.add_edge(START, "chunk_document")
builder.add_conditional_edges("chunk_document", dispatch_to_summarizers)  # 扇出
builder.add_edge("summarize_chunk", "aggregate")  # 注意：实际需要将所有并行节点指向 aggregate，此处可通过 Send 返回值自动处理？

重要说明：Send 返回的节点执行完毕后，需要有一个汇聚节点收集结果。LangGraph 会自动等待所有并行 Send 完成，然后继续执行图的下一条边。通常做法是：扇出节点（如 chunk_document）后跟条件边返回 Send 列表，然后在每个目标节点（如 summarize_chunk）后添加一条普通边指向汇聚节点（aggregate）。LangGraph 保证所有并行任务全部完成后才会调用汇聚节点。

更严谨写法：

# 在上面的例子基础上，添加边：
builder.add_edge("summarize_chunk", "aggregate")
builder.add_edge("aggregate", END)

常见问题 FAQ

Q1：条件边的路由函数能否是异步函数？

A：可以。在 add_conditional_edges 中传入的 path 函数可以是 async def 形式。但注意整个图的执行环境必须支持异步（使用 graph.ainvoke 或 graph.abatch）。

Q2：如果路由函数返回的节点名称不在 graph 中会怎样？

A：运行时抛出 KeyError。因此建议使用 Literal 类型限制返回值，并确保所有返回值都已通过 add_node 添加。

Q3：tools_condition 默认返回 END，我能否让它重试其他节点？

A：可以通过 path_map 修改映射。例如：

graph.add_conditional_edges(
    "agent",
    tools_condition,
    {"tools": "tools", END: "fallback_handler"}  # 将 END 映射到自定义节点
)

Q4：Command 和 add_conditional_edges 哪个性能更好？

A：Command 略快一些，因为它避免了额外的一次函数调用开销。但差异微小，选择取决于设计偏好：Command 适合节点内自包含的决策，条件边适合多个节点共享同一路由逻辑。

Q5：使用 Send 并行时，如何保证结果按原始顺序合并？

A：每一个 Send 传递一个索引（如 "index": i），然后在汇聚节点中根据索引将结果插入到列表的正确位置。LangGraph 不保证并行任务的完成顺序，因此必须显式使用索引进行排序。

Q6：能否在条件边中动态添加节点？

A：不能。所有节点必须在编译图之前静态添加。如果需要动态行为，可以使用 Command 跳转到同一个「分派器」节点，由该节点内部去调用不同的子工作流。

Q7：条件边的路由函数能否访问外部变量？

A：可以，但要注意线程安全。如果图被并发执行，外部变量的修改可能引发竞态条件。建议将所有决策所需的数据放入 State 中。

Q8：tools_condition 如何与自定义状态类型（非 MessagesState）配合？

A：tools_condition 硬编码读取 state["messages"]。如果你的状态不包含该字段，可以自己编写一个等价的函数，从你自己的字段中提取最后一条消息并检查 tool_calls。

Q9：条件边支持返回 None 或空列表吗？

A：返回 None 或空列表会被 LangGraph 解释为继续执行当前节点的下一条普通边（如果有）。通常不推荐这样使用，容易造成歧义。最好显式返回 END 或具体节点。

Q10：如何调试条件边的跳转路径？

A：可以在路由函数中添加 print(f"Routing to: {result}")。使用 LangGraph Studio 可视化界面可以更直观地看到每条边实际被遍历的情况。另外，启用 langgraph.debug=True 环境变量会输出详细日志。

以上高级模式和 FAQ 覆盖了 LangGraph 条件边在生产环境中常见的复杂场景。建议结合具体业务场景选择合适的模式——简单二分支用普通条件边，工具调用用 tools_condition，节点内决策用 Command，大规模并行用 Send。