会话记忆自迭代:v3 抽取、用户记忆与 Agent 经验

30 秒导读: 一次对话结束(session.commit())后,OpenViking 不占用户等待,在后台异步跑一遍"抽取"。抽取分两条线:一条把对话里关于"用户是谁、偏好什么、发生了什么"的长期记忆合并回目录树;另一条由一种叫 cases 的记忆触发,把"这次 Agent 是怎么把活干成的"蒸馏成可复用经验写回。下次同一个用户来,检索(见 03)就能捞到这些回写,于是 Agent「越用越聪明」。本章只讲这个 会话 → 记忆回写 的闭环;底层怎么落盘看 04,怎么检索看 03。

1. 这是什么(零基础也能懂)

一句话定义: 会话记忆自迭代 = 每次对话结束后,系统自动读一遍这段对话,把值得长期记住的东西抽出来,写回 01 讲的那棵 viking:// 目录树。

解决什么问题: 大模型本身没有跨会话的长期记忆——这次告诉它"我叫张三、常坐靠窗",下次开新会话它就忘了。OpenViking 的做法是把这些事实沉淀成磁盘上的记忆文件;下次检索时再喂回去。

两类要沉淀的东西,必须分清(这是全章主线):

线	记的是谁的事	例子	面向
用户记忆(user)	关于用户的长期事实	"张三,素食,常坐靠窗"	让 Agent 更懂这个用户
Agent 经验(agent)	关于Agent 自己怎么把活干成的方法	"退改签任务的标准工具序列"	让 Agent 下次干得更好

用起来什么样: 用户只调一个 commit(),立刻拿到 task_id 返回,不用等抽取跑完。

result = session.commit()
# {"status": "accepted", "task_id": "...", "archived": True,
#  "archive_uri": "viking://user/{uid}/sessions/.../history/archive_001"}
# 抽取在后台异步跑;可用 task_id 轮询进度(依据:docs/en/concepts/08-session.md:66-80)

一句话直觉(这是类比,不是定义): 像人下班后写"复盘日记"——一部分记"今天认识的人是什么样"(用户记忆),一部分记"这活儿我下次该怎么干更顺"(经验)。区别是这里由 LLM 自动写、写进可检索的目录树。

2. 顶层全景(它大概怎么转)

怎么读下面这张图: 从上到下是时间顺序。commit() 只做同步归档就返回;真正的抽取在后台 Phase 2 里跑,产出分两条线回写目录树。

用户: session.commit()
        │
   ┌────┴─────────────────────────────┐
   │ Phase 1 同步(立即返回 task_id)     │
   │  归档本轮消息 → messages.jsonl      │  session.py:1092+ commit()
   └────┬─────────────────────────────┘
        │  asyncio.create_task(...)  session.py:1245
        ▼
   ┌──────────────────────────────────────────────┐
   │ Phase 2 异步后台                                │  _run_memory_extraction
   │                                                │  session.py:1266+
   │   extract_long_term_memories()  ← 唯一入口       │  compressor_v3.py:254
   │        │                                       │
   │   ┌────┴──── 第 1 步 ────┐  ┌── 第 2 步 ──┐      │
   │   │ 抽用户记忆(含 cases) │→│ 按 cases 训练 │      │
   │   └──────────┬──────────┘  └──────┬──────┘      │
   │              ▼                    ▼             │
   │        [用户记忆线]          [Agent 经验线]        │
   │   profile/preferences   trajectory→experience   │
   │   /events/entities...       /skills             │
   │              │                    │             │
   │              └────────┬───────────┘             │
   │                       ▼   第 3 步                │
   │             memory_diff.json 审计日志            │  compressor_v3.py:870
   └───────────────────────┬──────────────────────────┘
                           ▼
                回写 viking:// 目录树(落盘见 04)
                           ▼
                下次检索捞回(见 03)→ 越用越聪明

主要部件一句话职责:

部件	干什么	在哪
`SessionCompressorV3`	抽取总编排;三步走完一次会话	`openviking/session/compressor_v3.py:87`
`ExtractLoop`	ReAct 抽取循环:让 LLM 读上下文、产出记忆操作	`openviking/session/memory/extract_loop.py:82`
`StreamingMemoryUpdater`	把用户记忆操作缓冲、合流、patch-merge 落盘	`openviking/session/memory/streaming_memory_updater.py:120`
`TrajectoryRolloutAnalyzer`	经验线 Phase 1:把执行抽成 trajectory	`openviking/session/train/components/trajectory_analyzer.py:60`
`ExperienceGradientEstimator`	经验线 Phase 2:把 trajectory 蒸馏成 experience 更新信号	`openviking/session/train/components/gradient_estimator.py:41`
`MemoryTypeRegistry`	从 YAML 加载"有哪些记忆类型、写哪、怎么合并"	`openviking/session/memory/memory_type_registry.py:28`

