DeepSeek Harness 系列(05):Session 与记忆——对话历史是怎么活下来的

文章来源声明: 原文作者:冬奇Lab; 来源站点:掘金; 原文链接:https://juejin.cn/post/7684657593500532790; 本文基于上述来源整理/加工,觅优补充点评,仅供技术学习交流。版权归原作者所有。
觅优短评

把日志当唯一真源、历史靠派生,是 Agent 状态管理的干净解法。适合要做崩溃恢复、对话分叉与多方案探索的 Agent 工程团队参考。

先问一个实际问题 --------

你的 Agent 跑了一半,进程崩了。重启之后,之前的对话历史还在吗?

或者:你想从某一轮的某个节点"回到过去",试一条不同的路——dsh 支持吗?

这两个问题都指向同一个设计:Session


Session 是什么

很多人以为 Session 就是"把消息列表存起来"。dsh 的 Session 不是这样设计的。

dsh Session 是一份 仅追加(append-only)的类型化事件日志

┌─────────────────────────────────────────────────────┐
│  Session Log (seq = 日志位置,从 <span>0</span> 开始单调递增)      │
│                                                     │
│  seq=<span>0</span>:  turn/start     { turn: <span>1</span> }                 │
│  seq=<span>1</span>:  user/message   { role: <span>'user'</span>, ... }       │
│  seq=<span>2</span>:  system/message { message: {...} }          │
│  seq=<span>3</span>:  request/header { <span>header</span>: {...} }           │
│  seq=<span>4</span>:  assistant/message { message: {...} }       │
│  seq=<span>5</span>:  tool/call      { name: <span>'read_file'</span>, ... }  │
│  seq=<span>6</span>:  tool/result    { message: {...} }          │
│  seq=<span>7</span>:  assistant/message { message: {...} }       │
│  seq=<span>8</span>:  turn/end       { reason: { kind: <span>'completed'</span> } }│
│                                                     │
│  只能追加,永远不能修改或删除已有条目               │
└─────────────────────────────────────────────────────┘

模型的"消息历史"不是单独存储的——它从这份日志派生出来。

为什么要多此一举?因为仅追加日志有三个直接收益:

  1. 崩溃安全:进程中断时,已写入的事件不会丢失,也不会产生部分写入的脏状态
  2. 可回放:同一份日志重放,得到完全相同的派生历史——无论在哪台机器上
  3. 可审计:失败的尝试(assistant/attempt)永久保留在日志里,只是不会进入模型历史

这个思路在工程领域有个专门的名字叫"事件溯源"(Event Sourcing)。如果你用过 Git,就已经在用类似的概念了:每次提交只是新增一条记录,历史只增不改。


事件词汇:Session 里到底有什么

Session 日志由类型化的 SessionEvent 组成,每条事件有固定的结构:

<span>// packages/core/session/src/types.ts</span>

<span>type</span> <span>SessionEvent</span><T <span>extends</span> <span>SessionEventType</span> = <span>SessionEventType</span>> = {
  <span>// 类型(如 'turn/start'、'assistant/message')</span>
  <span>type</span>: T
  <span>// 在本 Session 里的单调递增序号(等于 log.length,连续无间隙)</span>
  <span>seq</span>: <span>SessionSeq</span>
  <span>// 追加时的 Unix 时间戳(毫秒)</span>
  <span>time</span>: <span>number</span>
  <span>// 事件的具体数据(必须是可 JSON 序列化的)</span>
  <span>data</span>: <span>SessionEventMap</span>[T]
  <span>// 可选:这条事件未知时读取方可以跳过(而不是拒绝整个 Session)</span>
  ignorable?: <span>true</span>
}

核心事件类型一览:

事件类型含义
`turn/start`一个 Turn 开始
`turn/end`一个 Turn 结束,携带结束原因
`step/start`一个 Step 开始
`step/end`一个 Step 结束
`user/message`用户发送的消息(或注入的上下文)
`system/message`渲染后的系统提示词
`assistant/message`模型成功输出(**进入派生历史**)
`assistant/attempt`模型尝试但失败(**不进入派生历史**)
`tool/call`模型请求调用一个工具
`tool/result`工具执行完毕的结果
`request/header`本次请求的配置快照(模型、token 上限等)

