Event Sourcing AI

以持久事件事实重建 AI 任务状态，区分调试日志、投影回放和外部副作用，明确顺序、版本与存储边界。

#concept #evergreen #ai #architecture

[!info] related notes

所属 MOC: LLM Streaming MOC

组合: Structured Streaming, Multi-channel Stream

综合: Streaming 高级协议模式

基础: Append Event Stream

Event Sourcing AI

一句话定义

AI 事件溯源将约定的任务状态变化保存为持久事件事实，并通过确定的 reducer 重建状态。它是状态持久化设计，不是流式协议的必经升级，也不要求保存所有模型输出。

核心机制

核心思想

传统方式只保存最终结果：

{ "role": "assistant", "content": "你好世界" }

下面是教学事件序列；实际只记录接口可观察且重建目标状态所必需的事实，不推断模型内部思考：

UserMessageCreated
GenerationStarted
TextDeltaRecorded("你")
TextDeltaRecorded("好")
TextDeltaRecorded("世界")
GenerationEnded
MessageFinalized

只有日志覆盖所需事实、保留可用版本且 reducer 兼容时，才能重建约定范围的状态；普通调试日志不一定满足这些条件。

示例事件类型

UserMessageCreated          ← 用户发送消息
GenerationStarted          ← 请求开始生成
TextDeltaRecorded              ← 文本增量被记录（可按块合并）
ChunkGenerated              ← 语义 chunk 生成
ToolCallRequested           ← 工具调用请求
ToolCallExecuted            ← 工具调用执行完成
ToolResultReturned          ← 工具结果返回
StructuredDataExtracted     ← 结构化信息提取
CitationAdded               ← 引用来源添加
ErrorOccurred               ← 错误发生
MessageFinalized            ← 消息生成完成

事件存储结构

interface AIEvent {
  eventId: string           // 全局唯一 ID
  sessionId: string         // 会话 ID
  messageId?: string        // 消息事件的目标；任务事件可不属于消息
  type: string              // 事件类型
  data: object              // 事件数据
  timestamp: number         // 时间戳
  sequenceNumber: number    // 会话内序号；写入端须保证唯一递增，时间戳不能代替顺序
}

状态重建

重建指定序号之前的状态。以下为伪代码，假设输入是同一 session 内已验证、去重且无缺口的事件；applyEvent 必须无外部副作用，不能在回放时再次调用模型或执行工具：

function rebuildState(events: AIEvent[], upToSequence: number): SessionState {
  let state: SessionState = { messages: [], toolCalls: [], status: "idle" }

  for (const event of events.filter(e => e.sequenceNumber <= upToSequence).sort((a, b) => a.sequenceNumber - b.sequenceNumber)) {
    state = applyEvent(state, event)
  }

  return state
}

最小场景

调试一个 AI 对话出错的场景：

// 事件内容示意，省略 ID、序号等字段，不可直接当 AIEvent[] 运行。
// 1. 保存重建所需事件日志
const events = [
  { type: "UserMessageCreated", data: { content: "我头疼" } },
  { type: "TextDeltaRecorded", data: { text: "你" } },
  { type: "TextDeltaRecorded", data: { text: "好" } },
  { type: "ToolCallRequested", data: { name: "symptom_extract", args: {...} } },
  { type: "ToolResultReturned", data: { result: {...} } },
  { type: "ErrorOccurred", data: { error: "JSON parse error" } },
  // ...
]

// 2. 回放到出错前的状态
const stateBeforeError = rebuildState(events, errorEvent.sequenceNumber - 1)

// 3. 分析问题
console.log("出错前的工具调用:", stateBeforeError.toolCalls)

适用场景

场景	说明
AI 对话调试	回放完整过程，定位问题
审计记录	保存已观察的请求、决策和结果，不能证明模型内部原因
A/B 测试	对比不同 prompt 的完整生成过程
质量评估	按实际保存的文本和结果评估；文本片段不等于 tokenizer token
用户体验研究	分析用户与 AI 的交互模式
状态恢复	断线重连后从事件日志恢复状态

优势

可追踪已记录的状态变化；未记录的外部状态和内部原因无法由日志推断
可回放保留范围内、版本兼容的事件序列，必要时从快照继续
方便调试：出错时可以精确回放到出错前的状态
可支持审计，但不自动满足完整性、访问控制和保留期限等要求
支持并行对比：同一份输入，不同 prompt 的完整过程可以并行对比

局限

存储开销取决于事件粒度、payload 与保留期限；无需每个 token 单独记录
实现复杂：需要事件存储、序列号管理、状态重建逻辑
延迟考虑：如果每个事件都持久化，会增加端到端延迟
不适合所有场景：简单聊天不需要这么重的方案

与 Kafka 日志的类比

Kafka 是可选的日志基础设施，不等于事件溯源；以下是设计类比，不是固定映射。顺序通常限定于分区，partition 不必对应 Message。

概念	Kafka	AI Event Sourcing
事件	Kafka Record	AIEvent
主题	Topic	Session / Conversation
顺序范围	Partition 内	自定义的 session/run 事件序列
消费者	Consumer	Frontend / Analytics / Debug Tool
偏移量	Offset	sequenceNumber

边界与易混淆点

Event Sourcing ≠ 必须用 Kafka：可用数据库或持久文件等；仅内存日志无法在进程丢失后恢复，Kafka 本身也不能代替领域事件契约。
不需要记录每个 token：实际系统通常记录 chunk 级别事件，token 级别太细。
Event Sourcing 和 CQRS 的关系：Event Sourcing 以事件为事实来源；CQRS 分离读写模型，可与其组合，也可独立使用。读模型可以是 UI、查询索引或统计投影。
和 Structured Streaming 的关系：Structured Streaming 是”事件格式”，Event Sourcing 是”事件存储和回放”。两者互补。