条件边与路由：动态控制流

条件边的核心契约：路由函数是纯读取的。它接收 State（可选第二参数 config），返回值决定 graph 下一步走向，但它不负责修改 State——状态更新永远由节点完成。把路由逻辑和状态更新分开，是 LangGraph 控制流可预测的根基。

add_conditional_edges 的三种形态

graph.add_conditional_edges(source, path, path_map=None) 有三个关键参数。source 是出发节点；path 是路由函数（读 State 返回字符串、字符串列表，或 Send 列表）；path_map 是可选的「标签 → 节点名」字典。理解三种返回值形态，就掌握了 LangGraph 几乎全部的动态控制流。

完整示例一：经典 ReAct 循环（条件边 + 循环）

下面是最具代表性的图：agent 节点调 LLM 决定是否需要工具；条件边 should_continue 读最后一条消息——有 tool_calls 就路由到 tools，否则路由到 END。tools 节点执行工具后用普通边无条件回到 agent，于是形成 agent → tools → agent → …→ END 的循环。这正是 prebuilt create_react_agent 内部的图结构。

# 依赖：langgraph 核心 + langchain 的 OpenAI 集成（任选一个聊天模型即可）
pip install -U langgraph langchain-openai langchain-core

# 设置你的 key（示例用 OpenAI，可换成任意支持 tool calling 的模型）
export OPENAI_API_KEY="sk-..."

from typing import Annotated, Literal, TypedDict

from langchain_core.messages import AnyMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages


# 1) State：messages 用 add_messages reducer，使每个节点返回的新消息被【追加】而非覆盖
class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]


# 2) 定义一个真实可调用的工具
@tool
def get_weather(city: str) -> str:
    """查询某城市的天气。"""
    # 真实项目里这里会调外部 API；示例返回固定值
    return f"{city} 现在 22°C，晴。"


tools = [get_weather]
tools_by_name = {t.name: t for t in tools}

# 把工具 schema 绑定到模型上，模型才会产出 tool_calls
model = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(tools)


# 3) agent 节点：调用 LLM，返回的 AIMessage 会被 add_messages 追加进 state
def agent(state: AgentState) -> dict:
    response = model.invoke(state["messages"])
    return {"messages": [response]}


# 4) tools 节点：执行模型请求的每一个 tool_call，产出 ToolMessage
def tool_node(state: AgentState) -> dict:
    last = state["messages"][-1]
    outputs = []
    for call in last.tool_calls:
        result = tools_by_name[call["name"]].invoke(call["args"])
        outputs.append(
            ToolMessage(content=str(result), name=call["name"], tool_call_id=call["id"])
        )
    return {"messages": outputs}


# 5) 路由函数：只【读】 state，返回逻辑标签，不修改 state
def should_continue(state: AgentState) -> Literal["tools", "end"]:
    last = state["messages"][-1]
    # 模型这一轮请求了工具 -> 去执行；否则 -> 结束
    if getattr(last, "tool_calls", None):
        return "tools"
    return "end"


# 6) 组图
builder = StateGraph(AgentState)
builder.add_node("agent", agent)
builder.add_node("tools", tool_node)

builder.add_edge(START, "agent")

# 条件边：should_continue 返回的标签经 path_map 翻译成真实节点/END
builder.add_conditional_edges(
    "agent",
    should_continue,
    {"tools": "tools", "end": END},  # path_map：标签 -> 目标
)

# 普通边：工具跑完无条件回到 agent，形成循环
builder.add_edge("tools", "agent")

graph = builder.compile()

# 7) 运行
result = graph.invoke({"messages": [("user", "北京天气怎么样？")]})
for m in result["messages"]:
    m.pretty_print()

示例二：用 Command 在节点内一步完成「更新 + 路由」

条件边把「更新」和「路由」拆成两个对象。当一个节点天然既要写状态又要决定去向时，可以让节点直接返回 Command：update 字段写状态、goto 字段指定下一个节点。这样就不需要再为它单独画一条条件边，路由逻辑内聚在节点里。注意：用了 Command 路由的节点，函数签名建议加返回类型注解（如 Command[Literal["node_a", "node_b"]]），让 LangGraph 能正确画出图的边。

from typing import Annotated, Literal, TypedDict
from operator import add

from langgraph.graph import StateGraph, START, END
from langgraph.types import Command


