LangGraph 核心机制

一句话：LangGraph 把流程写成“共享状态 + 节点 + 边”的图，节点只返回状态增量，边决定下一步；它换来的是分支、并行、暂停和恢复，代价是你要自己设计状态。

适用版本：LangGraph 1.x。图原语在 v1 保持稳定，主要新增的是持久化与流式能力。

为什么需要

create_agent 只能表达一种拓扑：模型和工具来回循环，直到模型不再调用工具。真实业务经常不是这样：先分类再走不同的处理链；召回多个来源并行执行后合并；先跑确定性校验，只有通过才让模型决策；中途停下来等人审批，几小时后继续；失败后从上一个成功步骤重试，而不是从头再来。

用 if/else 和函数调用也能写，但一旦需要持久化、重放、可视化，就得自己造一套状态机。LangGraph 就是这套状态机。

核心机制

三个原语

from typing import Annotated, TypedDict
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
class State(TypedDict):
    messages: Annotated[list, add_messages]   # reducer：追加而不是覆盖
    route: str
builder = StateGraph(State)
builder.add_node("classify", classify)        # 普通函数：state 进，部分更新出
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", lambda s: s["route"])
graph = builder.compile()

• State / Node：State 是所有节点共享的数据结构（通常是 TypedDict 或 dataclass）；Node 是普通函数，入参 state，返回部分更新而不是完整 state。
• Edge：固定边决定顺序，条件边用一个纯函数决定去哪个节点。START 和 END 是虚拟节点，不是你自己定义的。

编译期：循环是编译产物

用户写的是节点和边，编译后得到的是一个 super-step 循环。Pregel 风格的运行时会在编译时补上四个结构性节点（不同版本命名略有差异）：

节点	作用	与终止的关系
entry / exit	把输入送进第一个节点；把最终 state 返回调用方	决定从哪开始、在哪结束
loopEntry	每个 super-step 开始，收集本轮要跑的节点	还有待执行任务就继续
loopExit	本轮没有待执行节点、也没有挂起时离开循环	这是运行时的正常终止条件

所以“循环”不是你手写的 while，而是编译产物；你只声明拓扑和转移条件。终止条件因此分两层：编译期能查结构性终止——除 END 外每个节点都有出边、没有悬空边、所有节点都能走到 END；查不了语义终止——条件边完全可能永远返回同一个分支，只能靠 recursion_limit 在运行时兜底。

LangChain 与 LangGraph 的关系

create_agent 是 ReAct-first 的装配层：它把“模型—工具”循环、middleware、checkpointer 装配成一个 Agent，底层同样编译成 LangGraph 图运行时。所以 LangChain 不是“纯 ReAct、无图”，它只是把最常见的拓扑预置好了，你不需要手写节点和边；拓扑超出 ReAct 时，再直接下到 LangGraph 写图。

Reducer：状态怎么合并

默认行为是覆盖同名字段。用 Annotated[类型, reducer] 改变合并方式：

from operator import add
class State(TypedDict):
    total: Annotated[int, add]                # 各节点返回值累加
    messages: Annotated[list, add_messages]   # 按消息 ID 追加/更新
    status: str                               # 无 reducer：后写覆盖

add_messages 比 operator.add 多做的事：按消息 ID 去重和替换，支持人工修改历史消息。对话状态一律用它。

条件边与路由

def route(state: State) -> str:
    return "refund" if state["route"] == "refund" else "general"
builder.add_conditional_edges("classify", route, {"refund": "refund_node", "general": "general_node"})

路由函数必须是纯函数：只看 state，不调模型、不查库、不写日志。否则重放和调试时行为不可复现。

Command：节点自己决定去哪

固定边和条件边是“外部决定”，Command 是“节点内部决定”：

from typing import Literal
from langgraph.types import Command
def check(state: State) -> Command[Literal["pass", "fail"]]:
    goto = "pass" if state["score"] > 0.8 else "fail"
    return Command(goto=goto, update={"status": "ok" if goto == "pass" else "retry"})

注意：同一个节点不要同时用 Command(goto=...) 和静态 add_edge，否则两条路都会走。

Send：并行扇出

Send 用于 map-reduce：根据一个列表动态生成多个并行节点实例。

from langgraph.types import Send
def fan_out(state: State):
    return [Send("process_doc", {"doc": d}) for d in state["docs"]]
builder.add_conditional_edges("split", fan_out, ["process_doc"]); builder.add_edge("process_doc", "merge")

这张图只看 Send 的三段：在哪里展开、并行部分长什么样、结果在哪里收回。

Send 把列表展开成多个独立 state，并行结果靠 reducer 合回主 state；扇出越大，外部依赖的限流和合并成本越先成为瓶颈。条件边和 Command 解决“下一步去哪”，Send 解决“同一批任务并行几份”，三者不要混用。

子图与流式

编译后的图可以当节点用：builder.add_node("research", subgraph)。子图有独立的 checkpoint 命名空间，跨子图共享数据放父 state 或用 store。graph.stream(input, stream_mode="updates") 按节点返回增量；values 返回每步完整 state，messages 推送模型 token 与元数据，custom 供节点内主动推送。

实战场景

场景一：工单分诊图——模型决策放在节点里，路由保持纯函数

痛点：图运行时的卖点是可重放、可审计，而它最常被自己人破坏——把模型调用藏进路由函数，重放时模型输出变了，同一条工单两次走出不同分支，回放失去意义。正确分工是：分类节点调模型，路由只读 state。客服工单进来先分类，简单问题直接模板回复，退款走 Agent 子流程，其余转人工——这张图最重要的一个设计决定：

