让 LlamaIndex Workflows 崩了也能续跑：快照、检查点循环与崩溃恢复实操

import asyncio
from pydantic import BaseModel, Field
from workflows import Context, Workflow, step
from workflows.events import Event, StartEvent, StopEvent

class MyEvent(Event):
    msg: list[str]

class RunState(BaseModel):
    num_runs: int = Field(default=0)

class MyWorkflow(Workflow):
    @step
    async def start(self, ctx: Context[RunState], ev: StartEvent) -> MyEvent:
        async with ctx.store.edit_state() as state:
            state.num_runs += 1
        return MyEvent(msg=[ev.input_msg] * state.num_runs)

    @step
    async def process(self, ctx: Context[RunState], ev: MyEvent) -> StopEvent:
        data_length = len("".join(ev.msg))
        new_msg = f"Processed {len(ev.msg)} times, data length: {data_length}"
        return StopEvent(result=new_msg)

async def main():
    workflow = MyWorkflow()
    ctx = Context(workflow)
    result = await workflow.run(input_msg="Hello, world!", ctx=ctx)
    print("Workflow result:", result)

asyncio.run(main())

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

# 持久化快照（示例存文件，生产建议落数据库）
import json
with open("my-run.json", "w") as f:
    f.write(json.dumps(handler.ctx.to_dict()))

# ......新进程......
w2 = MyWorkflow()
with open("my-run.json") as f:
    ctx = Context.from_dict(w2, json.loads(f.read()))
result = await w2.run(ctx=ctx)  # 从恢复的状态继续

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:
        # 一个步骤刚跑完：此刻的状态是一致的检查点
        save_snapshot("my-run.json", handler.ctx.to_dict())
result = await handler

from typing import Annotated
from workflows.resource import Resource

class WorkItem(Event):
    item_id: int

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

def get_client():
    return MyApiClient(...)  # 重资产：连接、凭证都在这里

class EnrichBatch(Workflow):
    @step
    async def dispatch(self, ev: StartEvent) -> list[WorkItem]:
        return [WorkItem(item_id=i) for i 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)  # 可重复的重活
        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})

# run_batch.py：有快照就恢复，没有就冷启动（对应官方 durable 指南原文结构）
async def run_batch(item_ids):
    wf = EnrichBatch()
    if os.path.exists(CKPT):
        # 恢复：待办事件与扇入状态已在上下文里，
        # 不要再发 StartEvent
        ctx = Context.from_dict(wf, load_json(CKPT))
        handler = wf.run(ctx=ctx)
    else:
        handler = wf.run(item_ids=item_ids)
    await drive_with_checkpoints(handler)  # Step 3 的检查点循环

# 老包（llama_index.core.workflow）的内建检查点用法
from llama_index.core.workflow.checkpointer import WorkflowCheckpointer

ckptr = WorkflowCheckpointer(workflow=JokeFlow())
await ckptr.run(topic="chemistry")
# 按步骤名筛检查点，再从指定检查点续跑
after_gen = ckptr.filter_checkpoints(last_completed_step="generate_joke")[0]
await ckptr.run_from(checkpoint=after_gen)