Skip to content

Q49 · LangGraph 的 State、Node、Edge 各自职责是什么? ​

用户问客服:“订单 A123 发货了吗?”系统先查订单。如果查到了,就按真实状态答复;如果查不到,就让用户核对订单号。写成一段 if 当然也能运行。但当流程以后要加权限核验、补查、人工处理和失败重试时,程序必须清楚保存“目前知道什么”和“下一步执行谁”。LangGraph 用一张图表示这样的执行路线:State(状态)保存当前数据,Node(节点)执行一步工作并提交更新,Edge(边)决定下一步去哪里。LangGraph Graph API 官方总览用这三个组件描述图的核心。

这三个概念分别回答不同问题:手上有什么数据?这一步做什么?做完后去哪一步? 下文只用一笔虚构订单把三者接起来;不调用大模型,所以读者能直接看到运行结果,也能看清图编排与“模型思考”并非一回事。

术语与例子中的名字 ​

词或名字先用日常话理解本文的具体对应物
LangGraph / 图把多步程序组织成节点和连接,并管理运行中的状态订单查询、分流和答复的路线
State / 状态一次运行到当前为止保存的数据快照;由若干具名字段组成订单号、是否查到、状态、答复、事件记录
状态字段 / key状态里的一个有名字的数据格order_id、found、status、answer、events
Node / 节点接收当前状态、做一步工作、返回更新的函数lookup_order 查订单;compose_answer 写答复
Edge / 边指定一个节点完成后哪一个节点运行的连接查订单后,根据 found 去成功或缺资料分支
固定边 / 条件边固定边总指向指定后继;条件边先读取更新后的状态,再选去向开始后总去查订单;查完按 found 分流
Reducer / 归并函数指定某个字段收到新值后如何合并旧值和新值events 用 add 把新事件接到旧列表后
部分更新节点只返回这一步改动的字段,不重写整个状态查到订单时返回 found、status、events
Schema / 模式规定状态中有哪些字段及各自的数据类型OrderState 这个 TypedDict 类型声明
START / END图的虚拟起点 / 终点;不是执行业务代码的节点开始查订单;答复后结束
A123 / X999虚构订单号A123 存在;X999 不在演示数据中
ORDER_DB演示用的内存字典,模拟订单系统{"A123": "已发货"}

本文的业务事实均为教学假设:演示数据只有 A123,状态是“已发货”;X999 不存在。这里不推断运单号、到货时间或其他现实订单信息。生产环境里,用户身份、订单权限和订单服务的最新结果都要独立核实。

把同一个问题拆成三个责任 ​

State 管“当前知道什么”。 起点只知道 order_id=A123,事件列表 events=[]。查询后才知道 found=True 和 status=已发货;答复生成后才有 answer。这份状态是本次图运行的当前快照,不是模型永久记忆,更不等于自动存入数据库。LangGraph 文档说状态由字段模式和每个字段的 reducer 构成;若要跨请求保存并恢复,需要另行配置持久化机制,例如检查点(checkpointer)。Graph API:State 与编译

Node 管“这一步做什么”。 lookup_order 读取 order_id,查演示数据;找到就返回 found=True、status=已发货,找不到就只返回 found=False。compose_answer 根据已知状态生成答复;cannot_answer 说明缺资料。这些节点都是普通 Python 函数;真实系统可以在节点中调用 API 或模型,但是否调用模型并不改变 Node 的职责。节点只需返回这一步的部分更新,LangGraph 再按各字段规则合并到 State;不必每次返回所有旧字段。Graph API:Nodes 与部分更新

Edge 管“下一步去哪”。 起点到查询节点是一条固定边。查询节点后用条件边读取 found:真则去写订单答复,假则去说明缺资料。两个答复节点各有一条到终点的固定边。条件边的路由函数只负责选择去向,不查订单、不产生答复,也不应假装自己更新了业务字段。Graph API:普通边与条件边

State 保存订单数据,查询 Node 返回更新,固定边与条件边安排下一节点

图把“起始 State → 查订单 Node → 合并更新后的 State → 按 found 选 Edge”画在一起。右侧“查到 / 未查到”来自查询节点的返回值;图只展示了 A123 查到时的状态更新,未查到的另一条路线在下面用代码展开。

从状态字段写到可运行的图 ​

先说明代码中将出现的名字。OrderState 是一次订单查询的状态结构;order_id 是用户给出的订单号;found 记录查询是否找到;status 仅在找到时有值;answer 是最终答复;events 是逐步追加的事件列表。lookup_order 执行查询,route_after_lookup 根据查询结果返回 "found" 或 "missing" 两个路线标签;compose_answer 与 cannot_answer 分别完成两种答复。builder 是还在组装的图,graph 是编译后可运行的图。

