zhuangang/topics/ai/langgraph/01-state-graph.md

id, title, type, domain, stack, tags, level, prerequisites, est_hours, related, week, order

id	title	type	domain	stack	tags	level	prerequisites	est_hours	related	week	order
ai.langgraph.state-graph	LangGraph 状态图：状态、节点、边、条件路由和流式输出	lesson	ai	langgraph	LangGraph StateGraph reducer 条件边 流式输出 stream_mode	intermediate	ai.function-calling.tool-loop	3	ai.langgraph.persistence-hitl ai.agent-design.react-planning ai.langgraph.interview	11	1

LangGraph 状态图：状态、节点、边、条件路由和流式输出

基于 langgraph 1.2 编写，API 变化较快，以官方文档为准：https://docs.langchain.com/oss/python/langgraph/overview

解释说明

LangGraph 把一个 Agent / 工作流写成有状态的图：

概念	说明
State（状态）	一个 TypedDict（或 pydantic 模型），所有节点共享。节点返回“要更新的字段”，而不是整个状态
Reducer（合并函数）	决定字段怎么更新。默认覆盖；Annotated[list, operator.add] 表示追加；消息列表常用 add_messages（按消息 id 追加或替换）
Node（节点）	普通函数：输入状态，返回部分更新。可以调用模型、工具、数据库
Edge（边）	固定边 add_edge(a, b)；条件边 add_conditional_edges(a, 路由函数)，路由函数返回下一个节点名
START / END	入口和出口
compile	编译成可执行对象，支持 invoke / stream / ainvoke / astream；传入 checkpointer 后支持持久化（下一课）

为什么不直接写 while 循环？手写循环（第 10 周）能帮你理解原理；但当流程有分支、并行、人工审批、失败续跑时，LangGraph 把状态保存、恢复、流式事件、可视化都提供好了。

流式输出的几种模式

stream_mode	输出内容	用途
"values"	每一步之后的完整状态	调试
"updates"	每个节点返回的更新	前端展示“当前在哪一步”
"messages"	大模型逐 token 输出	打字机效果
"custom"	节点里用 get_stream_writer() 主动发送的任意数据	进度条、中间结果

可以同时订阅多种：stream_mode=["updates", "custom"]，返回 (模式, 数据) 元组。后端把这些事件转成 SSE 发给前端，就实现了“AI 原生”界面里的流式输出 + 任务进度。

案例

工单处理流程：分类 → 起草回复 → 金额大的退款需要人工审批 → 发送。分类结果用 pydantic 校验（结构化输出），不合法时走“转人工”分支；前端订阅 custom 流显示“正在分类 / 正在起草 / 等待审批”。

代码示例

最小的条件路由图（可直接运行）：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph


class State(TypedDict):
    question: str
    route: str
    answer: str
    steps: Annotated[list[str], operator.add]     # 追加而不是覆盖


def route(state: State):
    r = "billing" if "发票" in state["question"] else "general"
    return {"route": r, "steps": [f"route:{r}"]}


def billing(state: State):
    return {"answer": "开票请在订单详情页申请。", "steps": ["billing"]}


def general(state: State):
    return {"answer": "已转给客服。", "steps": ["general"]}


g = StateGraph(State)
g.add_node("route", route)
g.add_node("billing", billing)
g.add_node("general", general)
g.add_edge(START, "route")
g.add_conditional_edges("route", lambda s: s["route"], ["billing", "general"])
g.add_edge("billing", END)
g.add_edge("general", END)
app = g.compile()

for update in app.stream({"question": "怎么开发票", "steps": []}, stream_mode="updates"):
    print(update)
print(app.invoke({"question": "怎么开发票", "steps": []})["steps"])   # ['route:billing', 'billing']

完整的工单图（结构化输出校验、自动重试、人工审批、流式进度、崩溃续跑）见 examples/ticket-flow。

易错点

• 节点返回了整个状态并修改了列表：应该只返回要更新的字段；列表字段要用 reducer 追加，不要原地 append。
• 条件边函数里修改状态：路由函数只负责返回下一个节点名，状态修改放在节点里。
• 条件边没有声明可能的目标（第三个参数），图的可视化和校验不完整。
• 图中有循环却没有退出条件：设置最大步数（recursion_limit，以及状态里自己的计数器）。
• 混用同步和异步：节点是 async def 时要用 ainvoke / astream。

面试常问

• LangGraph 和 LangChain 的关系？为什么需要 LangGraph？
• reducer 是什么？消息列表为什么用 add_messages？
• 怎么把图的执行过程流式推给前端？