Agent 实现方法论 From Scratch

持久层：会话与事件存储

持久层：会话与事件存储

第 07 章我们写出了最小 Agent Loop，第 08 章把工具系统工程化了，第 09 章学会了给上下文瘦身。但有个问题一直悬着：Agent 的记忆只活在进程内存里。while 循环一停、进程一退出，用户上一句说了什么、Agent 查过什么、工具调到了哪一步，全部蒸发。这一章给它装上持久层：把每一次对话变成一条条事件存进 SQLite，让 Agent 学会「忘不掉」。

本章目标

读完本章并做完配套练习后，你应该能够：

开场反例：没有持久层会怎样

我们先把「没有持久层」的痛处摊开。想象一个已经跑起来的客服 Agent 服务（这正是第 15 章之后要搭的形态），处理逻辑大致是：收到用户消息 → 拼进内存里的消息数组 → 调模型 → 把回复也推进数组 → 回给用户。一切正常，直到：

场景一：进程崩溃。 服务因为某个未捕获异常退出，或者被运维误杀。重启之后，内存里那条消息数组从头开始：用户刚说完的「帮我查一下订单数量」没了，Agent 只会傻乎乎地问「您想查什么」。用户情绪立刻崩坏：我刚刚不是说了吗？

场景二：断线重连。 用户在网页上聊到一半，网络抖了一下，前端重连。新的连接带着一个新的空会话上下文，Agent 完全不记得刚才聊到哪了。没有「恢复上一会话」这回事，因为上一会话根本没留下任何痕迹。

场景三：多开隔离。 用户同时开着两个会话：「订单客服」在查订单，「文档助手」在总结文档。两个会话的消息都堆在同一个进程内存数组里，如果没按会话切分，A 会话的上下文会混进 B 会话的请求；如果切分了，又要自己写一堆「按 sessionId 分组的 Map」逻辑，还得提防内存泄漏。

三个场景指向同一个根因：状态放在内存里，进程死，状态死。而 Agent 的核心价值恰恰是「连续性」——记得住才谈得上智能。这一章的解法一句话：把会话状态变成进程外、可重放、可隔离的数据。

存储模型之争：快照还是事件溯源

要让状态「活过进程」，第一件事是选存储模型。两条主流路线：

路线一：消息快照（state snapshot）。 存「现在的样子」：每个会话一行/一条记录，保存到目前位置的全部消息、上下文状态。用户每发一条消息，就把整条会话记录覆盖写一次。读起来最直接：SELECT ... WHERE session_id = ? 拿到的就是当前完整状态。

路线二：事件溯源（event sourcing）。 存「怎么一步步变成现在的样子」：每次发生的事（用户发了消息、Agent 回复了、调用了工具）是一条事件，事件只能追加（append-only），永不更新、永不删除。当前状态不直接存储，需要时把该会话的事件**按顺序重放（replay）**一遍，由事件序列折叠出当前状态。银行流水、账本都是这个思路：余额不存，从每一笔交易算出来。

两条路线对照：

维度消息快照事件溯源
存储内容当前状态（最终结果）导致状态的全部事件流（过程）
写入方式覆盖写（UPDATE）只追加（INSERT），永不更新/删除
读取当前状态直接查快照重放事件折叠得出（或走投影表）
历史与审计需要额外设计变更表天然完整：每条事件都在
重建任意时点状态基本不可能重放到某条事件即可
崩溃一致性覆盖写一半可能损坏追加式天然利于 WAL 恢复

对 Agent 场景，事件溯源有三个无法拒绝的理由：

  1. 审计与调试。Agent 的行为链条（用户说了什么 → 模型决定调工具 → 工具返回什么 → 模型怎么总结）天然是一条事件流。出问题时，「回放这段对话」比「看一条被覆盖了的状态」有用得多：你能看到每一步，而不是只有结局。
  2. 崩溃安全。追加（INSERT）远比覆盖（UPDATE）容易做到崩溃一致：配合数据库的日志机制（下一节讲 WAL），已提交的事件在进程崩溃后一个不少。
  3. 第 16 章的前置。断线恢复靠「事件序号 + 重连续传」：客户端说「我从第 4 条事件之后开始补」，服务端把第 5 条起的剩余事件推过去。这个设计直接建立在「事件有全局单调序号、可分段重放」之上（TP15 正是接在 TP9 的事件存储上）。