Surface:派生历史的来源

Session 日志里所有事件里,只有 4 种事件类型会产生模型消息,合称 SurfaceEventType

  • system/message
  • user/message
  • assistant/message
  • tool/result

这 4 种事件构成了"Surface"(surface,可以理解为"浮出水面"的那部分)。其余事件(turn/startstep/endassistant/attempt 等)是日志层面的结构信息,不产生任何 LLM 消息。

每个 Surface 事件都会携带一个 surfaceOp 标记,说明它是怎么加入这个有序队列的:

<span>type</span> <span>SurfaceOp</span> =
  <span>// 追加到尾部——所有正常消息走这条路</span>
  | <span>'append'</span>
  <span>// 替换 startSeq ~ endSeq 之间的节点(压缩对话历史时用)</span>
  | { <span>op</span>: <span>'replace'</span>; <span>startSeq</span>: <span>SessionSeq</span>; <span>endSeq</span>: <span>SessionSeq</span> }

平时用的都是 'append'replace 主要用于对话历史压缩(compaction)——当对话太长时,把一段对话摘要成一条消息替换掉原来的,节省 token。


deriveMessages():历史是怎么派生的

Session.deriveMessages() 就是那个"从日志重建消息列表"的核心方法:

<span>// packages/core/session/src/index.ts(简化)</span>

<span>class</span> <span>Session</span> {
  <span>/**
   * 从有序的 Surface 事件派生 LLM 消息历史。
   *
   * 带缓存:每个 surface 节点只投影一次;surface 被替换时重建。
   * 每次调用返回新数组,但数组内的 Message 对象是共享的深冻结引用。
   */</span>
  <span>deriveMessages</span>(): <span>Message</span>[] {
    <span>// 遍历 surface.nodes(有序的 surface 事件 seq 列表)</span>
    <span>// 对每个节点调用 deriveEventMessage(event)</span>
    <span>// 返回不为 null 的结果</span>
  }
}

投影规则(deriveEventMessage)很简单:

<span>// packages/core/session/src/surface.ts</span>

<span>export</span> <span>function</span> <span>deriveEventMessage</span>(<span>event: SessionEvent</span>): <span>Message</span> | <span>null</span> {
  <span>switch</span> (event.<span>type</span>) {
    <span>case</span> <span>'user/message'</span>:
      <span>// 原样投影为 user 角色消息</span>
      <span>return</span> event.<span>data</span>

    <span>case</span> <span>'system/message'</span>:
    <span>case</span> <span>'assistant/message'</span>:
      <span>// 内容为空时返回 null(空内容不应该出现在 transcript 里)</span>
      <span>if</span> (event.<span>data</span>.<span>message</span>.<span>content</span>.<span>length</span> === <span>0</span>) <span>return</span> <span>null</span>
      <span>return</span> event.<span>data</span>.<span>message</span>

    <span>case</span> <span>'tool/result'</span>:
      <span>// 投影为带 tool-result block 的 user 消息</span>
      <span>return</span> event.<span>data</span>.<span>message</span>

    <span>default</span>:
      <span>// turn/step 边界、assistant/attempt 等——不产生消息</span>
      <span>return</span> <span>null</span>
  }
}

两个细节值得注意:

内容为空的 assistant/message:这种情况出现在模型因 max-tokens 被截断但还没来得及输出任何内容的步骤。这条事件还是会写入日志(用来记录 token 用量和 stream 信息),但不会产生消息——否则模型下次会看到一条空白的 assistant 轮次,造成混乱。

assistant/attempt 完全不参与投影:无论发生了什么(网络错误、上下文超限、用户取消),失败的尝试只写日志,不进历史。模型下次请求时看到的是干净的状态,就好像那次失败的尝试从来没有发生过。


assistant/message vs assistant/attempt:更深一层

上一篇文章(Agent Loop)介绍了这两个事件的区别,这里从 Session 层面再看一遍:

`assistant/message``assistant/attempt`
何时写入模型成功完成输出网络错误 / 上下文超限 / 取消 / 流错误
Surface 事件✅ 有 `surfaceOp`,进入派生历史❌ 无 `surfaceOp`,仅日志
模型下次请求看到✅ 会看到❌ 看不到
保留在日志里✅(用于审计和 usage 统计)