代码使用 Python 3.11 以上语法与 LangGraph 1.x 的 Graph API。先在隔离环境安装 langgraph,把下面内容保存为 order_graph.py 后运行 python order_graph.py。这里的 ORDER_DB 是固定内存数据,不连接真实订单系统,因此代码可运行,业务数据仅作演示:

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

from langgraph.graph import END, START, StateGraph


class OrderState(TypedDict):
    order_id: str
    events: Annotated[list[str], add]
    found: NotRequired[bool]
    status: NotRequired[str]
    answer: NotRequired[str]


ORDER_DB = {"A123": "已发货"}


def lookup_order(state: OrderState) -> dict:
    order_id = state["order_id"]
    if order_id in ORDER_DB:
        return {
            "found": True,
            "status": ORDER_DB[order_id],
            "events": ["查到订单"],
        }
    return {"found": False, "events": ["未查到订单"]}


def route_after_lookup(state: OrderState) -> Literal["found", "missing"]:
    return "found" if state["found"] else "missing"


def compose_answer(state: OrderState) -> dict:
    return {
        "answer": f"订单 {state['order_id']} {state['status']}。",
        "events": ["已生成答复"],
    }


def cannot_answer(state: OrderState) -> dict:
    return {
        "answer": f"未查到订单 {state['order_id']},请核对订单号。",
        "events": ["已说明缺少资料"],
    }


builder = StateGraph(OrderState)
builder.add_node("lookup_order", lookup_order)
builder.add_node("compose_answer", compose_answer)
builder.add_node("cannot_answer", cannot_answer)

builder.add_edge(START, "lookup_order")
builder.add_conditional_edges(
    "lookup_order",
    route_after_lookup,
    {"found": "compose_answer", "missing": "cannot_answer"},
)
builder.add_edge("compose_answer", END)
builder.add_edge("cannot_answer", END)

graph = builder.compile()

for order_id in ("A123", "X999"):
    print(order_id)
    for snapshot in graph.stream(
        {"order_id": order_id, "events": []}, stream_mode="values"
    ):
        print(snapshot)

TypedDict 声明“字典中应该有哪些键”;NotRequired 表示该字段在起点可以尚不存在,后续节点才填。Annotated[list[str], add] 把 Python 的列表加法 add 指定为 events 的 reducer;其余字段不指定 reducer,默认由新值覆盖旧值。Literal["found", "missing"] 说明路由函数只能返回这两个标签,便于读代码和类型检查。类型提示本身不等于运行时的权限验证或对外部数据的业务校验。LangGraph reducer 文档解释了默认覆盖和自定义合并的区别。

StateGraph(OrderState) 按状态模式创建图。三次 add_node 把业务函数注册为节点;add_edge(START, "lookup_order") 给出起点的固定去向;add_conditional_edges 在查询完成后运行 route_after_lookup,并把路线标签映射到真实节点名。最后两条固定边都通向 END。compile() 在调用前构建可执行图并做结构检查;graph.stream(..., stream_mode="values") 逐步给出完整状态快照,因此能看见旧字段保留下来。若只想看每步新产生的字段,可用 stream_mode="updates"。Graph API:编译与流式状态

沿 A123 把每次状态更新走完 ​

运行代码时,A123 的三个快照如下;表中只省略了 Python 字典的引号,值与实测输出一致:

时刻已有状态谁做了什么、下一步去哪
输入刚进入图order_id=A123,events=[]START 的固定边安排 lookup_order
查询节点完成后order_id=A123,found=True,status=已发货,events=[查到订单]lookup_order 只返回三个更新字段;条件边读取更新后的 found=True,选择 compose_answer
写答复节点完成后原字段保留,新增 answer=订单 A123 已发货。,events=[查到订单, 已生成答复]compose_answer 返回答复和一个新事件;固定边到 END

第二步很能说明部分更新:查询节点没有返回 order_id,但它并未消失;第三步答复节点也没有再返回 status,它仍保留在状态中。found、status、answer 没有特殊 reducer,后续若再次写同一字段就以新值覆盖。events 不同:旧列表 ["查到订单"] 与节点返回的新列表 ["已生成答复"] 经 add 连接,得到两个事件。LangGraph 官方用“旧值(left)+ 节点更新(right)”解释 reducer 的合并方式。Graph API:Reducer 参数

