事件溯源与确定性重放

更新 2026-10-05

我的学习 →

正在读取学习记录…

事件溯源是把“发生过什么”作为可恢复的记录。退款任务从申请到批准再到完成，每次变化都追加一条业务事件；读取这些事件时，可以还原任务在任意一步的状态。这里的事件是已发生的业务事实，例如“审批通过”，不是发给模型的聊天消息。

确定性重放要求相同历史得到相同状态。若重放时重新读当前时间、重新抽随机数或再次问模型，结果就可能变化；已完成的外部调用应记录结果，恢复时使用历史回执。检查点保存某一时刻的状态，事件历史保留状态如何走到这里，两者可以一起使用。

用退款历史还原三个时刻

固定申请金额为180元，依次发生申请、批准、记录退款回执三个事件。每条事件有从1开始的序号。程序先重放第一条，再重放前两条和完整历史，最后故意缺少第二条。你可以直接看到历史前缀与状态之间的对应关系。

事件前缀逐步还原状态，重放读取已存回执，不重新调用退款服务 — 图1：重放读取已存回执，不重新调用退款服务。

保存完整程序并运行

先在 PowerShell 创建并进入学习目录：

mkdir "F:\学习\l4-replay" -Force
cd "F:\学习\l4-replay"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

新建 F:\学习\l4-replay 文件夹，把下面完整代码保存为 replay_demo.py，然后在PowerShell运行：

cd "F:\学习\l4-replay"
ls
python -X utf8 replay_demo.py

先在 PowerShell 创建并进入学习目录：

mkdir "F:\学习\l4-replay" -Force
cd "F:\学习\l4-replay"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

"""保存业务事件，再按原顺序还原退款任务的状态。"""
from copy import deepcopy


def apply(state, event):
    # 事件序号必须连续，缺失和乱序不能悄悄跳过。
    if event["seq"] != state["seq"] + 1:
        raise ValueError("事件序号不连续")
    kind = event["type"]
    if kind == "requested" and state["status"] == "new":
        state["amount"] = event["amount"]
        state["status"] = "waiting"
    elif kind == "approved" and state["status"] == "waiting":
        state["status"] = "approved"
    elif kind == "refund_recorded" and state["status"] == "approved":
        # 使用历史保存的回执；重放阶段不再调用退款接口。
        state["receipt"] = event["receipt"]
        state["status"] = "completed"
    else:
        raise ValueError("事件不符合当前状态")
    state["seq"] = event["seq"]


def replay(events):
    state = {"seq": 0, "status": "new", "amount": 0, "receipt": None}
    for event in events:
        apply(state, deepcopy(event))
    return state


def run_demo():
    events = [
        {"seq": 1, "type": "requested", "amount": 180},
        {"seq": 2, "type": "approved"},
        {"seq": 3, "type": "refund_recorded", "receipt": "refund-42"},
    ]
    lines = []
    for end in (1, 2, 3):
        state = replay(events[:end])
        lines.append(f"重放到事件{end}: {state['status']}, 金额={state['amount']}, 回执={state['receipt']}")
    lines.append(f"重复重放相同历史: {replay(events) == replay(events)}")
    try:
        replay([events[0], events[2]])
    except ValueError as error:
        lines.append(f"缺失事件: {error}")
    return lines


if __name__ == "__main__":
    print("\n".join(run_demo()))

完整固定输出如下：

重放到事件1: waiting, 金额=180, 回执=None
重放到事件2: approved, 金额=180, 回执=None
重放到事件3: completed, 金额=180, 回执=refund-42
重复重放相同历史: True
缺失事件: 事件序号不连续

从输出追到事件处理函数

第一行的 waiting 是待审批，金额来自申请事件。第二行进入 approved，但没有回执，因为批准与实际退款是两件事。第三行进入 completed，历史中的 refund-42 成为当前回执。第四行比较两次重放结果，它们相等；第五行说明缺失批准事件时，程序拒绝把任务恢复为已完成。

apply 每次先检查序号，再检查事件类型与当前状态的组合。序号防止缺漏，状态条件防止业务顺序非法。把第三个事件的序号改成2仍然不能让它直接跟在申请后，因为退款回执要求当前状态已批准。两道检查回答不同问题。

重放为什么不能再次退款