为什么 assistant/attempt 也要写入日志?

因为失败也算消耗了 token。即使模型没有返回有效内容,提供方可能已经按输入 token 收费了。把 attempt 保存在日志里,可以在事后统计真实的 token 用量(包括失败的那些请求),而不是只看成功的。


Session Header:格式版本与元数据

每个 Session 除了事件日志,还有一个存储在日志之外的 SessionHeader

<span>// packages/core/session/src/types.ts(简化)</span>

<span>interface</span> <span>SessionHeader</span> {
  <span>// 格式版本,目前是 3</span>
  <span>readonly</span> <span>version</span>: <span>typeof</span> <span>SESSION_FORMAT_VERSION</span>  <span>// = 3</span>
  <span>// Session 的唯一 ID</span>
  <span>readonly</span> <span>id</span>: <span>SessionId</span>
  <span>// Session 创建时间(Unix 毫秒)</span>
  <span>readonly</span> <span>createdAt</span>: <span>number</span>
  <span>// Session 所在的工作目录</span>
  <span>readonly</span> cwd?: <span>string</span>
  <span>// 这个 Session 是从哪个 Session fork 出来的</span>
  <span>readonly</span> parentSession?: <span>SessionId</span>
  <span>// 是否包含继承的 fork 前缀</span>
  <span>readonly</span> <span>isSeeded</span>: <span>boolean</span>
  <span>// 如果是子 Agent,记录委托深度(防止无限递归)</span>
  <span>readonly</span> delegationDepth?: <span>number</span>
}

<span>// 当前最新格式版本</span>
<span>export</span> <span>const</span> <span>SESSION_FORMAT_VERSION</span> = <span>3</span>

格式版本的用途是迁移。dsh 的 Session 日志会存在磁盘上,可能被不同版本的 dsh 读取。SESSION_FORMAT_VERSION 保证了这套机制:

  • 旧格式日志被读取时,迁移链把它升级到当前版本
  • 如果读取到完全未知的版本,dsh 拒绝处理,不会静默读坏

这和 SQLite 的 PRAGMA user_version 或 Git 对象格式的考量是一回事。


Fork:从任意历史节点分叉

这是 dsh Session 最有趣的功能之一。

假设你有一个对话,已经进行了 10 轮。到第 7 轮时,模型选了一条你觉得不好的路。你想回到第 7 轮结束时的状态,从那里重新开始,试一个不同的方法。

这在 dsh 里叫 Session Fork

<span>// packages/core/session/src/index.ts(SessionStore 的方法)</span>

<span>/**
 * 从一个活跃 Session 的稳定前缀创建子 Session。
 *
 * <span>@param</span> <span>source</span> - 活跃的源 Session 对象或 ID
 * <span>@param</span> <span>boundary</span> - 可选,切分点 SessionSeq(含);
 *                   默认截取到当前最后一个事件。
 *                   要求截取点结束时没有开放的 Turn。
 * <span>@param</span> <span>childSessionId</span> - 可选的子 Session ID
 */</span>
<span>fork</span>(
  <span>source</span>: <span>SessionForkSource</span>,
  boundary?: <span>SessionSeq</span>,
  childSessionId?: <span>SessionId</span>,
): <span>Session</span>