代价也要说清楚：读当前状态需要重放（或维护投影表）；事件不断累积，需要快照/归档兜底（大型系统会周期性存快照 + 只重放快照之后的事件，这是第 11 章「记忆与检索」的地盘）。本章的立场：会话层用事件溯源做事实源，再用投影表把读路径做快：两头的好处都拿。

手写实现：一个事件存储

下面用 Node ≥ 22.5 内置的 node:sqlite（零依赖）把这个存储实现出来。完整可运行代码在仓库 code/ch10-persistence/（store.mjs 是核心，mock-server.mjs 是把它包成 HTTP 服务的演示 Agent，demo.mjs 驱动完整场景）。

1. 建库建表：四张表各司其职

先看 schema 设计。核心洞察是「事件是唯一事实源，其余都是它的投影」：

关系图画成 ER 图（关系：会话拥有多条事件/消息/工具调用；消息与工具调用分别是某条事件的投影）：

erDiagram
  sessions {
    text id PK "sess_xxx"
    text title "会话标题"
    integer created_at "unix ms"
  }
  events {
    integer event_id PK "AUTOINCREMENT 全局单调"
    text session_id FK "属于哪个会话"
    text type "user_message | assistant_message | tool_call | tool_result"
    text payload "JSON 详情"
    integer created_at "unix ms"
  }
  messages {
    integer msg_id PK
    text session_id FK
    integer event_id UK "唯一：一条事件一条投影"
    text role "user | assistant"
    text content "消息正文"
    integer created_at
  }
  tool_calls {
    integer id PK
    text session_id FK
    integer event_id UK "唯一：一条事件一条投影"
    text call_id "工具调用 id，tool_result 靠它回填"
    text name "工具名"
    text arguments "参数 JSON"
    text result "可空：tool_result 事件到达时回填"
    integer created_at
  }
  sessions ||--o{ events : "has"
  sessions ||--o{ messages : "has"
  sessions ||--o{ tool_calls : "has"
  events ||--|| messages : "derives"
  events ||--|| tool_calls : "derives"

建表 SQL（含三个关键 pragma，后面逐个解释）：

PRAGMA journal_mode = WAL;      -- 预写日志：崩溃恢复 + 并发读
PRAGMA synchronous = NORMAL;    -- WAL 下"进程崩溃不丢已提交事务"的性价比档位
PRAGMA foreign_keys = ON;       -- 外键约束默认关闭，必须显式打开

CREATE TABLE IF NOT EXISTS sessions (
  id         TEXT PRIMARY KEY,          -- 'sess_xxx'
  title      TEXT NOT NULL,
  created_at INTEGER NOT NULL
);

CREATE TABLE IF NOT EXISTS events (
  event_id   INTEGER PRIMARY KEY AUTOINCREMENT,  -- 全局单调事件序号
  session_id TEXT NOT NULL REFERENCES sessions(id),
  type       TEXT NOT NULL,
  payload    TEXT NOT NULL,             -- JSON 字符串
  created_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_events_session ON events(session_id, event_id);

CREATE TABLE IF NOT EXISTS messages (
  msg_id     INTEGER PRIMARY KEY AUTOINCREMENT,
  session_id TEXT NOT NULL REFERENCES sessions(id),
  event_id   INTEGER NOT NULL UNIQUE REFERENCES events(event_id),
  role       TEXT NOT NULL,
  content    TEXT NOT NULL,
  created_at INTEGER NOT NULL
);

CREATE TABLE IF NOT EXISTS tool_calls (
  id         INTEGER PRIMARY KEY AUTOINCREMENT,
  session_id TEXT NOT NULL REFERENCES sessions(id),
  event_id   INTEGER NOT NULL UNIQUE REFERENCES events(event_id),
  call_id    TEXT NOT NULL,
  name       TEXT NOT NULL,
  arguments  TEXT NOT NULL,
  result     TEXT,                      -- 由 tool_result 事件回填
  created_at INTEGER NOT NULL
);

几个设计决策值得停下来看：

为什么 event_id 用 INTEGER PRIMARY KEY AUTOINCREMENT？ SQLite 的 INTEGER PRIMARY KEY 直接复用行号（rowid），自增保证严格单调；配合 AUTOINCREMENT 还保证「即使删过行也不复用编号」：事件溯源里编号一旦复用，审计和回放就会错乱。事件日志的全局顺序 = event_id 顺序，这是后面回放和断线重续的地基。

为什么事件带 session_id 外键？ 因为多会话隔离不靠「一个会话一个库」这种笨办法，靠的是数据模型本身：每一行都归属一个会话，查询永远先过滤会话（第 4 小节展开）。外键保证「事件的会话必须先存在」，孤儿事件根本写不进去。

为什么投影表要 UNIQUE 的 event_id？ 一条事件最多派生一条投影行，UNIQUE 约束让「事件与投影一一对应」成为数据库强制的事实：同一事件被重复投影会直接报错。

2. 追加与查询：先落库，再回复

写入是本章的心脏。原则一句话：先落库，再回复，服务收到用户消息，先把 user_message 事件持久化，再调用模型，再持久化 assistant_message，最后才把回复发回给用户。这样「回复丢了」最多是「少一次展示」，而「对话记录丢了」这种不可逆的事故被彻底排除。

appendEvent 的实现：一个事务里同时干两件事，向事件日志追加事件，并维护对应的投影行：

// type: 'user_message' | 'assistant_message' | 'tool_call' | 'tool_result'
function appendEvent(db, { sessionId, type, payload }) {
  db.exec('BEGIN');
  try {
    // ① 追加事件（事实源）
    const { lastInsertRowid } = db
      .prepare('INSERT INTO events (session_id, type, payload, created_at) VALUES (?, ?, ?, ?)')
      .run(sessionId, type, JSON.stringify(payload), Date.now());
    const eventId = Number(lastInsertRowid);

    // ② 维护投影（与事件同事务，永不漂移）
    if (type === 'user_message' || type === 'assistant_message') {
      db.prepare(
        'INSERT INTO messages (session_id, event_id, role, content, created_at) VALUES (?, ?, ?, ?, ?)',
      ).run(sessionId, eventId, type === 'user_message' ? 'user' : 'assistant', payload.content, Date.now());
    } else if (type === 'tool_call') {
      db.prepare(
        'INSERT INTO tool_calls (session_id, event_id, call_id, name, arguments, created_at) VALUES (?, ?, ?, ?, ?, ?)',
      ).run(sessionId, eventId, payload.callId, payload.name, payload.arguments, Date.now());
    } else if (type === 'tool_result') {
      // 投影允许更新：把结果回填到对应工具调用行（事件日志本身永不更新）
      db.prepare('UPDATE tool_calls SET result = ? WHERE session_id = ? AND call_id = ?')
        .run(payload.content, sessionId, payload.callId);
    }

    db.exec('COMMIT');
    return { eventId, sessionId, type, payload };
  } catch (err) {
    db.exec('ROLLBACK');
    throw err;
  }
}

注意两个「边界画线」：

读路径有两条，对应两种需求：

// 快路径：读投影表（聊天界面展示用，免回放）
function getMessages(db, sessionId) {
  return db.prepare(
    'SELECT event_id, role, content FROM messages WHERE session_id = ? ORDER BY event_id',
  ).all(sessionId);
}

// 事件溯源路径：从事件日志折叠重建会话（重发请求、恢复上下文用）
function replaySession(db, sessionId) {
  const rows = db.prepare(
    'SELECT event_id, type, payload FROM events WHERE session_id = ? ORDER BY event_id',
  ).all(sessionId);
  const messages = [];
  for (const row of rows) {
    const payload = JSON.parse(row.payload);
    if (row.type === 'user_message' || row.type === 'assistant_message') {
      messages.push({ role: row.type === 'user_message' ? 'user' : 'assistant', content: payload.content });
    } else if (row.type === 'tool_call') {
      messages.push({
        role: 'assistant',
        content: null,
        toolCalls: [{ toolCallId: payload.callId, name: payload.name, arguments: payload.arguments }],
      });
    } else if (row.type === 'tool_result') {
      messages.push({ role: 'tool', toolCallId: payload.callId, content: payload.content });
    }
  }
  return messages;
}

replaySession 的折叠规则不是随便定的：它输出的正是第 02 章定义、第 07 章 Agent Loop 一直在用的 OpenAI 风格消息形状：tool_call 事件折叠成带 toolCalls 的 assistant 消息、tool_result 事件折叠成 role: 'tool' 消息。这意味着回放出来的数组可以直接重新塞给模型，会话恢复不需要任何特殊逻辑，只需要「把事件折叠成消息，重新发一遍」。

3. 崩溃恢复与会话回放：WAL 的秘密

现在回答本章开头最尖锐的问题：进程被杀，数据怎么活下来？

答案藏在建表时那句 PRAGMA journal_mode = WAL 里。WAL（Write-Ahead Logging，预写日志）是 SQLite 的日志模式：写入不直接改主数据库文件，而是先顺序追加到 -wal 日志文件；主库文件只在特定检查点才被合并。配合 synchronous = NORMAL，语义是这样的：

说白了，SIGKILL 这类「来不及运行任何清理代码」的崩溃，正是 WAL 最典型的适用场景。崩溃恢复不是靠你的 finally 块，而是靠数据库引擎自己的日志。我们的 demo 就做了这个实验：服务进程被 SIGKILL（任何 close/清理代码都不会执行），重启后同一数据库文件打开，会话原样回来。

完整流程画成时序图（对应演示 demo 的 phase 1 → 2 → 3）：

sequenceDiagram
  autonumber
  participant U as demo（客户端）
  participant S as Agent 服务
  participant DB as SQLite（WAL）
  U->>S: POST 消息 "帮我查一下订单数量"
  S->>DB: BEGIN + INSERT 事件(user_message) + COMMIT
  S->>S: 脚本模型决定调用 query_order_count
  S->>DB: BEGIN + INSERT 事件(tool_call) + COMMIT
  S->>DB: BEGIN + INSERT 事件(tool_result) + COMMIT
  S->>DB: BEGIN + INSERT 事件(assistant_message) + COMMIT
  S-->>U: 回复 "你当前有 42 个订单。"
  Note over S: 进程被 SIGKILL —— 无任何清理代码执行
  U->>S2: 重启服务（同一个数据库文件）
  S2->>DB: 打开数据库 —— SQLite 自动回放 WAL，已提交事务恢复
  S2->>DB: SELECT ... FROM events WHERE session_id=? ORDER BY event_id
  DB-->>S2: 6 条事件
  S2-->>U: 回放出的完整会话（含工具调用链）

demo 里的实际证据（两次对话，订单客服 会话，6 条事件）：

event_idsessiontypepayload
1sess_auser_message{"content":"你好"}
2sess_aassistant_message{"content":"你好！我是订单客服助手，可以帮你查询订单数量。"}
3sess_auser_message{"content":"帮我查一下订单数量"}
4sess_atool_call{"callId":"call_1","name":"query_order_count","arguments":"{}"}
5sess_atool_result{"callId":"call_1","content":"42"}
6sess_aassistant_message{"content":"你当前有 42 个订单。"}

SIGKILL 后重启，replaySession 把 6 条事件折叠成可直接重发的消息序列：

[user]      你好
[assistant] 你好！我是订单客服助手，可以帮你查询订单数量。
[user]      帮我查一下订单数量
[assistant] tool_calls: [call_1 query_order_count args={}]
[tool]      call_1 -> 42
[assistant] 你当前有 42 个订单。

注意工具调用是两条独立事件（tool_call 与 tool_result），回放时以「assistant 的 toolCalls + tool 消息」的形式重建——这正是模型期望看到的工具调用上下文。崩溃前 Agent 想到哪一步、工具查到了什么，重放后一清二楚。

4. 多会话隔离：靠数据模型，不靠临时 Map

回到开场反例的场景三。多会话隔离在本章的实现里几乎「免费」，因为隔离是数据模型的属性，不是运行时的临时分组：

  1. 每一行都归属一个 session_id（外键约束），查询时 WHERE session_id = ? 永远先过滤会话；
  2. 组合索引 (session_id, event_id) 让「取某会话全部事件」走索引，而不是全表扫描；
  3. 会话之间没有任何共享的行：新会话的写入只是追加自己的事件，物理上无法污染别的会话。

demo 的 phase 4 验证了这一点：新建 文档助手 会话并对话两次后，8 条事件分布为 sess_a 6 条 + sess_b 2 条，各自查询互不可见；sess_a 在 sess_b 建立前后的查询结果完全一致（这一点也会在本章练习的判题里用纯逻辑验证）。

顺带注意一个细节：getMessages(sess_a) 返回的消息 event_id 是 [1, 2, 3, 6]（跳过了 4、5）。因为 messages 是投影表，只存 user/assistant 消息，工具事件不在里面（它们在 tool_calls 投影）。这不是 bug，是「投影是切片」的正常表现：展示用投影、恢复用事件日志，两条读路径各司其职。如果你发现投影和事件对不上，那一定是写路径没走同一个事务，这正是第 2 小节强调的。

5. 为什么练习不直接连 SQLite？

你可能已经注意到：本章练习（浏览器里的判题）没有让你写任何 SQL，而是把「事件日志」抽象成一个纯内存数据结构 + 纯函数（追加事件、按会话过滤、回放重建、单调 eventId）。这是刻意为之，两个原因：

  1. 判题环境限制。练习的判题跑在浏览器的沙箱里：那里没有 node:sqlite，没有 node: 系列的任何 API，也不该让学习者代码直接碰文件系统或数据库（安全边界）。SQLite 的真实 SQL 在演示代码（code/ch10-persistence/）里完整呈现并可运行。
  2. 本章的核心是「事件日志的语义」，不是 SQL 语法。INSERT ... WHERE session_id 的写法查一下文档就会；难的是想清楚「追加式、不可变、按会话隔离、可重放」这些性质怎么设计、怎么测试。这些性质与数据库引擎无关，用数组和纯函数就能完整覆盖，判题只断言行为（追加后旧事件不变、跨会话不串、回放形状正确），不锁定写法。

所以练习与 demo 的关系是：练习测「语义」，demo 演「落地」。把练习里练会的纯函数思维，套进 demo 的 appendEvent / replaySession，就是一套完整的持久层。

常见坑与失败模式

坑一：先回复，后落库。 如果先给用户回消息、再异步写库，崩溃窗口期里「用户看到了回复，但会话记录丢了」，下次恢复时对话出现断层，用户还得重复一遍。先落库、再回复（write-ahead 思路在应用层的体现）：回复丢了最多重发，记录丢了不可逆。

坑二：事件日志允许 UPDATE / DELETE。 一旦「改一条事件」成了习惯（比如手滑写错 payload 想直接改），审计与回放的可靠性就崩了。你无法确定重放出的历史是不是被篡改过的。纪律：事实只能追加；想修正，追加一条新事件（甚至可以带 supersedes 字段指向前一条）。

坑三：投影与事件不在同一事务。 先写事件、再单独写投影，进程在两步之间崩溃，投影缺行，读路径与回放结果不一致。凡是派生数据，必须与源数据同事务提交。

坑四：忘了 PRAGMA foreign_keys = ON。 SQLite 的外键约束默认关闭：不开这个 pragma，REFERENCES sessions(id) 形同虚设，孤儿事件能写进去，多会话隔离的「数据模型保证」直接失效。每个连接都要显式打开。

坑五：低估 WAL 的语义边界。 WAL + synchronous=NORMAL 保的是进程崩溃；断电场景要用 FULL。另外 WAL 会产生 -wal / -shm 两个伴生文件。备份/拷贝数据库时只拷主文件，会把没检查点的已提交数据丢在外面。

坑六：快照思维写事件表。 习惯 ORM 的人容易把 events 当成「状态表」来更新（比如「把这条消息标成已读」直接 UPDATE events）。记住表的分工：events 是账本，只进不出；「已读」这种可变状态，属于投影或单独的元数据表。

小结

下一章（ch11）做记忆与检索：持久层负责「不丢」，记忆负责「记得住、找得到」，届时你会看到，本章的会话事件存储正是长期记忆的底层。先把这一章练熟：现在去浏览器里完成练习，把 createEventLog、appendEvent、getEventsForSession、listSessions、replaySession 亲手补全，把「追加式、隔离、回放」这些性质变成肌肉记忆。

延伸阅读

完成阅读，去做练习 →