refund_recorded 分支从事件复制回执，反复查看历史时复用同一结果。执行阶段调用工具并把结果追加到历史；重放阶段则把历史结果应用到状态。工具调用成功但事件未写入时，可以按业务编号查询原操作。下一篇的重试与幂等机制会进一步处理这种响应和记录缺失的情况。

Agent 的模型建议也有类似问题。记录工具名、经校验的参数和工具结果，能够恢复已经完成的步骤；若想比较新模型，则应另建运行标识，不能把新结果偷偷写进旧历史。事件内容含敏感信息时需要脱敏与访问控制。

历史版本变化以后怎么办

代码接受三种固定事件。旧历史若出现无法识别的事件，会报业务状态错误。扩展事件格式时可以增加版本字段，由不同版本的解析函数处理；变更业务规则后也要验证旧历史能否保持原语义。存量历史不能因为更新代码而被解释成另一笔退款。

事件数量多时，可以保存某个序号的检查点，再应用后续事件。检查点必须注明历史序号，恢复时从下一条开始，否则同一事件会应用两次。示例从第一条全量重放，便于看清连续性；没有实现快照裁剪和多工作者写入。

检查历史缺漏和非法顺序

配套测试覆盖完整历史、历史前缀、缺失事件、重复事件与未经批准的回执。重复消息应在写入事件日志时以唯一事件标识去重，而不是在重放时随意跳过所有重复项；这里直接拒绝重复序号，避免掩盖损坏的历史。

Temporal 工作流源码可以作为进一步阅读入口：沿历史事件、重放和外部活动的边界查找实现，重点观察外部活动结果如何进入历史。示例使用内存事件列表，接数据库后还要保证序号分配和事件追加的原子性。

改一处输入，核对机制变化

把 run_demo 中申请事件的 "amount": 180 改为 120，保存后运行 python -X utf8 replay_demo.py。前三行金额都变为120，回执仍为refund-42；重放中的金额来自申请事件。恢复180后重跑，核对原输出。

运行配套测试

将下面测试保存为同一文件夹的 test_replay_demo.py。它使用固定场景检查状态与业务结果，可以直接运行：

python -X utf8 -m unittest -v test_replay_demo.py

完整测试：test_replay_demo.py

import unittest
from replay_demo import apply, replay, run_demo


class ReplayTests(unittest.TestCase):
    def test_prefix_and_result(self):
        # 相同历史重复恢复，回执和原事件均保持一致。
        events = [{"seq": 1, "type": "requested", "amount": 180},
                  {"seq": 2, "type": "approved"},
                  {"seq": 3, "type": "refund_recorded", "receipt": "r-1"}]
        self.assertEqual(replay(events[:1])["status"], "waiting")
        self.assertEqual(replay(events)["receipt"], "r-1")
        self.assertEqual(replay(events), replay(events))
        self.assertEqual(events[0]["amount"], 180)

    def test_gap_and_duplicate_are_rejected(self):
        event = {"seq": 1, "type": "requested", "amount": 180}
        with self.assertRaises(ValueError):
            replay([event, event])
        with self.assertRaises(ValueError):
            replay([{"seq": 2, "type": "approved"}])

    def test_wrong_transition_is_rejected(self):
        with self.assertRaises(ValueError):
            replay([{"seq": 1, "type": "refund_recorded", "receipt": "r"}])

    def test_demo_has_all_prefixes(self):
        self.assertEqual(len(run_demo()), 5)


if __name__ == "__main__":
    unittest.main()

课后练习展开习题收起习题

正在读取本地记录…

不调用模型，不执行代码。简答、代码和场景题仅保存答案并展示自查标准，不生成评分、徽章或认证。输入内容不会自动保存，请点击提交；不要填写个人信息或密钥。

练习 1 / 4 · v1.0.1

选择题 · 先判断边界

关于“事件溯源与确定性重放”，哪种做法更稳妥？

对应知识点：L4-replay

选择一个答案A. 先明确输入、证据和停止条件B. 默认输出永远正确C. 用一次成功推断全部场景

练习 2 / 4 · v1.0.0

简答题 · 解释失败

写出一个失败输入并说明记录字段。

对应知识点：L4-replay

你的回答

练习 3 / 4 · v1.0.0

代码题 · 边界测试

补充正常、边界、失败三组输入和预期结果。

对应知识点：L4-replay

你的代码或伪代码

练习 4 / 4 · v1.0.0

场景题 · 结果未知

证据不足时如何设计状态和人工接管？

对应知识点：L4-replay

你的回答