Fork 的行为:

  1. 从源 Session 复制 boundary 之前(含)的所有事件,作为子 Session 的"继承前缀"
  2. 子 Session 持有继承事件的计数(inheritedEventCount
  3. 子 Session 的后续写入不会影响源 Session
  4. session/end-seed 事件标记了继承前缀的边界
源 Session:<span>[turn1]</span><span>[turn2]</span><span>[turn3]</span><span>[turn4]</span>...<span>[turn10]</span>
                              ↑ boundary
Fork 后:
  源 Session:继续...
  子 Session:<span>[turn1]</span><span>[turn2]</span><span>[turn3]</span> ← 继承,不可变
                                    <span>[新turn4]</span>... ← 子 Session 独有

一个实际用途:让 Agent 探索不同的解题方案,各走一条 fork,最后比较结果——而不用开多个独立对话,浪费重复的上下文。


崩溃恢复:interrupted 结束原因

如果进程在 Turn 执行中途崩溃,Session 日志里会有一个没有对应 turn/endturn/start

dsh 在下次加载这个 Session 时,会检测到这种"开放的 Turn",并自动合成一条:

<span>// turn/end 的结束原因类型之一</span>
{ <span>kind</span>: <span>'interrupted'</span> }

这是唯一一个不由 loop 主动发出turn/end 原因——它只在崩溃恢复路径里由持久化层合成。

合成这条 turn/end 后,Session 处于一致的、可继续的状态,崩溃前的所有已写事件完整保留,之后可以正常 resume。


实战:读取历史 Session,跨会话传递记忆

把以上概念落地成代码。下面的插件在每次 Turn 结束时,把上一个 Session 的摘要注入当前 Session:

<span>// 跨 Session 记忆插件(伪代码,说明核心逻辑)</span>
<span>export</span> <span>const</span> name = <span>'cross-session-memory'</span>
<span>export</span> <span>const</span> inject = [<span>'sessions'</span>]

<span>export</span> <span>function</span> <span>apply</span>(<span>ctx: Context</span>): <span>void</span> {
  ctx.<span>on</span>(<span>'agent/turn-stopping'</span>, <span>async</span> (payload) => {
    <span>const</span> { agent } = payload

    <span>// 获取当前 Session</span>
    <span>const</span> currentSession = agent.<span>session</span>

    <span>// 找到上一个 Session(实际中可以从持久化存储里查询)</span>
    <span>const</span> prevSessionId = <span>await</span> <span>getPreviousSessionId</span>()
    <span>if</span> (!prevSessionId) <span>return</span>

    <span>// 加载上一个 Session 的事件</span>
    <span>const</span> prevSession = <span>await</span> <span>loadSession</span>(prevSessionId)

    <span>// 从事件日志派生消息历史</span>
    <span>const</span> prevMessages = prevSession.<span>deriveMessages</span>()

    <span>// 提取最后几条对话作为"记忆摘要"</span>
    <span>const</span> summary = <span>buildSummary</span>(prevMessages.<span>slice</span>(-<span>10</span>))

    <span>// 把摘要注入当前 Session 的下一步</span>
    agent.<span>inject</span>({
      <span>type</span>: <span>'user'</span>,
      <span>content</span>: [{ <span>type</span>: <span>'text'</span>, <span>text</span>: <span>`[记忆] <span>${summary}</span>`</span> }],
    })
  })
}

关键点:deriveMessages() 是纯函数式的——给定同一份事件日志,结果总是相同的。你可以随时从任意一个 Session 重建历史,不需要额外的"历史存储层"。


Session 的完整生命周期

ctx.sessions.<span>create</span>()
  ↓
session/created(通知监听器)
  ↓
[agent loop 开始运行 Turn]
  ↓
  事件追加到日志:turn/start → ... → turn/end
  ↓
  session/flush(持久化层把缓冲事件写入磁盘)
  ↓
[Agent idle 或者任务完成]
  ↓
session/disposed(Session 从 store 里移除)

持久化是插件负责的(session/flush 监听器),核心的 Session 类本身只管内存中的事件日志和派生逻辑。这样设计的好处是:持久化后端可以随时替换(JSONL 文件、SQLite、远端 API),Session 的核心语义保持不变。


设计总结

dsh Session 的整个设计可以用一句话概括:

日志是唯一真源,历史是从日志派生的。

设计决策原因
仅追加日志崩溃安全 + 可回放 + 审计完整
`assistant/attempt` 不进历史失败不污染模型看到的上下文
`deriveMessages()` 是缓存的纯函数多处读取不产生额外开销;重放结果一致
格式版本号支持日志迁移,而不是静默读坏旧格式
Fork API从任意稳定节点分叉,不破坏源 Session
持久化由插件负责核心与存储后端解耦

系列下一篇

下一篇 Profile 与 Bundle:dsh 的配置装配系统 讲 dsh 是怎么从配置文件组装出一个完整的 Agent 运行时的——Bundle 是什么、Profile 是什么,以及你怎么用这套机制为不同任务定制不同的 Agent 组合。


PrimeSkills 可以找到已在真实企业场景验证过的 AI Agent 技能和工作流,不是演示级的,是用在实际项目里的。

更多内容见我的个人主页