StateGraph 与 Checkpointer：理解 LangGraph 的状态机制

create_agent 好用，但它背后是 LangGraph 的 StateGraph。搞懂 StateGraph，你才能做更复杂的自定义流程。

什么是 StateGraph

StateGraph 是一个有限状态机。你定义状态结构、节点函数、边的连接关系，框架按图执行。

```python from typing import Annotated from operator import add from langgraph.graph import StateGraph, START, END from typing_extensions import TypedDict

class State(TypedDict): foo: int bar: Annotated[list[str], add]

def node_a(state: State) -> State: return {"foo": "a", "bar": ["a"]}

def node_b(state: State) -> State: return {"foo": "b", "bar": ["b"]}

workflow = StateGraph(State) workflow.add_node(node_a) workflow.add_node(node_b) workflow.add_edge(START, "node_a") workflow.add_edge("node_a", "node_b") workflow.add_edge("node_b", END) ```

Annotated[list[str], add] 是个有意思的设计。bar 字段用 add 做 reducer，每个节点返回的值会追加而不是覆盖。执行完 node_a 和 node_b，bar 变成 ["a", "b"]。foo 没有 reducer，所以被 node_b 覆盖成 "b"。

加上 Checkpointer

```python from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver() config = {"configurable": {"thread_id": "1"}}

graph = workflow.compile(checkpointer=checkpointer) result = graph.invoke({"foo": ""}, config=config) print(result) # {'foo': 'b', 'bar': ['a', 'b']}

state = graph.get_state(config=config) print(state) ```

编译时传入 checkpointer，每个节点执行完都会保存状态快照。

Checkpointer 的四大用途

记忆管理，多轮对话的消息持久化。时间旅行，回到某个历史状态重新执行。暂停恢复，Human-in-the-Loop 的基础。容错，节点失败后从上一个检查点恢复。

小结

StateGraph 给你完全的控制权。状态怎么流转、节点怎么连接、哪些字段追加哪些覆盖，都是你说了算。create_agent 是 StateGraph 的高层封装，遇到它满足不了的场景，直接用 StateGraph 就行。