class State(TypedDict):
    text: str
    log: Annotated[list[str], add]  # 用 add reducer 累积日志


# 节点直接返回 Command：同时 update 状态 + goto 路由
def classify(state: State) -> Command[Literal["handle_question", "handle_statement"]]:
    is_question = state["text"].strip().endswith("?")
    return Command(
        update={"log": ["classified"]},                 # 状态更新
        goto="handle_question" if is_question else "handle_statement",  # 路由
    )


def handle_question(state: State) -> Command[Literal["__end__"]]:
    return Command(update={"log": [f"answered: {state['text']}"]}, goto=END)


def handle_statement(state: State) -> Command[Literal["__end__"]]:
    return Command(update={"log": [f"acknowledged: {state['text']}"]}, goto=END)


builder = StateGraph(State)
builder.add_node("classify", classify)
builder.add_node("handle_question", handle_question)
builder.add_node("handle_statement", handle_statement)
builder.add_edge(START, "classify")
# 注意：没有写任何 add_conditional_edges —— 路由完全由 Command(goto=...) 承担

graph = builder.compile()

print(graph.invoke({"text": "今天会下雨吗?", "log": []}))
# {'text': '今天会下雨吗?', 'log': ['classified', 'answered: 今天会下雨吗?']}
print(graph.invoke({"text": "今天是周一.", "log": []}))
# {'text': '今天是周一.', 'log': ['classified', 'acknowledged: 今天是周一.']}

示例三：Send API —— map-reduce 式动态扇出

前面的 fan-out（返回节点名列表）扇出的分支数是写死的。真正的 map-reduce 场景里，分支数量运行时才知道（比如「把这篇文章拆成 N 个段落，每段并行总结」）。这时路由函数返回一个 Send 列表：Send("node_name", {独立的子 state})。LangGraph 会为列表里每个 Send 起一个该节点的并行实例，且每个实例拿到的是 Send 里指定的私有 state，而不是全局 state。各分支产出的结果再经 reducer 汇聚（reduce）回主 state。

from typing import Annotated, TypedDict
from operator import add

from langgraph.graph import StateGraph, START, END
from langgraph.types import Send


# 主 state：summaries 用 add reducer，承接各并行分支的汇聚结果
class OverallState(TypedDict):
    paragraphs: list[str]
    summaries: Annotated[list[str], add]


# 子任务 state：每个并行分支只看到自己那一段
class WorkerState(TypedDict):
    paragraph: str


# map 阶段的路由函数：运行时根据段落数量动态生成 N 个 Send
def fan_out(state: OverallState):
    # 为每个段落派发一个 summarize 实例，各自带独立的私有 state
    return [Send("summarize", {"paragraph": p}) for p in state["paragraphs"]]


# worker 节点：只处理自己收到的那一段
def summarize(state: WorkerState) -> dict:
    p = state["paragraph"]
    summary = p[:10] + "..." if len(p) > 10 else p   # 占位「总结」逻辑
    return {"summaries": [f"[摘要] {summary}"]}        # 经 add reducer 汇聚回主 state


def reduce(state: OverallState) -> dict:
    # reduce 阶段：所有并行分支结束后才执行，此时 summaries 已汇齐
    print(f"共汇聚 {len(state['summaries'])} 条摘要")
    return {}


builder = StateGraph(OverallState)
builder.add_node("summarize", summarize)
builder.add_node("reduce", reduce)

# 条件边的路由函数返回 Send 列表 —— 注意 path_map 不适用于 Send
builder.add_conditional_edges(START, fan_out, ["summarize"])
builder.add_edge("summarize", "reduce")  # 所有 summarize 实例都跑完后进入 reduce
builder.add_edge("reduce", END)

graph = builder.compile()

result = graph.invoke({
    "paragraphs": ["LangGraph 是有状态图框架。", "条件边实现动态路由。", "Send 实现并行扇出。"],
    "summaries": [],
})
print(result["summaries"])
# ['[摘要] LangGraph 是有...', '[摘要] 条件边实现动态...', '[摘要] Send 实现并行扇...']

至此你已经能让图做决策、循环、并行扇出。但这些图每次 invoke 都是无记忆的一次性运行——重启就丢。下一章进入 Checkpointer 持久化与 thread_id 会话，让图能在多次调用之间记住状态、断点续跑，这也是后面 Human-in-the-loop 中断恢复的地基。