这也解释了为什么给 events 传入完整旧列表会出错:如果第二个节点返回 ["查到订单", "已生成答复"],add 会再把旧列表拼一次,出现重复事件。每个节点只返回新增的事件。反过来,对需要保留“当前唯一状态”的 status 字段,不应无缘无故使用列表追加 reducer;旧状态过期时应由新状态替换。

X999 的缺资料路线与真实故障 ​

输入换成 X999 时,ORDER_DB 不包含这个键。lookup_order 返回 found=False 与新事件“未查到订单”,没有返回 status。条件边读到 False,走 cannot_answer;它给出“未查到订单 X999,请核对订单号。”,再到 END。最终状态包含 order_id=X999、found=False、两条事件和 answer,但没有 status。这样才能避免把上一笔订单 A123 的“已发货”拿来回答 X999。由于每次 graph.stream 都启动新的运行,这两笔订单的状态不会自动混在一起。

“没查到”与“订单服务故障”在真实系统里还需分开。演示的内存字典能立即给出确定的存在/不存在;生产查询遇到超时,结果是未知,不能走 missing 分支,也不能宣称不存在。应把查询状态扩展成例如 found / missing / error 三类,给 error 单独的条件边,按服务契约有限重试或告知暂不能确认。还需在查询节点或更外层根据登录身份验证订单访问权限;TypedDict 不会替你做授权。Q39:工具调用失败如何处理讨论了超时和重试的细分。

若把条件边错误地写成“无论结果都去 compose_answer”,X999 分支会读取不存在的 status,产生异常或错误答复。若给同一节点同时接无条件边与条件边,还要小心多个后继节点都被触发:普通边不是“默认分支”。LangGraph 的边文档明确普通边可让多个后继在下一轮运行;想二选一就用条件边表达,别假定其中一条会自动被取消。

Reducer、循环和持久化容易混淆的边界 ​

Reducer 解决的是“更新怎么合并”,不是“下一步去哪”。 events 用 add 是累积;status 默认覆盖。要清空一个累积列表,简单返回 [] 并不会清除旧值,因为旧列表加空列表仍是旧列表;官方文档提供 Overwrite 等方式在需要时绕过累积规则。Graph API:重置 reducer 字段 并行节点若在同一轮更新同一字段,还需选合适 reducer 或调整图结构;否则可能因冲突无法安全合并。

循环由边连回节点,但必须有出口。 例如查询失败后去“改写查询词”,再回到查询节点;State 需存补查次数,条件边要在成功、达到次数上限或无法安全继续时去 END 或人工节点。仅仅连一条回边会让图反复运行,直到运行限制报错。官方循环指南演示了条件终止。本文代码没有循环,因为一次订单查询的教学流程不需要补查。

State 快照不等于自动持久化。 compile() 可以接入 checkpointer,把执行进度保存到支持的存储中,供中断后继续;未配置时,不能声称服务重启后会恢复。即使配置了检查点,外部写操作仍要设计幂等:节点可能在恢复或重试时重新执行,重复创建工单或退款会造成实际副作用。Graph API:编译和节点重执行对此有说明。本文只有只读查询和字符串答复,不演示写入外部系统。

面试时可以怎样回答 ​

“LangGraph 的 State 是一次图运行的共享数据快照,定义字段及各字段如何合并更新;Node 是做事的函数,读当前 State,返回自己修改的字段;Edge 在节点完成后决定下一个节点,固定边直接连接,条件边根据更新后的 State 分流。比如查订单 A123,起点状态只有订单号;查询节点写入 found=True 和 status=已发货;条件边选写答复节点;它只写 answer,旧订单字段仍在 State。默认字段新值覆盖旧值,事件列表若用 reducer 可以逐步追加。查不到订单就走缺资料分支,服务超时要另设错误分支。图要先编译再运行;State 是否能跨请求恢复取决于是否配置持久化,不能把它等同于模型记忆。”

若被追问“Node 一定要返回完整 State 吗”,回答是:“不用,返回部分更新即可;框架按每个字段的 reducer 合并,未指定 reducer 的字段默认覆盖。”若被追问“条件边会调用大模型吗”,回答是:“不一定;它可以像例子一样只是一个普通 Python 判断。是否在节点或路由中调用模型,是应用自己的设计。”

参考资料 ​

最后更新2026-09-26
难度P1
频率high
阅读20 min
主题langgraph / state / node
觉得有帮助?把这个链接转给正在求职的朋友 · 用 Ctrl + K 全站搜索其它题