Durable Agent Execution

Durable Execution:把 agentic loop 跑成可重放的日志

背景与原理

普通 agent loop 的状态活在进程内存里:messages 数组、循环计数、局部变量。进程一死,全没了。Durable execution 引擎(Temporal / Restate / DBOS)换了个模型:你的 workflow 函数每执行一个「持久步骤」,引擎就把这一步的输入和输出写进一条 journal(event log)。进程崩溃后,引擎在新 worker 上重新运行同一个函数,但已经在 journal 里的步骤不再真正执行——直接把记录的返回值replay 回去,直到追上崩溃点,再继续往下跑。

对 agent 这个模型是量身定做:LLM 调用又慢又贵(1–30 秒、真金白银),工具调用可能有副作用,任务动辄跨分钟。Temporal 每个持久步骤只加约 10–50ms 开销——相对 LLM 调用可忽略,换来的是「worker 随便被杀,agent 都能从上一个成功步骤接着走」。DBOS 更轻,直接把 journal 存进你已有的 Postgres,一个 pip 包、不需要额外的 broker 或控制面。

实战示例

DBOS 的写法最能看清「哪些是持久步骤」——@DBOS.workflow 是可重放的编排,@DBOS.step 是被 journal 记录的原子步骤:

from dbos import DBOS

@DBOS.step()                       # 每个 step 的返回值被写进 Postgres journal
def call_llm(msgs): return client.messages.create(...)

@DBOS.step()
def run_tool(name, args): return TOOLS[name](args)

@DBOS.workflow()                   # 崩溃后从上一个已完成 step 重放续跑
def agent(task):
    msgs = [{"role":"user","content":task}]
    for _ in range(30):
        r = call_llm(msgs)               # 已成功的轮次 replay 时不重发
        if r.stop_reason == "end_turn": return r
        out = run_tool(...)              # 崩在这里,重启后前面全部跳过
        msgs += [...]

你几乎没改 agent 逻辑——只是给「昂贵/有副作用」的操作套上 step,给循环套上 workflow。执行语义就从「崩了全丢」变成「崩了续跑」。

Checkpoint ≠ Durable Execution:快照救不了「节点跑一半崩」

背景与原理

很多人以为「我用了 LangGraph checkpointer / 存了 state 到 DB」就等于 durable。不是。Checkpoint 是在每个 super-step(节点边界)之后存一张状态快照;它给你的是记忆、human-in-the-loop、和节点级的容错。但崩溃的粒度是残酷的:如果一个节点内部先调了 LLM、又发了邮件,然后在第三行崩了——重启时整个节点从第一行重放,LLM 会重调、邮件会重发。Diagrid 那篇 Checkpoints Are Not Durable Execution 把这点说透了:快照记的是「到过哪些格子」,不是「格子里做过哪些不可撤销的动作」。

LangGraph 自己暴露了三档 durability 来让你选权衡:exit(只在退出时落盘,最快,崩了不可恢复)、async(异步落盘,有小概率丢最后一步)、sync(每步同步落盘再继续,最慢但最稳)。选 sync 也只把恢复粒度做到节点边界,节点内的原子性仍然要你自己保证。这就是它和 Temporal/Restate 这类步骤级 journal 引擎的本质差距。

实战示例

# ❌ 危险:一个节点里塞了两个有副作用的动作
def node_book_trip(state):
    flight = charge_card(state)      # 副作用 1:扣款成功
    email  = send_confirm(flight)    # 这里崩 → 重放时 charge_card 再扣一次!
    return {"booking": email}

# ✅ 修法 A:拆成单副作用节点,让 checkpoint 边界=副作用边界
graph.add_node("charge", charge_node)   # 每个节点最多一个不可逆动作
graph.add_node("email",  email_node)
# 编译时选最稳一档,并绑定 thread_id 作为续跑句柄
app = graph.compile(checkpointer=saver, durability="sync")
app.invoke(inp, config={"configurable":{"thread_id":uid}})

核心纪律:checkpoint 引擎里,「一个节点 = 一个可安全重放的单元」。把节点切到「最多含一个不可逆副作用」,快照边界才真正对齐了危险边界。要么就升级到步骤级 journal 的引擎(§03)。

