条件边 Conditional Edge

上一篇的普通边只能"确定地走某条路"。但真实场景里你常常需要"看情况"——根据当前状态决定走哪条分支。这就是条件边（ConditionalEdge）的用武之地，它也是构建循环（如 ReAct 智能体）的关键。

一、条件边是什么

条件边：执行完 source 节点后，调用一个路由函数读取当前 state，返回"下一节点名"，框架据此跳转。

API：

graph.add_conditional_edges(
    source,         # 从哪个节点出发
    router_fn,      # 路由函数：(state) -> str | list[str]
    path_map,       # 可选：返回值 -> 节点名 的映射
)

graph LR
    A[节点A] --> R{router 函数}
    R -->|返回 'yes'| Y[节点Y]
    R -->|返回 'no'| N[节点N]
    R -->|返回 'end'| E([END])

二、路由函数签名

def router(state: State) -> str:
    # 读 state，决定下一步
    if state["score"] > 60:
        return "pass"
    return "fail"

• 输入：当前 state。
• 输出：字符串（下一节点的名字）或字符串列表（多路分发，见后文）。
• 不修改 state：路由函数只读不写。

返回的字符串有两种解释方式：

1. 直接是节点名：返回 "pass"，框架去找名为 pass 的节点。
2. 通过 path_map 映射：返回 "pass"，path_map 把它映射到节点 celebrate_node。

三、path_map 的作用

path_map 是可选的，把路由函数返回值映射到实际节点名：

def router(state) -> str:
    return "yes" if state["ok"] else "no"

g.add_conditional_edges(
    "decide",
    router,
    {
        "yes": "celebrate",   # 返回 "yes" 时去 celebrate 节点
        "no":  "retry",       # 返回 "no"  时去 retry 节点
    },
)

path_map 的好处：

• 解耦：路由函数返回业务语义（"yes"/"no"），不直接耦合到节点名（"celebrate"/"retry"）。改名节点不用动路由函数。
• 可读性：图代码一眼看清分支。
• 可校验：编译时框架会检查 path_map 里的值是否对应已注册的节点。

如果不用 path_map，路由函数直接返回节点名也行：

def router(state) -> str:
    return "celebrate" if state["ok"] else "retry"

g.add_conditional_edges("decide", router)

新手推荐用 path_map，更清晰。

四、经典示例：基于状态字段的 if-else

# cond_edge_demo.py
from typing import TypedDict
from langgraph.graph import StateGraph, START, END


class State(TypedDict):
    user_input: str
    is_question: bool
    answer: str


def classify(state: State) -> dict:
    """判断用户输入是不是问题。"""
    text = state["user_input"]
    is_q = text.endswith("?") or text.endswith("？")
    return {"is_question": is_q}


def answer_question(state: State) -> dict:
    return {"answer": f"这是个问题：{state['user_input']}"}


def echo(state: State) -> dict:
    return {"answer": f"你说的是：{state['user_input']}"}


def route(state: State) -> str:
    return "answer" if state["is_question"] else "echo"


g = StateGraph(State)
g.add_node("classify", classify)
g.add_node("answer", answer_question)
g.add_node("echo", echo)

g.add_edge(START, "classify")
g.add_conditional_edges(
    "classify",
    route,
    {"answer": "answer", "echo": "echo"},
)
g.add_edge("answer", END)
g.add_edge("echo", END)
app = g.compile()

print(app.invoke({"user_input": "什么是 LangGraph？"})["answer"])
# 输出：这是个问题：什么是 LangGraph？

print(app.invoke({"user_input": "你好"})["answer"])
# 输出：你说的是：你好

图结构：

graph LR
    S([START]) --> C[classify 分类]
    C --> R{route 路由}
    R -->|问题| A[answer 回答]
    R -->|非问题| E[echo 复述]
    A --> END1([END])
    E --> END2([END])

注意两个分支都连到 END——条件边后每个分支要单独连到 END（或下一节点）。

五、进阶示例：基于 LLM 判断走不同分支

实际场景：用户问问题时，判断是否需要先检索知识库。让 LLM 来做这个判断：

# llm_router.py
import os
from typing import TypedDict
from typing_extensions import Annotated
from operator import add
from dotenv import load_dotenv
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

load_dotenv()


class State(TypedDict):
    user_query: str
    needs_retrieval: bool
    docs: list[str]
    answer: str


llm = ChatOpenAI(
    model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)


def decide(state: State) -> dict:
    """用 LLM 判断是否需要检索。"""
    sys = "你是路由器。用户问题需要查资料才能准确回答就输出 YES，否则输出 NO。只输出 YES 或 NO。"
    resp = llm.invoke([
        SystemMessage(content=sys),
        HumanMessage(content=state["user_query"]),
    ])
    needs = "YES" in resp.content.upper()
    return {"needs_retrieval": needs}


def route(state: State) -> str:
    return "retrieve" if state["needs_retrieval"] else "generate"


def retrieve(state: State) -> dict:
    """模拟检索。"""
    return {"docs": [f"关于《{state['user_query']}》的资料"]}


def generate(state: State) -> dict:
    """生成回答。"""
    docs = state.get("docs", [])
    if docs:
        return {"answer": f"基于资料 {docs} 回答：..."}
    return {"answer": f"直接回答：{state['user_query']} 是 ..."}