class Ticket(TypedDict):
    messages: Annotated[list, add_messages]
    category: str                       # 节点写入
    triaged: Annotated[list, add]       # reducer：结果累加，不覆盖

def classify(state):                    # 模型在这里调，结果落 state
    r = router_model.with_structured_output(Category).invoke(state["messages"])
    return {"category": r.category}     # refund / faq / human

def faq_reply(state):
    return {"messages": [("ai", FAQ_TEMPLATES[state["category"]])]}

b.add_conditional_edges("classify", lambda s: s["category"],
                        {"refund": refund_agent_subgraph, "faq": faq_reply, "human": escalate})

escalate（转人工）就是 interrupt({"question": "确认转人工？"}) 加末尾通知节点——分诊图里的写副作用（建工单、发消息）全部收在 Command(goto=END) 前的最后一个节点，前面任何一步恢复重跑都不产生外部影响。

场景二：多源研报聚合——Send 扇出的三个成本拐点

问题：一篇调研请求要并行查 5 个数据源，汇总后生成对比。Send 写法本身十行，实战要盯的是三个拐点：

1. 扇出数与下游限流：5 个源各自有 QPS 配额，并发 20 个请求等于自我 DoS——fan_out 里按 state["sources"][:MAX_PARALLEL] 截断，超出的排队下一批；被限流的表现是整批延迟变高而不是报错，监控要看 P99 不看成功率；
2. reducer 漏配：并行实例都往 triaged 写，字段没标 Annotated[list, add] 的话后写覆盖，症状是"结果数量随机"——这是图并行最隐蔽的 bug，上线前专门加一条"5 个源全部返回 5 条结果"的断言测试；
3. 合并节点的部分失败：3/5 成功时 merge 拿到缺项列表——Send 不自动聚合错误，要么子节点内部降级（返回带 error 标记的占位结果，合并时展示"来源 X 暂不可用"），要么整单失败重试；两种都行，但必须选一种并写进 state 契约，不能让合并节点靠猜。

子图复用的实战边界：refund_agent_subgraph（create_agent 编译产物）直接 add_node 挂进来，父子图之间只靠 state 字段名交接——字段改名是破坏性变更，旧 checkpoint 恢复会断，所以跨子图共享的字段要集中定义在一个 SharedState TypedDict 里，改一处全图可见。

最小上手

from typing import Annotated, TypedDict
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
State = TypedDict("State", {"messages": Annotated[list, add_messages], "route": str})
def classify(s): return {"route": "refund" if "退款" in s["messages"][-1].content else "general"}
b = StateGraph(State); b.add_node("classify", classify)
b.add_node("refund", lambda s: {"messages": [("ai", "七天内可退")]}); b.add_node("general", lambda s: {"messages": [("ai", "请补充细节")]})
b.add_edge(START, "classify"); b.add_conditional_edges("classify", lambda s: s["route"], {"refund": "refund", "general": "general"})
b.add_edge("refund", END); b.add_edge("general", END)
print(b.compile().invoke({"messages": [("user", "怎么退款")]})["messages"][-1].content)

生产动作

• state 要小且可序列化：只放 ID、结论、必要上下文；大文档、图片放外部存储，state 里存引用。路由函数保持纯粹，这是图可重放、可测试的前提。
• 节点要幂等，副作用放末尾：重试、恢复、时间旅行都会重跑节点；前半段已发消息的话，恢复时整段重跑会重复发送。
• 控制递归深度：默认 recursion_limit 是 25 步，复杂图要显式设置并监控，别靠调大掩盖死循环。
• 图结构也要版本化：改节点名、改边会让旧 checkpoint 无法恢复；升级时保留旧图或做状态迁移。
• 可视化与追踪：用 LangSmith 或 Mermaid 导出图，线上问题先看“卡在哪个节点”。

边界

• 不是通用工作流引擎：没有定时触发、事务补偿、人工表单等企业工作流能力；纯确定性流程用普通代码更简单。
• 状态设计错了会很难改：state 是图的公共契约，后期加字段容易，改语义很难。
• 并行和 checkpoint 都有成本：外部 API 有速率限制，扇出太大只会一起被限流；每步都写状态，长对话和大 state 会让存储和延迟上升。
• 节点不是隔离的：共享 state，节点之间通过约定协作，缺少强类型边界。

讲清楚

LangGraph 的模型是：定义一份共享 state，把每一步写成节点，节点只返回要更新的字段，边决定下一步走哪。默认字段是覆盖，用 Annotated 加 reducer 可以改成追加或累加，对话消息用 add_messages。

路由有两种：条件边用纯函数看 state 决定，适合静态分支；Command 让节点自己决定去哪，适合运行时才知道的跳转。Send 按列表并行扇出，适合 map-reduce；编译后的子图可以直接当节点用。编译时运行时会把拓扑补成 entry、loopEntry、loopExit、exit 四个结构性节点，循环是编译产物，结构性终止能静态检查，语义终止靠 recursion_limit 兜底。

这套东西真正的价值在持久化：每个 super-step 都能落 checkpoint，所以能暂停、恢复、回放。代价是节点会被重跑，state 必须可序列化，副作用必须幂等，图结构变更要考虑旧 checkpoint 的兼容。LangChain 的 create_agent 是 ReAct-first 装配层，底层同样是这套图运行时。

适用：LangGraph 1.x | 初稿 2026-10