《AI Agent 高级编程》5.1 读 LangGraph：图执行器与检查点机制

/tmp/agentbook/venv3/bin/python -c "import importlib.metadata as m; print(m.version('langgraph'), m.version('langgraph-checkpoint'), m.version('langchain-core'))"
/tmp/agentbook/venv3/bin/python -c "import langgraph.graph as g; print(g.__file__)"

1.2.14 4.2.0 1.6.9
/tmp/agentbook/venv3/lib/python3.14/site-packages/langgraph/graph/__init__.py

# langgraph/graph/state.py:1177
def compile(
    self,
    checkpointer: Checkpointer = None,
    *,
    cache: BaseCache | None = None,
    store: BaseStore | None = None,
    interrupt_before: All | list[str] | None = None,
    interrupt_after: All | list[str] | None = None,
    ...
) -> CompiledStateGraph[StateT, ContextT, InputT, OutputT]:

# langgraph/graph/state.py:1197
If provided, this `Checkpointer` serves as a fully versioned
"short-term memory" for the graph, allowing it to be paused, resumed,
and replayed from any point.

# langgraph/graph/state.py:1204
config = {"configurable": {"thread_id": "my-thread"}}
graph.invoke(inputs, config)

# langgraph/pregel/main.py:3026
# Similarly to Bulk Synchronous Parallel / Pregel model
# computation proceeds in steps, while there are channel updates.
# Channel updates from step N are only visible in step N+1
while loop.tick():
    for task in loop.match_cached_writes():
        loop.output_writes(task.id, task.writes, cached=True)
    for _ in runner.tick(
        [t for t in loop.tasks.values() if not t.writes],
        timeout=self.step_timeout,
        get_waiter=get_waiter,
        schedule_task=loop.accept_push,
    ):
        yield from _output(...)
    loop.after_tick()

# langgraph/pregel/_loop.py:614
def tick(self) -> bool:
    """Execute a single iteration of the Pregel loop."""
    if self.step > self.stop:
        self.status = "out_of_steps"
        return False
    self.tasks = prepare_next_tasks(
        self.checkpoint, self.checkpoint_pending_writes, self.nodes,
        self.channels, self.managed, self.config, self.step, self.stop, ...
    )
    if not self.tasks:
        self.status = "done"
        return False
    ...
    return True

# langgraph/pregel/_loop.py:698
def after_tick(self) -> None:
    # finish superstep
    writes = [w for t in self.tasks.values() for w in t.writes]
    ...
    self.updated_channels = apply_writes(
        self.checkpoint, self.channels, self.tasks.values(),
        self.checkpointer_get_next_version, self.trigger_to_nodes,
    )
    ...
    self.checkpoint_pending_writes.clear()
    self.is_replaying = False
    # save checkpoint
    self._put_checkpoint({"source": "loop"})

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver

class S(TypedDict):
    n: int

def a(state: S) -> dict:
    return {"n": state["n"] + 1}

def b(state: S) -> dict:
    return {"n": state["n"] * 10}

g = StateGraph(S)
g.add_node("a", a); g.add_node("b", b)
g.add_edge(START, "a"); g.add_edge("a", "b"); g.add_edge("b", END)
app = g.compile(checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "t1"}}

for chunk in app.stream({"n": 1}, cfg, stream_mode="updates"):
    print("SUPERSTEP", chunk)

SUPERSTEP {'a': {'n': 2}}
SUPERSTEP {'b': {'n': 20}}

# langgraph/checkpoint/base/__init__.py:92
class Checkpoint(TypedDict):
    v: int                     # 检查点格式版本，当前为 1
    id: str                    # 唯一且单调递增，可用来排序
    ts: str                    # ISO 8601 时间戳
    channel_values: dict[str, Any]   # 通道名 -> 快照值
    channel_versions: ChannelVersions  # 通道名 -> 单调递增版本串
    versions_seen: dict[str, ChannelVersions]  # 节点 ID -> 它见过的通道版本
    updated_channels: list[str] | None

# langgraph/pregel/_checkpoint.py:41
def empty_checkpoint() -> Checkpoint:
    return Checkpoint(
        v=LATEST_VERSION,
        id=str(uuid6(clock_seq=-2)),
        ts=datetime.now(timezone.utc).isoformat(),
        channel_values={}, channel_versions={}, versions_seen={},
    )

# langgraph/checkpoint/memory/__init__.py:442
c = checkpoint.copy()
thread_id = config["configurable"]["thread_id"]
checkpoint_ns = config["configurable"]["checkpoint_ns"]
values: dict[str, Any] = c.pop("channel_values")
for k, v in new_versions.items():
    self.blobs[(thread_id, checkpoint_ns, k, v)] = (
        self.serde.dumps_typed(values[k]) if k in values else ("empty", b"")
    )
self.storage[thread_id][checkpoint_ns].update(
    {
        checkpoint["id"]: (
            self.serde.dumps_typed(c),
            self.serde.dumps_typed(get_checkpoint_metadata(config, metadata)),
            config["configurable"].get("checkpoint_id"),  # parent
        )
    }
)

NUM_CHECKPOINTS 4
CK step=-1 source=input  next=('__start__',)
CK step=0  source=loop   next=('a',)
CK step=1  source=loop   next=('b',)
CK step=2  source=loop   next=()

# langgraph/pregel/_loop.py:1194
do_checkpoint = self._checkpointer_put_after_previous is not None and (
    exiting or self.durability != "exit"
)
self.checkpoint = create_checkpoint(
    self.checkpoint, self.channels if do_checkpoint else None,
    self.step, id=self.checkpoint["id"] if exiting else None,
    updated_channels=self.updated_channels,
    get_next_version=self.checkpointer_get_next_version if do_checkpoint else None,
    ...
)