g = StateGraph(State)
g.add_node("decide", decide)
g.add_node("retrieve", retrieve)
g.add_node("generate", generate)
g.add_edge(START, "decide")
g.add_conditional_edges(
    "decide",
    route,
    {"retrieve": "retrieve", "generate": "generate"},
)
g.add_edge("retrieve", "generate")
g.add_edge("generate", END)
app = g.compile()

result = app.invoke({"user_query": "LangGraph 0.2 的 interrupt 怎么用？", "docs": []})
print(result["answer"])

图结构：

graph LR
    S([START]) --> D[decide LLM 判断]
    D --> R{route}
    R -->|需要检索| RT[retrieve 检索]
    R -->|不需要| G[generate 生成]
    RT --> G
    G --> E([END])

这是经典的"路由式 RAG"骨架。真实项目里把 retrieve 接到向量库即可。

六、条件边是构建循环的关键

普通边只能向前走，条件边可以根据状态决定"回头"。这是 ReAct 智能体的核心机制：

graph LR
    S([START]) --> A[agent 思考]
    A --> C{是否需要工具?}
    C -->|是| T[tools 执行]
    T --> A
    C -->|否| E([END])

伪代码：

def should_continue(state) -> str:
    last_msg = state["messages"][-1]
    # 如果 LLM 说要调工具，就去 tools 节点
    if last_msg.tool_calls:
        return "tools"
    return "end"

g.add_edge(START, "agent")
g.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END})
g.add_edge("tools", "agent")  # tools 执行完回到 agent，形成循环

LangGraph 内置的 tools_condition 就是干这事的快捷函数（见 智能体 模块）。

七、多路分发（返回列表）

路由函数可以返回字符串列表，表示"同时去多个节点"（并行扇出）：

def route(state) -> list[str]:
    return ["search_web", "search_db", "search_cache"]

g.add_conditional_edges("dispatch", route)

注意：返回列表时不能用 path_map（path_map 只支持单值）。这种"扇出"更推荐用 Send 对象（见 进阶特性 模块），它能给每个目标传不同的输入。

八、返回的节点名必须存在

这是新手最常踩的坑：路由函数返回了 "retrieve"，但图里没注册这个节点——运行时报错。

def route(state) -> str:
    return "retrive"   # 拼错了，少了 e

# 实际注册的是 "retrieve"
g.add_node("retrieve", retrieve)
g.add_conditional_edges("decide", route)

运行时：路由函数返回 "retrive"，框架找不到这个节点，报错。

防御方法：

1. 用 path_map：path_map 里没列出的返回值会触发明确错误，便于排查。

2. 路由函数里用常量而非裸字符串：

   NODE_PASS = "pass"
   NODE_FAIL = "fail"
   def route(state) -> str:
       return NODE_PASS if state["ok"] else NODE_FAIL

3. 加单测覆盖路由函数所有分支。

九、与 LangGraph4j 对比

维度	Python	Java
API	add_conditional_edges(src, router, path_map)	addConditionalEdges(src, router, pathMap)
路由签名	(state) -> str | list[str]	(state) -> String | List<String>
path_map 类型	dict[str, str]	Map<String,String>
多路分发	返回 list 或用 Send	返回 List 或用 Send

完全对应。

十、常见踩坑

1. 路由函数返回的节点名拼错：运行时找不到节点。用 path_map 或常量防御。
2. 忘了 path_map 导致路由到不存在的节点：返回 "yes" 但没有名为 "yes" 的节点。要么用 path_map 映射，要么节点直接叫这名字。
3. 路由函数有副作用：路由应该纯读 state。在里面改 state、调 LLM 都不规范（虽然能调 LLM，但通常拆成独立"决策节点"更清晰）。
4. 条件分支后忘记连 END：A -> 条件 -> B 或 C，但 B/C 都没连 END——图会卡住或循环到死。
5. 循环没有退出条件：agent -> tools -> agent 但条件函数永远返回 "tools"——死循环直到 recursion_limit。一定要有"够了就 end"的判断。
6. 多个条件边叠加：一个节点同时有普通出边和条件出边会冲突。一个节点要么一条普通出边，要么一个条件边。
7. 路由返回 None：路由函数应返回字符串或列表，不要返回 None。必要时显式 return "default"。
8. path_map 漏了某分支：路由可能返回三种值，path_map 只列了两种——第三种触发 KeyError。
9. 依赖状态字段未设置：路由函数读 state["score"] 但前面节点没设这个字段——KeyError。要么初始 state 设默认值，要么 state.get("score", 0)。
10. 条件边后想再分支：可以——条件边到的节点本身也能再连条件边，形成多层决策树。

十一、可视化检查

条件边在 mermaid 里会画成菱形决策节点：

print(app.get_graph().draw_mermaid())

肉眼检查分支是否符合预期。

十二、小结

• 条件边 add_conditional_edges(source, router, path_map) 让图根据状态动态选下一节点。
• 路由函数 (state) -> str 只读不写，返回下一节点名（或语义标签 + path_map 映射）。
• 推荐用 path_map，业务语义和节点名解耦。
• 经典用法：if-else 分支、LLM 路由、构建循环（ReAct）。
• 多路分发可返回列表，但更推荐用 Send。
• 防御路由返回错节点名：用 path_map、常量、单测。

下一篇讲 StateGraph 状态图——把节点和边装起来的核心容器。