副作用的 Exactly-Once:LLM 和工具调用必须 journal 化 + 幂等

背景与原理

durable execution 的「exactly-once」有个容易被误读的边界:引擎保证的是step 的返回值在 journal 里只被物化一次——重放时不会二次调用你的函数体。Restate 的 ctx.run()、Temporal 的 activity、DBOS 的 @DBOS.step 都是这个语义。但引擎无法阻止被包裹的那次调用本身对外部世界产生重复效果——如果你在 step 第一次执行时就同时扣了两次款,journal 只会忠实记下这个结果。

所以真正的工程纪律是两层:(1)把每个外部调用都包成一个 step/activity,让引擎负责「不重放执行」;(2)对非幂等的外部 API,在 step 内部再带上 idempotency key,让下游服务负责「重复请求去重」。Restate 甚至支持请求级 idempotency key 自动去重。LLM 调用虽然「无外部副作用」,也必须包成 step——否则 replay 时会重新烧一次 token,贵得毫无必要。

实战示例

# 幂等键 = workflow 内稳定可复现的 ID(不能用随机数/时间戳!)
@DBOS.step()
def charge(user, amount, idem_key):
    return stripe.PaymentIntent.create(
        amount=amount, customer=user,
        idempotency_key=idem_key)      # ← 下游 Stripe 去重,重放/重试都只扣一次

@DBOS.workflow()
def checkout(user, cart):
    # key 由 workflow 身份派生,replay 时值不变 → 天然幂等
    key = f"pay-{DBOS.workflow_id}"
    charge(user, cart.total, key)      # 崩溃重放:step 已记录 → 不再执行
    send_receipt(user)                 # 若在此崩,charge 不会二次触发

关键在幂等键的来源:它必须从 workflow 的持久身份(workflow_id + 步骤序号)派生,这样每次 replay 算出来的 key 都一样。用 uuid4() 或时间戳当 key 是最经典的坑——每次重放生成新 key,下游认不出是同一笔,去重失效。

Replay 确定性 vs LLM 非确定性:把不确定赶进 activity

背景与原理

replay 能工作的前提是:重新运行 workflow 函数时,它的控制流走得和第一次逐字节一致——否则重放到一半,代码走到了 journal 里不存在的分支,引擎就懵了。这要求 workflow 主体里不能直接出现:随机数、now()/时钟、直接的网络/文件 IO、以及……直接 LLM 调用。这些都是「不确定源」。这正好和 Day 56 讲的采样确定性是两种不同的确定性:那期讲「同 prompt 同输出」,这里讲「同函数重放走同一条路」。

解法统一而优雅:把每一个不确定源都包进一个 step/activity。第一次执行时它真跑、结果进 journal;replay 时它不跑、直接吐 journal 里的旧值——于是从 workflow 的视角看,连「掷骰子」都变成了确定的。LLM 的非确定输出被这一层「冻结」进日志,workflow 主体就重新变回一个纯粹、可重放的状态机。这也顺带解锁了 durable execution 最杀的一个能力:长时暂停——用 durable timer + signal,agent 可以「睡」几天等人审批,期间不占任何常驻进程,审批信号一到就从睡点唤醒续跑。

实战示例

# Temporal 风格:workflow 确定,不确定性走 activity
@workflow.defn
class ApprovalAgent:
    def __init__(self): self._approved = None
    @workflow.signal                       # 外部审批用 signal 打进来
    def approve(self, ok): self._approved = ok
    @workflow.run
    async def run(self, task):
        plan = await workflow.execute_activity(  # LLM 调用 = activity
            call_llm, task, start_to_close_timeout=TIMEOUT)
        # 睡到「有人 approve」或 3 天超时——零常驻进程
        await workflow.wait_condition(
            lambda: self._approved is not None,
            timeout=timedelta(days=3))
        if self._approved:
            await workflow.execute_activity(apply_plan, plan, ...)

注意 call_llm 从不在 workflow 体里直接调——它永远是个 activity。wait_condition + signal 让 agent 挂起数天而不烧一分钱算力,这是「异步人审批队列」(Day 34)最干净的落地方式。