工作流默认是短命的。一旦 run() 返回，状态就消失了，下一次 run() 从零开始。但工作流往往正是你写下的那种临时并发过程：对上百个文档扇出、每块调用一次 LLM、再把结果收集起来。这样的运行，你不会希望进程跑到一半被杀掉时从零重来。

本页演示如何对这次运行做检查点，并在重启后恢复它。

对上下文做快照

运行是靠受控地消减事件向前推进的，因此它在每个步骤边界上都有明确定义的状态。这个状态就是仍在途的事件加上状态存储。Context.to_dict() 把它序列化，Context.from_dict() 把它重建，run(ctx=...) 则从那里继续 —— 即便是在另一个进程里：

w = MyWorkflow()
handler = w.run()
result = await handler

# Persist the snapshot
db.save("my-run", json.dumps(handler.ctx.to_dict()))

# ...new process...
w = MyWorkflow()
ctx = Context.from_dict(w, json.loads(db.load("my-run")))
result = await w.run(ctx=ctx)   # continues with the restored state

这样能熬过重启，但熬不过运行期间的崩溃。你手上的只是最后一次 to_dict() 时刻的状态。要能应对崩溃，就得随着运行推进持续做快照。

恢复时，被恢复的运行会重新派发那些仍在挂起的事件，并重建部分完成的扇入缓冲。已完成的步骤不会重跑，因为它们的产出已在恢复的状态里。而在你抓取状态时正处于执行中间的步骤会被回卷、从头再跑一次。恢复是至少一次语义，因此步骤的副作用必须可安全重复。

检查点循环

没有任何需要开启的内置检查器。工作流事件流会在每次步骤状态变化时发出一个内部事件，而你在看到某个步骤结束时做快照。stream_events(expose_internal=True) 会把这些内部事件暴露出来，而带 StepState.NOT_RUNNING 的 StepStateChanged 就标记了一个步骤的完成：

from workflows.events import StepStateChanged, StepState

handler = w.run()
async for ev in handler.stream_events(expose_internal=True):
    if isinstance(ev, StepStateChanged) and ev.step_state == StepState.NOT_RUNNING:
        db.save("my-run", json.dumps(handler.ctx.to_dict()))
result = await handler

要在崩溃后恢复，请加载最后一个快照并把它传给 run()：

w = MyWorkflow()
ctx = Context.from_dict(w, json.loads(db.load("my-run")))
result = await w.run(ctx=ctx)

在繁忙的运行里，你通常不必每个边界都写一次。每次快照都是一次序列化加一次写入，所以工作流很嘈杂时要做限流。这样一次崩溃最多损失这个间隔内的重做工作。

快照就是你抓取那一刻的状态。自触发写入的那个事件之后，其他 worker 可能已经向前推进了，但状态依然是自洽且可恢复的。

能熬过重启的并发扇出

下面这个工作流会扇出工作、并发执行各条目并收集结果。加上检查点后，跑到一半被杀掉也不用重做已完成的条目：

import json, os
from typing import Annotated
from workflows import Context, Workflow, step
from workflows.events import Event, StartEvent, StepState, StepStateChanged, StopEvent
from workflows.resource import Resource

class WorkItem(Event):
    item_id: int

class WorkDone(Event):
    item_id: int
    result: str

# Heavy inputs come from a Resource (see below). They aren't serialized into
# the snapshot; they're re-created on resume.
def get_client() -> MyApiClient:
    return MyApiClient(...)

class EnrichBatch(Workflow):
    @step
    async def dispatch(self, ev: StartEvent) -> list[WorkItem]:
        return [WorkItem(item_id=item_id) for item_id in ev.item_ids]

    @step(num_workers=8)
    async def enrich(
        self,
        ev: WorkItem,
        client: Annotated[MyApiClient, Resource(get_client)],
    ) -> WorkDone:
        result = await client.enrich(ev.item_id)   # the expensive, repeatable work
        return WorkDone(item_id=ev.item_id, result=result)

    @step
    async def collect(self, events: list[WorkDone]) -> StopEvent:
        return StopEvent(result={ev.item_id: ev.result for ev in events})

用检查点循环来驱动它，并在进程死掉时从最后一个快照恢复：

CKPT = "enrich-batch.json"

async def run_batch(item_ids):
    wf = EnrichBatch()
    if os.path.exists(CKPT):
        # Resume from the snapshot. The pending work and partial fan-in state
        # are already in the context, so don't send the StartEvent again.
        handler = wf.run(ctx=Context.from_dict(wf, json.load(open(CKPT))))
    else:
        handler = wf.run(item_ids=item_ids)

    async for ev in handler.stream_events(expose_internal=True):
        if isinstance(ev, StepStateChanged) and ev.step_state == StepState.NOT_RUNNING \
           and ev.name == "enrich":
            json.dump(handler.ctx.to_dict(), open(CKPT, "w"))
    return await handler

假设它在 100 个条目完成 60 个之后被杀掉。已完成的 enrich 调用属于被恢复的工作流状态，而挂起的 WorkItem 仍在挂起。恢复时 dispatch 不会重跑，已完成的条目不会被再次加工，而 list 扇入会在两次运行合计凑齐 100 个 WorkDone 时触发一次。

在途的工作确实会被重做。任意时刻最多有 num_workers 个 enrich 调用正在运行而尚未完成，恢复时它们每一个都会被回卷并重新加工。num_workers=8 时最多有 8 个条目会再跑一次。这浪费的是算力而不是产出正确性，也正是 enrich 必须可安全重复的原因。

让快照既便宜又正确

快照有多便宜、多可靠，取决于你往事件与状态里放什么。

工作流实例的属性不属于快照。它们对于与实例同生命周期的普通 Python 对象仍然有用，但不是持久状态。如果某个值需要熬过重启，请把它放进事件或状态存储。

把笨重或不可序列化的输入留在 Resource 里，而不是放进事件或状态。Resource 工厂只解析一次、缓存在工作流上，且永远不会被序列化进 context。恢复时它会通过再次调用工厂来重新创建。所以请把 API 客户端、模型句柄与大型参考数据放进资源，让事件只携带小的标识符。这样快照足够小，也能避开下面那条序列化规则。

在途事件与状态存储里的一切都必须可序列化。快照使用 JSON 序列化器，Pydantic 模型也不例外。一旦遇到无法编码的值，to_dict() 就会抛错，整个快照都失败，而不只是那个字段。所以不要把原始字节或打开的连接放在事件上。

用于持久化的运行时插件

检查点循环是你自己掌控的持久化。何时快照由你决定，恢复也只有几行代码。如果你不想管这些，可以换一个负责持久化与恢复的运行时插件。

默认运行时在进程内执行工作流。插件运行时可以在不改动工作流代码的前提下换一套执行模型。例如 DBOS 运行时会把步骤流转写入数据库日志，因此崩溃的工作流不需要你的检查点循环也能恢复。WorkflowServer 也能用同一个运行时插件执行工作流，于是对外提供的工作流获得相同的持久化与恢复行为。