《AI Agent 高级编程》6.1 为什么需要持久化执行：崩溃/重启/长任务改造

import os, atexit

def work():
    try:
        print("进入 try: 开始干活", flush=True)
        os._exit(1)                 # 硬崩溃
    except Exception as e:
        print("except 捕获:", e)     # 捕获不到
    finally:
        print("finally 执行了")      # 不会执行
        open("/tmp/ab_dur/finally_ran.txt", "w").write("ran")

atexit.register(lambda: print("atexit 钩子执行了"))   # 不会执行
work()

进入 try: 开始干活
退出码=1
finally_ran.txt 存在? 否

"""持久化执行最小实现：append-only 日志 + 可重放步骤。
第一次运行在第 3 步硬崩溃，第二次运行从日志恢复。"""
import json, os, sys

RUN = "/tmp/ab_dur/c1"
JOURNAL, CRASH, MLOG = RUN + "/journal.jsonl", RUN + "/crashed", RUN + "/model.log"


def fake_model(prompt):
    with open(MLOG, "a") as f:
        f.write(prompt + "\n")          # 记一次真实"模型调用"
    return {"plan": "拆成 Q1/Q2 两个子问题", "summarize": "结论: 跳过已完成步骤"}[prompt]


def fake_search(q):
    return {"durable execution": "Temporal/DBOS 用事件溯源日志驱动工作流",
            "deterministic replay": "重放要求相同输入产生相同输出"}[q]


class Durable:
    def __init__(self):
        self.done = {}
        if os.path.exists(JOURNAL):
            for line in open(JOURNAL, encoding="utf-8"):
                if line.strip():
                    r = json.loads(line)
                    self.done[r["step"]] = r["output"]
        self.replayed, self.executed = [], []

    def step(self, name, fn):
        if name in self.done:            # 命中日志 -> 重放，不真执行
            self.replayed.append(name)
            return self.done[name]
        out = fn()
        self.executed.append(name)
        with open(JOURNAL, "a", encoding="utf-8") as f:
            f.write(json.dumps({"step": name, "output": out}, ensure_ascii=False) + "\n")
            f.flush(); os.fsync(f.fileno())      # 落盘后才算"完成"
        return out


def search_q2():
    if not os.path.exists(CRASH):        # 首次运行：第 3 步硬崩溃
        open(CRASH, "w").write("x")
        print("!! 第 3 步 [search:q2] 执行中硬崩溃 os._exit(1)")
        sys.stdout.flush(); os._exit(1)
    return fake_search("deterministic replay")


def run_task():
    d = Durable()
    d.step("plan",      lambda: fake_model("plan"))
    d.step("search:q1", lambda: fake_search("durable execution"))
    d.step("search:q2", search_q2)                    # <-- 崩溃点
    d.step("summarize", lambda: fake_model("summarize"))
    d.step("write_report", lambda: "report.md 已写入")
    return d


if __name__ == "__main__":
    d = run_task()
    n = sum(1 for _ in open(MLOG)) if os.path.exists(MLOG) else 0
    print("已重放(来自日志):", d.replayed)
    print("本次真执行:      ", d.executed)
    print("累计真实模型调用:", n, "次")

!! 第 3 步 [search:q2] 执行中硬崩溃 os._exit(1)
退出码=1

已重放(来自日志): ['plan', 'search:q1']
本次真执行:       ['search:q2', 'summarize', 'write_report']
累计真实模型调用: 2 次
退出码=0

{"step": "plan", "output": "拆成 Q1/Q2 两个子问题"}
{"step": "search:q1", "output": "Temporal/DBOS 用事件溯源日志驱动工作流"}
{"step": "search:q2", "output": "重放要求相同输入产生相同输出"}
{"step": "summarize", "output": "结论: 跳过已完成步骤"}
{"step": "write_report", "output": "report.md 已写入"}

# naive.py 的核心：没有日志，每次调用都真执行
def run_task():
    plan = fake_model("plan")          # 每次运行都重新调用模型
    q1 = fake_search("durable execution")
    q2 = search_q2()                   # 第 3 步崩溃
    summ = fake_model("summarize")
    return plan, q1, q2, summ