首页 > 人工智能 > 人工智能 > 拆解 dsh:Session 的事件溯源与状态重建
const { request, preparedCall } = await this.buildRequest(
turn,
step,
assembly.tools,
system,
this.session.deriveMessages(),
signal,
)
this.session.deriveMessages()。每次发起模型请求前,Session 都会从事件日志中取出当前需要进入模型上下文的内容,生成这一次请求携带的
messages。本文基于@deepseek-ai/dsh0.1.1-rc.2,对应仓库标签dsh-v0.1.1-rc.2。dsh 当前仍处于 developer preview,会话格式为SESSION_FORMAT_VERSION = 0,暂不提供格式兼容承诺。
messages 和持久化日志分开维护,Agent 运行过程中就需要一直同步两份状态。工具调用产生新记录,两边都要更新;上下文发生压缩,两边也要一起调整;到了会话恢复时,还得确认它们对应的是同一段执行历史。SessionEvent Log ↓ Surface ↓ deriveMessages() ↓ Message[]
SessionEvent 记录进日志。deriveMessages() 再据此生成对应的
Message[]。恢复 Session 时,系统也会重新加载这些事件,并沿用同一套规则还原模型可见的消息历史。persistence-catalog.md 中,共定义了 48 种可持久化事件。其中 13 种由
dsh-session 定义,其他事件由各个插件扩展到
SessionEventMap 中。type SurfaceEventType = | 'user/message' | 'assistant/message' | 'tool/result'
turn/start、
step/end、
tool/call、
assistant/chunk,以及重试、审批和压缩过程中产生的事件,都会保留在日志里,但不会进入模型消息历史。append<T extends SessionEventType>( type: T, data: SessionEventMap[T], ...opts: T extends SurfaceEventType ? [opts: SurfaceIntent] : [] ): SessionEvent<T>
append() 时必须传入
SurfaceIntent,其他事件则不能传入这组参数。也就是说,哪些事件可以进入模型上下文,在
Session.append() 的类型定义里就已经被限定下来。seq = log.length
seq 表示事件在日志中的连续位置。负责 Session 日志实际存储的 Persistence Backend,可以自行决定磁盘编码、批处理方式和存储结构,但重新
load() Session 时,仍要按原来的顺序完整还原这条事件日志。这也是
assistant/chunk 要保留在日志中的原因。assistant/chunk assistant/chunk assistant/chunk ↓ assistant/message
assistant/message。Message A Message B Tool Result C Message D Message E
Summary Message D Message E
SessionSurface:interface SessionSurface {
readonly nodes: readonly number[]
readonly replaceGeneration: number
}
surfaceOp 声明自己如何进入当前模型历史:type SurfaceOp =
| 'append'
| { op: 'replace'; start: number; end: number }
append 把新节点加入尾部,
replace 用当前事件替换 Surface 中的一段节点。被替换的事件依旧留在 SessionEvent Log 中。Event Log ────────────────────────────── seq 10 user/message seq 11 assistant/message seq 12 tool/result seq 13 assistant/message seq 14 user/message(summary) ────────────────────────────── Surface before 10 → 11 → 12 → 13 Surface after replace 14
compaction/start compaction/summary compaction/end compaction/prune
compaction/* 事件记录,真正进入模型上下文的摘要仍是一条
user/message,通过
replace 接管原来的 Surface 区间。compaction/prune,记录被替换的 Surface 节点和 token 数量,再同步追加替换用的新
tool/result:compaction/prune ↓ replacement tool/result ↓ Surface replace
compaction/prune 调整当前 Surface 的计价状态。
deriveMessages() 最终按照 Surface 节点顺序,将三类事件转换成模型消息:user/message → user Message assistant/message → assistant Message tool/result → 带 tool-result block 的 user Message
assistant/message 会被跳过。生成
Message[] 的过程也使用了增量缓存:普通追加时只处理新节点,发生 Replace 后再重新构建对应的消息历史。Message[] 时,都不需要重新遍历和复制整段会话。Message[],还不足以还原一次完整的模型请求。一次大模型调用会带上模型配置、System Prompt 和工具定义等信息:provider model system prompt tool schemas reasoning effort sampling config ...
request/header来保存完整的
EpochHeader:interface EpochHeader {
config: LlmCallConfig
adapterDefaults?: LlmCallConfigAdapterDefaults
system?: string
tools?: ToolSchema[]
}
initial;新的 Agent Loop 实例接手现有 Session 时会写入
resume;后续如果模型配置、System Prompt 或工具定义发生变化,则写入
change。没有变化的 Step 继续沿用上一份 Header。Session Event Log │ ├── Surface │ ↓ │ Message[] │ └── request/header ↓ config / system / tools │ ↓ Conversation Request
adapterDefaults 用来标记哪些配置来自 Adapter 的默认值。agent/request 前,Agent Loop 会移除由上一轮 Adapter 补入的默认参数,再由当前路由重新确定这些默认值。调用方明确设置的参数则会继续保留。request/context 中:{
provider,
model,
contextWindow
}
@deepseek-ai/dsh-agent-loop/invariant 插件。它会从 Session Log 中重新生成消息边界和 Request Header,再与 Agent Loop 实际准备发送的请求进行核对。真实 LLM Request = Session Log 可重建出的 Request
Persistence Backend 负责存到哪里 Checkpoint Policy 负责什么时候必须落盘
session/flush 排空待写入事件。
dsh-session-checkpoint-policy 则会在三个关键位置先确认相关事件完成持久化:模型请求派发前 模型发起的工具调用执行前 下一次 agent/pre-step 前
llm/stream。相关请求事件完成持久化之前,下游 stream 不会启动。工具执行这边,checkpoint 会放在执行前策略和安全检查之后、工具真正开始执行之前。也就是说,工具通过前置检查后,还要先确认相关事件完成持久化,才能进入实际执行。tool/call ↓ policy / guard ↓ checkpoint ↓ tool body ↓ tool/result
ABORTED_BEFORE_DISPATCH,工具也不会进入实际执行。记录调用 ↓ 确认耐久 ↓ 允许外部执行
assistant 请求工具 │ ├── 没有持久化的 tool/call │ ↓ │ TOOL_NOT_STARTED │ └── 有 tool/call 没有可靠结果 ↓ TOOL_OUTCOME_UNKNOWN
TOOL_NOT_STARTED 表示模型提出了工具请求,但日志中没有对应的持久化
tool/call。恢复时,系统可以判断这次工具调用还没有进入执行。如果任务仍然需要这次操作,Agent 可以重新发起调用。TOOL_OUTCOME_UNKNOWN 表示日志中存在
tool/call,但没有可靠的完成结果。工具可能还没有执行,也可能只执行了一部分,还可能已经在外部系统中完成,只是执行结果还没有写回 Session。已记录调用 ≠ 已知执行结果
turn/start ↓ step/start ↓ ... ↓ step/end ↓ turn/end
turn/start ↓ step/start ↓ ... ↓ <EOF>
SessionPersistence.load(id) 会先等待内存中的最新事件完成持久化。只有当前 Turn 和 Step 都完整结束后,系统才会返回这份 Session。如果当前还有一个 Turn 正在执行,
load() 会拒绝加载,也不会额外补上一条
interrupted 来结束这个 Turn。turn/end。因此,这一轮任务可以继续沿用原来的执行状态。turn/end,并将结束原因标记为
interrupted:turn/end {
reason: {
kind: 'interrupted'
}
}
interrupted 只会在会话恢复时生成,正常运行的 Agent Loop 不会写入这个状态。Repair 只负责处理 Turn、Step 和 Tool 这些核心执行记录。像
compaction/start、
compaction/end 这样的插件事件,则由对应插件自己判断和处理。session/end-seed:seed events seed events session/end-seed ─────────────── live events live events
assistant/chunk 也可以继续提供细粒度 UI 回放。fork(source, boundary?, childSessionId?)
Session A 0 ─ 1 ─ 2 ─ 3 ─ 4 ─ 5 ↑ boundary │ └──── Fork ↓ Session B 0 ─ 1 ─ 2 ─ 3
(state, event) => nextState。Todo、运行状态等信息都可以按照 SessionEvent 的顺序更新。state。Registry 再通过
Object.is 判断状态有没有变化:same reference ↓ no state change ↓ no downstream work
┌─ Message History ├─ Replay ├─ Persistence SessionEvent Log ────┼─ Crash Recovery ├─ Fork └─ Projection
assistant/chunk 这样的事件也会一直保留在日志中。任务运行时间越长,从头读取并重新处理全部事件的成本也会越高。session-query 提供统一的 Session 查询接口,可以同时查询当前正在运行的 Session 和已经持久化的 Session。全文检索由具体的存储实现负责,
session-query-sqlite 提供了一套 SQLite 实现。session-projection-cache 则会缓存 Projection 处理后的状态,并用
seq 记录当前处理到了日志中的哪个位置。再次读取时,只需要从这个位置之后继续处理新增事件。SESSION_FORMAT_VERSION = 0,遇到未知且没有标记
ignorable: true 的事件时,Session 会拒绝恢复。this.session.deriveMessages()
来自 “ ITPUB博客 ” ,链接:https://blog.itpub.net/70050060/viewspace-3183911/ ,如需转载,请注明出处,否则将追究法律责任。