事件溯源：除了现在是什么，还要知道怎么走到这里

本文是 Agent 工程里的经典架构 系列的一篇。文中的业务场景和代码为讲解设计，产品能力以所链接的官方资料为准。

排查线上问题时，数据库里一个 status=failed 经常帮不上忙。它可能表示库存不足、审批拒绝，也可能只是最后一次接口调用超时。如果中间过程都被新状态覆盖，我们只能去几台机器的日志里拼凑事实。等日志过期，问题就成了没有证据的讨论。

Event Sourcing 把状态变化保存为有序事件，状态可以由这些事件重新计算。它关心的不只是保留调试信息，而是让业务事实成为状态的来源。下面的工单案例是我的简化设计，用来解释这种选择会带来什么约束。

假设客服 Agent 负责跟进退款。与其反复覆盖一行状态，不如记录“申请已创建”“资料已补齐”“主管已批准”“退款已确认”。事件的命名要表达已经发生的事实。“请批准退款”是命令；“退款已批准”才是事件。收到命令不代表能够产生事件，中间还需要权限检查和业务判断。

例子用订单金额说明事件和投影的区别。事件保存价格当时的变化，投影负责算出当前金额。另一份投影可以统计修改次数，两者共享事实，却服务于不同查询。这样更容易看出事件流和数据库查询表之间的关系。

from copy import deepcopy


def reduce_order(events):
    state = {"amount": 0, "approved": False}
    for event in events:
        if event["type"] == "amount_set":
            state["amount"] = event["amount"]
            state["approved"] = False
        elif event["type"] == "approved":
            state["approved"] = True
        else:
            raise ValueError("未知事件类型")
    return state


def append(events, expected_version, event):
    # 单进程演示；生产环境需在存储端原子地校验与追加。
    if len(events) != expected_version:
        raise ValueError("版本冲突")
    events.append(deepcopy(event))


events = []
append(events, 0, {"type": "amount_set", "amount": 12000})
append(events, 1, {"type": "approved"})
assert reduce_order(events) == {"amount": 12000, "approved": True}
assert reduce_order(events[:1])["approved"] is False
try:
    append(events, 1, {"type": "amount_set", "amount": 9000})
except ValueError:
    pass
else:
    raise AssertionError("应拒绝过期版本")
print("当前状态与历史状态均可重建，过期写入被拒绝")

完整示例可下载，使用 Python 3.9 或以上版本运行：python3 example.py。只依赖标准库，不调用真实业务服务。

reduce_order 没有读取当前时间，也没有重新查询商品价格。同一组事件应该得到同样的状态。如果回放“增加商品”时再去查今天的价格，昨天的订单就会被算成今天的金额。对 Agent 来说也是一样：回放已经确认的模型结果，不能变成重新问一次模型，否则历史就被新的一次推理改写了。

这也解释了事件溯源与审计日志的区别。业务数据库可以先更新，另外顺手写一条操作日志，那仍然可能丢失某些变化。采用事件溯源时，需要把事件追加放在权威写入路径中，并保证成功响应、事件持久化和版本推进的关系明确。否则我们只是换了个名字记录日志，无法依靠它完整重建状态。

并发是下一道问题。两个处理器都读到版本八，然后各自决定修改订单，如果追加时不校验版本，业务决策就可能互相覆盖。通常需要“期望版本为八才允许追加”，成功者产生版本九，失败者重新读取后再判断。事件有序不等于命令天然不会竞争，存储层还得提供条件写入或相应事务保证。

我会把这种方案优先用于历史具有业务价值的场景，比如订单生命周期、额度调整、审批流程。它也方便回答“当时为什么做这个决定”，前提是记录了输入证据和规则版本。普通展示页的点击次数、可以重新采集的缓存数据，就未必值得付出相同复杂度。

需要付出的成本也很具体。事件结构一旦落盘就要考虑兼容；查询往往需要额外投影；投影异步更新时，写入成功与查询可见之间可能有延迟。包含个人信息的事件还要有数据最小化与删除策略，不能因为喜欢不可变日志，就把所有原始对话永久保存。不可变是一种建模约束，并不免除数据治理责任。

事件发布到消息队列也存在边界。数据库已经保存事件，发送消息前进程退出，下游投影就可能缺少更新。可以在同一事务里保存待发布记录，由后台可靠投递；消费者再根据事件标识去重。持久化事实与传播事实是两个环节，需要分别验证。

如果后续发现一条业务事实需要纠正，我会优先追加表达纠正的新事件，保留纠正的原因和关联编号。至于敏感数据删除，应按既定的数据治理方案处理，不能用“任何历史都不准变化”回避真实需求。

落地时我会先回答三个问题：哪些业务事实值得留下，哪些状态能够由它们推导，哪些外部影响不能在回放时再次触发。只有把这三个边界说清楚，事件溯源才会帮助我们理解 Agent 的行为，而不是再建一套更难维护的数据系统。

上一篇 · 系列目录 · 下一篇