主线走一遍(高层): 会话消息 → extract_long_term_memories → 用户记忆线抽出 profile/events/... 顺带抽出 cases → cases 触发经验线 trajectory→experience → 两条线的改动都汇进 memory_diff.json → 回写目录树。

3. 核心原理(逐个机制,由浅入深)

3.1 唯一入口:`extract_long_term_memories` 的三步

它要解决的小问题: 后台该按什么顺序把一次会话"消化"完?

思路: V3 把整个消化过程收敛成一个方法、三步顺序——先抽用户记忆(其中就包含 cases),再用抽出来的 cases 去训练经验,最后把两步的所有改动合并写成一份审计。

真实实现——SessionCompressorV3.extract_long_term_memories(compressor_v3.py:254)主体三步:

# 示意,非源码;对应 compressor_v3.py:279-308
result = await self._extract_user_memories(messages, ...)      # 第1步 抽用户记忆
train_result = await self.train_from_extracted_cases(          # 第2步 用 cases 训练经验
    cases=result.cases, messages=messages, ...)
await self._write_final_memory_diff(                           # 第3步 汇总审计
    archive_uri, ctx,
    memory_diffs=[result.memory_diff, train_result["memory_diff"]])

关键点:经验线不是独立抽的,它挂在用户记忆抽出的 cases 上。 没有 cases,train_from_extracted_cases 直接空转返回(compressor_v3.py:558-560)。这条设计是 V3 区别于 V2 的核心,下一节展开。

还有一条"快路径":如果本次消息第一条就是批量训练用的 CaseSpec 协议头,直接走 _commit_training_case_fast_path(compressor_v3.py:315),跳过用户记忆抽取。这是离线批量训练用的,正常会话走不到。

3.2 V3 vs V2:为什么 README 说"始终用 v3"

先给结论: 现在的 OpenViking 无条件使用 V3;memory.version 配置项已废弃、被忽略(依据:README.md:346、openviking/session/__init__.py:34-40)。

# openviking/session/__init__.py:34-40  create_session_compressor
if memory_version is not None:
    logger.warning("memory.version is deprecated and ignored; using v3 memory compressor")
return SessionCompressorV3(vikingdb=vikingdb, skill_processor=skill_processor)

两代的分工差异(承重区别):

维度	V2	V3(当前)
经验/轨迹怎么抽	单独一个 `extract_execution_memories` 方法,独立 LLM 调用	不再单独抽;`cases` 是一种普通用户记忆类型,抽到它就顺手触发训练
用户记忆落盘	目录级记忆锁	无目录锁的 patch-merge 流(`StreamingMemoryUpdater` 缓冲合流)
入口	`extract_long_term` + `extract_execution` 两个	只 `extract_long_term_memories` 一个

证据:extract_execution_memories 只在 V2 里有(compressor_v2.py:497),V3 类里没有这个方法。而 session.py 里那段"执行记忆抽取"分支有个前置判断 hasattr(compressor, "extract_execution_memories")(session.py:1431)——对 V3 实例恒为假,所以 V2 那条独立执行记忆通道在 V3 下根本不会触发;经验线全部改由 cases 驱动。V3 类文档字符串把这层意思写死了(compressor_v3.py:3-11)。

提示:docs/design/session-memory-extraction-flow.md 与常量 EXECUTION_MEMORY_TYPES = {trajectories, experiences}(openviking/session/memory/constants.py:6)仍描述 V2 风格的 stage: agent 独立执行记忆抽取。读代码时以"V3 恒用、执行记忆走 cases 训练"为准——那份设计文档描述的是仍保留在 session.py/V2 里、但在 V3 路径上不生效的老通道。

3.3 用户记忆线:ReAct 抽取 → patch-merge 落盘

这条线负责"关于用户的长期事实"。看 _extract_user_memories(compressor_v3.py:434)。

第一步:ReAct 抽取。 ExtractLoop(extract_loop.py:82)是个简化版 ReAct 编排器:给 LLM 一份工具(read/search/ls)+ 一份由 YAML schema 动态生成的输出 JSON Schema,循环最多几轮,让模型要么调工具看已有记忆、要么直接产出记忆操作(extract_loop.py:145 run)。

