《AI Agent 高级编程》6.2 确定性重放与副作用：模型调用变可重放步骤

import time, random, uuid, os
print("time.time() =", time.time())
print("random()    =", random.random())
print("uuid4()     =", uuid.uuid4())
print("hash('durable') =", hash("durable"))
print("set order   =", list({"alpha","beta","gamma","delta","epsilon"}))
print("PYTHONHASHSEED =", os.environ.get("PYTHONHASHSEED"))

# 进程 A
time.time() = 1791662290.1723719
random()    = 0.8060616367153092
uuid4()     = 0f266e42-0355-4fc2-a4f5-e57445896006
hash('durable') = -2797282880968727216
set order   = ['epsilon', 'beta', 'delta', 'alpha', 'gamma']
PYTHONHASHSEED = None
# 进程 C（PYTHONHASHSEED=0）
hash('durable') = -4978027187027173919
set order   = ['alpha', 'beta', 'gamma', 'delta', 'epsilon']
PYTHONHASHSEED = 0

"""确定性重放：把模型调用录到 tape；重放命中 tape 就不再真调。"""
import json, os, sys, hashlib, time

RUN_DIR = "/tmp/ab_dur/tape"
TAPE = RUN_DIR + "/tape.jsonl"
EFFECTS = RUN_DIR + "/effects.jsonl"
OUT_FILE = RUN_DIR + "/report.txt"
os.makedirs(RUN_DIR, exist_ok=True)
MODE = sys.argv[1] if len(sys.argv) > 1 else "record"


def key(prompt): return hashlib.sha256(prompt.encode()).hexdigest()[:12]


def real_model(prompt):                       # 带非确定性: 时间戳
    return f"{prompt} => answer@{time.time():.6f}"


class RecordedModel:
    def __init__(self, mode):
        self.mode = mode
        self.tape = {}
        if os.path.exists(TAPE):
            for line in open(TAPE, encoding="utf-8"):
                if line.strip():
                    r = json.loads(line)
                    self.tape[r["key"]] = r["output"]
        self.real_calls = 0

    def complete(self, prompt):
        k = key(prompt)
        if self.mode == "replay":
            if k not in self.tape:
                raise KeyError(f"重放缺记录: {prompt}")
            return self.tape[k]                   # 命中 tape, 不真调
        out = real_model(prompt)                  # 真调
        self.real_calls += 1
        with open(TAPE, "a", encoding="utf-8") as f:
            f.write(json.dumps({"key": k, "prompt": prompt, "output": out},
                               ensure_ascii=False) + "\n")
        return out


def notify(msg):                              # 有副作用的步骤, 幂等
    ik = hashlib.sha256(msg.encode()).hexdigest()[:12]
    done = set()
    if os.path.exists(EFFECTS):
        for line in open(EFFECTS, encoding="utf-8"):
            if line.strip():
                done.add(json.loads(line)["ik"])
    if ik in done:
        return f"[skip] 副作用已生效过 ik={ik}"
    with open(EFFECTS, "a", encoding="utf-8") as f:
        f.write(json.dumps({"ik": ik, "msg": msg}, ensure_ascii=False) + "\n")
    return f"[exec] 已发送通知 ik={ik}"


def workflow(m):
    a = m.complete("总结持久化执行")
    b = m.complete("总结确定性重放")
    eff = notify("周报已生成")
    with open(OUT_FILE, "w", encoding="utf-8") as f:
        f.write(a + "\n" + b + "\n")
    return a, b, eff


if __name__ == "__main__":
    m = RecordedModel(MODE)
    a, b, eff = workflow(m)
    h = hashlib.sha256((a + b).encode()).hexdigest()[:16]
    print(f"mode={MODE}  真实模型调用={m.real_calls} 次")
    print(f"副作用: {eff}")
    print(f"输出摘要 hash = {h}")

mode=record  真实模型调用=2 次
副作用: [exec] 已发送通知 ik=177a045e952d
输出摘要 hash = f1caf298eeb6b74f

mode=replay  真实模型调用=0 次
副作用: [skip] 副作用已生效过 ik=177a045e952d
输出摘要 hash = f1caf298eeb6b74f

ik = hashlib.sha256(msg.encode()).hexdigest()[:12]   # 由业务内容派生

症状1 key 在 tape 中? False
症状1 -> KeyError('重放缺记录: 从未录制过的问题')

症状2 两次 key: 126b883a9c89 fe97ddd3ff3f 相同? False

def logical_key(step, args):
    stable = {k: v for k, v in args.items() if k not in ("ts", "request_id")}
    payload = step + "|" + "|".join(f"{k}={stable[k]}" for k in sorted(stable))
    return hashlib.sha256(payload.encode()).hexdigest()[:12]

原始 prompt 含 ts -> key 稳定? False
逻辑 key（剔除 ts/request_id）-> 稳定? True