ReAct 一轮(extract_loop.py:249-314):
  ┌─ prefetch: 系统先 ls + 读 .overview + search 铺上下文
  │
  ├─ LLM 调用 ──► 返回 tool_calls ? ──► 执行工具,continue 下一轮
  │                    │
  │                    └─► 返回最终 operations ?
  │                            ├─ 有未读到的已有文件 → 补读 refetch,continue
  │                            ├─ patch 校验失败 → 发修复指令,重试一次
  │                            └─ 都过了 → break 返回操作
  └─ 每种记忆类型是 schema 里一个字段(extract_loop.py:191-192)

哪些记忆类型可抽,由 MemoryTypeRegistry 从 openviking/prompts/templates/memory/*.yaml 加载(memory_type_registry.py:326 create_default_registry)。每个 YAML 定义了写到哪个目录、文件名模板、字段怎么合并。真实模板举例:

记忆类型	目录	说明
`profile`	`.../memories/profile.md`	用户画像,单文件
`preferences`	`.../memories/preferences/{user}/{topic}.md`	按主题分文件
`events`	`.../memories/events/{年}/{月}/{日}/{名}.md`	按日期归档,`add_only`
`cases`	`.../memories/cases/{名}.md`	触发经验训练的那种,`peer_enabled: false`

第二步:patch-merge 落盘。 抽出的操作不直接写,而是提交给进程内单例 StreamingMemoryUpdater(streaming_memory_updater.py:120,经 get_streaming_memory_updater 取用 :1821)。它的职责是:把多个并发 commit 的记忆写操作缓冲一个小窗口,合流后再 patch-merge 落盘——避免同一记忆文件被并发写覆盖。

# streaming_memory_updater.py:64  窗口配置
max_operations_per_update: int = 8      # 攒够 8 个操作
max_wait_seconds: float = 10.0          # 或等满 10 秒

字段级怎么合并由 MergeOp 决定(merge_op/base.py:99):patch(SEARCH/REPLACE 增量改)、replace(整体替换)、sum(累加计数)、immutable(只写一次)。用户记忆文件就是靠这些 op 原地演进,而不是每次覆盖重写。底层 MemoryUpdater 真正写盘、写关系链的部分复用 04。

3.4 `stage` 与 `peer_enabled`:记忆分给谁

两个 schema 字段管"这条记忆归谁、写哪":

stage(dataclass.py:203)—— user 表示长期用户记忆,agent 表示"执行派生"记忆。默认 user。它把记忆类型分成"用户线/经验线"两组。
peer_enabled(dataclass.py:207)—— 该记忆是否按对话中的"他人"(peer)分目录存。默认 True。

peer 是什么: 一段对话里除了当前用户,可能提到别人(peer)。开启 peer 记忆时,关于某个稳定 peer 的记忆会写到独立子空间 .../peers/{peer_id}/(memory_isolation_handler.py:29-33 peer_user_space)。

peer_enabled: false 的语义(以 cases 为例): 这类记忆忽略 peer_id 和 ranges 的 peer 目标,恒写当前用户空间(memory_isolation_handler.py:208-210)。cases.yaml:17 就设了 peer_enabled: false——因为"这次任务怎么干成的"是 Agent 自己的事,不该按 peer 拆分。

一条抽出的记忆操作,路由决策(memory_isolation_handler.py:191 calculate_memory_uris):
  schema.peer_enabled == false ?
        └─是─► 只写 self(当前用户空间),丢弃 peer_id
        └─否─► 看操作带的 peer_id / ranges:
                 ├─ 无 → 写 self
                 ├─ 合法 peer_id 且在允许集 → 写该 peer 子空间
                 └─ 非法/未授权 peer → 跳过

会话级还有一层 MemoryPolicy(memory_policy.py:53)总开关:self.enabled / peer.enabled / memory_types 白名单 / working_memory.enabled,决定这次 commit 允许写哪些线、哪些类型。

3.5 Agent 经验线:cases → trajectory → experience

它要解决的小问题: 光记住"用户是谁"不够;还要让 Agent 记住"我这类任务的正确干法",下次照着干。

思路——两阶段蒸馏(承重流程): 原始执行记录先落成 trajectory(轨迹:这次一步步干了啥),再从轨迹里蒸馏出 experience(经验:可复用的方法论)。这一层的完整重构见 docs/design/traj-exp-experience-learning-redesign.md。

入口是 train_from_extracted_cases(compressor_v3.py:543)。对每个 cases 记忆:

每个 case(compressor_v3.py:655-707):
  Rollout(case, messages)                              一次"回放"单元
        │
        ▼  Phase 1  compressor_v3.py:663
  rollout_analyzer.analyze() ──► trajectories[]         把执行抽成轨迹
        │  (内部 AgentTrajectoryContextProvider + ExtractLoop)
        │   trajectory_analyzer.py:155;可同时抽 session skills
        ▼  Phase 2  compressor_v3.py:666
  ExperienceGradientEstimator.estimate() ──► gradients[] 蒸馏成"经验更新信号"
        │  (内部 AgentExperienceContextProvider + ExtractLoop)
        │   gradient_estimator.py:106
        ▼  compressor_v3.py:675
  exp_trainer.submit_gradients() ──► 串行 plan→apply    把经验 patch 合并落盘
        │
        └─ 若 case 有 skills 梯度 → skill_trainer 单独应用(compressor_v3.py:690-705)

两个 Provider 是这条线的"两只手":

Phase 1 AgentTrajectoryContextProvider(agent_trajectory_context_provider.py:29):把归档对话喂给 ExtractLoop,产出 trajectories;include_session_skills=True 时,在同一趟 ReAct 里顺带抽出可复用的可执行技能(skills)。
Phase 2 AgentExperienceContextProvider(agent_experience_context_provider.py:42):拿新 trajectory 去检索候选 experience(top-5),让 LLM 决定 更新 / 替换 / 新建 / 跳过;规则是"一个用户意图一条经验、拿不准就拆不合并"(agent_experience_context_provider.py:71-88)。它不直接写,只输出"经验该怎么改"的 patch 语义梯度。

为什么叫"梯度": 这是借了训练术语——把 experiences 目录当成一份可优化的策略集(Experience Policy Set),每条 trajectory 给出一个"往哪个方向改经验"的信号(gradient),由 StreamingPolicyTrainer 串行 reload→plan→apply 安全合并。整条链路是 analyze → estimate → plan → apply(docs/design/traj-exp-experience-learning-redesign.md:1-40)。

3.6 memory_diff:每次 commit 的审计日志

两条线的所有改动,最后并成一份 memory_diff.json 写进本次归档目录(_write_final_memory_diff,compressor_v3.py:870-890)。它记录本次 commit 的 adds / updates / deletes,用于审计与回滚(docs/en/concepts/08-session.md:183)。

_build_memory_diff(compressor_v3.py:152)有个值得学的细节:过滤 no-op 更新——有些 upsert 会"成功"但最终文件内容和旧内容逐字节相同(空合并/重复序列化),这类不写进 diff(compressor_v3.py:237-243 + _same_memory_file)。即使全零也会写一份空 diff,保证每次 commit 都有审计条目。

3.7 记忆生命周期:冷热打分

记忆越攒越多,检索时得让"常被用到、近期更新"的记忆优先。memory_lifecycle.hotness_score(openviking/retrieve/memory_lifecycle.py:19)给每条记忆算一个 0~1 的热度:

# memory_lifecycle.py:48-62(示意公式)
score = sigmoid(log1p(active_count)) * exp(-ln2/half_life * age_days)
#        ↑ 访问频率越高越接近1          ↑ 距上次更新越久衰减越狠(默认半衰期7天)

这个分数会和语义相似度混合,给"高频+新鲜"的记忆在检索结果里加权(检索细节见 03)。active_count 在 Phase 2 抽取时更新(docs/en/concepts/08-session.md:121)。这就补上了闭环最后一环:沉淀的记忆不仅写得进去,还会因为被反复用到而在检索里越排越靠前。

4. 巧妙之处(可借鉴的技术)

把"训练"折叠进"记忆抽取"。 V3 不为经验单开一次 LLM 抽取,而是让 cases 成为一种普通用户记忆类型,抽到就触发训练(compressor_v3.py:3-11)。省一次 LLM 往返,且两条线共享同一份归档消息。
无锁 patch-merge 合流。 并发 commit 的记忆写不抢目录锁,而是攒进小窗口(8 个操作 / 10 秒)合并后一次落盘(streaming_memory_updater.py:64),用批处理换掉锁竞争。
trajectory / experience 两阶段解耦。 原始轨迹(全量、可追溯)与蒸馏经验(精炼、可复用)分开存,经验更新走"梯度→串行 plan/apply"保证并发安全(设计文档"串行边界")。
diff 过滤 no-op。 只把真正改了内容的写进审计(compressor_v3.py:237-243),让 memory_diff.json 干净、可信。
冷热打分做时间衰减。 用 log1p + sigmoid 压访问频率、指数衰减压时间(memory_lifecycle.py:48-62),把"常用+新鲜"直接量化进检索排序。

5. 边界与局限(诚实)

经验线依赖 cases。 若用户记忆抽取没产出 cases,train_from_extracted_cases 空转返回(compressor_v3.py:558-560),这次会话就不产经验。经验质量间接受"cases 抽得准不准"制约。
V2 的独立执行记忆通道在 V3 下是死代码。 session.py:1431 的 hasattr(..., "extract_execution_memories") 对 V3 恒假;EXECUTION_MEMORY_TYPES 常量与 session-memory-extraction-flow.md 仍描述该老路径,容易误读。以 V3 的 cases 驱动为准。
全靠 LLM 判断,可能抽错/抽漏。 抽取是 ReAct + JSON Schema 约束,有格式重试与 patch 修复(extract_loop.py:298-313),但语义层面的漏抽/错归属没有硬保证。
异步不保证即时可见。 抽取在后台 Phase 2 跑,commit() 返回时记忆尚未写完;需要用 task_id 轮询(docs/en/concepts/08-session.md:66-80)。
peer 记忆需谨慎授权。 只有在 allowed_peer_ids 里的合法 peer 才写,非法/未授权 peer 直接跳过(memory_isolation_handler.py:178-189)——设计上防串号,但也意味着未授权 peer 的信息不会沉淀。

6. 代码地图(导航索引)

主题	文件路径	符号名
抽取总编排(三步)	`openviking/session/compressor_v3.py`	`SessionCompressorV3.extract_long_term_memories`
用户记忆抽取	`openviking/session/compressor_v3.py`	`SessionCompressorV3._extract_user_memories`
cases 触发经验训练	`openviking/session/compressor_v3.py`	`SessionCompressorV3.train_from_extracted_cases`
memory_diff 汇总审计	`openviking/session/compressor_v3.py`	`_build_memory_diff` / `_write_final_memory_diff`
快路径(批量训练)	`openviking/session/compressor_v3.py`	`_commit_training_case_fast_path`
恒用 V3	`openviking/session/__init__.py`	`create_session_compressor`
V2 独立执行记忆(对照)	`openviking/session/compressor_v2.py`	`extract_execution_memories`
执行记忆类型常量	`openviking/session/memory/constants.py`	`EXECUTION_MEMORY_TYPES`
ReAct 抽取循环	`openviking/session/memory/extract_loop.py`	`ExtractLoop.run`
用户记忆合流落盘	`openviking/session/memory/streaming_memory_updater.py`	`StreamingMemoryUpdater.submit` / `get_streaming_memory_updater`
记忆类型加载	`openviking/session/memory/memory_type_registry.py`	`MemoryTypeRegistry` / `create_default_registry`
stage / peer_enabled 定义	`openviking/session/memory/dataclass.py`	`MemoryTypeSchema`
peer / self 路由	`openviking/session/memory/memory_isolation_handler.py`	`MemoryIsolationHandler.calculate_memory_uris`
字段合并算子	`openviking/session/memory/merge_op/base.py`	`MergeOp`
经验线 Phase 1(轨迹)	`openviking/session/memory/agent_trajectory_context_provider.py`	`AgentTrajectoryContextProvider`
经验线 Phase 2(经验)	`openviking/session/memory/agent_experience_context_provider.py`	`AgentExperienceContextProvider`
轨迹分析器	`openviking/session/train/components/trajectory_analyzer.py`	`TrajectoryRolloutAnalyzer.analyze`
经验梯度估计	`openviking/session/train/components/gradient_estimator.py`	`ExperienceGradientEstimator.estimate`
会话级记忆策略	`openviking/session/memory_policy.py`	`MemoryPolicy`
冷热生命周期打分	`openviking/retrieve/memory_lifecycle.py`	`hotness_score`
后台 Phase 2 编排	`openviking/session/session.py`	`_run_memory_extraction`

1. 这是什么(零基础也能懂)​

2. 顶层全景(它大概怎么转)​

3. 核心原理(逐个机制,由浅入深)​

3.1 唯一入口:extract_long_term_memories 的三步​

3.2 V3 vs V2:为什么 README 说"始终用 v3"​

3.3 用户记忆线:ReAct 抽取 → patch-merge 落盘​

3.4 stage 与 peer_enabled:记忆分给谁​

3.5 Agent 经验线:cases → trajectory → experience​

3.6 memory_diff:每次 commit 的审计日志​

3.7 记忆生命周期:冷热打分​

4. 巧妙之处(可借鉴的技术)​

5. 边界与局限(诚实)​

6. 代码地图(导航索引)​