数据截至 (上游 commit 3590b47a25bd)
会话记忆自迭代: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:184 |
ExtractLoop | ReAct 抽取循环:让 LLM 读上下文、产出记忆操作 | openviking/session/memory/extract_loop.py:82 |
StreamingMemoryUpdater | 把用户记忆操作缓冲、合流、patch-merge 落盘 | openviking/session/memory/streaming_memory_updater.py:125 |
TrajectoryRolloutAnalyzer | 经验线 Phase 1:把执行抽成 trajectory | openviking/session/train/components/trajectory_analyzer.py:66 |
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:352)主体三步:
# 示意,非源码;对应 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:865-867)。这条设计是 V3 区别于 V2 的核心,下一节展开。
还有一条"快路径":如果本次消息第一条就是批量训练用的
CaseSpec协议头,直接走_commit_training_case_fast_path(compressor_v3.py:463),跳过用户记忆抽取。这是离线批量训练用的,正常会话走不到。
3.2 V3 vs V2:为什么 README 说"始终用 v3"
先给结论: 现在的 OpenViking 无条件使用 V3;memory.version 配置项已废弃、被忽略(依据:openviking/session/__init__.py:27-35,docstring 明写 "Deprecated and ignored; v3 is always used" 并对旧配置打 warning;早期 README 里的对应说明条目已随 README 精简移除)。
# 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 一个 |
证据:V2 的独立执行记忆入口 extract_execution_memories 在当前 commit 已随 compressor_v2.py 整体删除,session.py 也不再 hasattr 探测——session.py:62 直接 from ...compressor_v3 import SessionCompressorV3 as SessionCompressor,运行时只有 V3 一个实现;经验线全部改由 cases 驱动。V3 类文档字符串把这层意思写死了(compressor_v3.py:3-11:训练案例不是独立 LLM 调用,而是同一 ExtractLoop 产出的普通 cases 记忆类型)。
提示:
docs/design/session-memory-extraction-flow.md与常量EXECUTION_MEMORY_TYPES = {trajectories, experiences}(openviking/session/memory/constants.py:9)仍描述 V2 风格的stage: agent独立执行记忆抽取。读代码时以"V3 恒用、执行记忆走 cases 训练"为准——那份设计文档描述的是仍保留在session.py/V2 里、但在 V3 路径上不生效的老通道。
3.3 用户记忆线:ReAct 抽取 → patch-merge 落盘
这条线负责"关于用户的长期事实"。看 _extract_user_memories(compressor_v3.py:593)。
第一步: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 加载(openviking/session/memory/memory_type_registry.py:312 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:125,经 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:124):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:253-256)。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:88)总开关: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:849)。对每个 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:1197-1217)。它记录本次 commit 的 adds / updates / deletes,用于审计与回滚(docs/en/concepts/08-session.md:178)。
_build_memory_diff(compressor_v3.py:250)有个值得学的细节:过滤 no-op 更新——有些 upsert 会"成功"但最终文件内容和旧内容逐字节相同(空合并/重复序列化),这类不写进 diff(compressor_v3.py:335-341 + _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:69),用批处理换掉锁竞争。 - trajectory / experience 两阶段解耦。 原始轨迹( 全量、可追溯)与蒸馏经验(精炼、可复用)分开存,经验更新走"梯度→串行 plan/apply"保证并发安全(设计文档"串行边界")。
- diff 过滤 no-op。 只把真正改了内容的写进审计(
compressor_v3.py:335-341),让memory_diff.json干净、可信。 - 冷热打分做时间衰减。 用
log1p + sigmoid压访问频率、指数衰减压时间(memory_lifecycle.py:48-62),把"常用+新鲜"直接量化进检索排序。
5. 边界与局限(诚实)
- 经验线依赖
cases。 若用户记忆抽取没产出cases,train_from_extracted_cases空转返回(compressor_v3.py:865-867),这次会话就不产经验。经验质量间接受"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:300-312),但语义层面的漏抽/错归属没有硬保证。 - 异步不保证即时可见。 抽取在后台 Phase 2 跑,
commit()返回时记忆尚未写完;需要用task_id轮询(docs/en/concepts/08-session.md:66-80)。 - peer 记忆需谨慎授权。 只有在
allowed_peer_ids里的合法 peer 才写,非法/未授权 peer 直接跳过(memory_isolation_handler.py:224-230的_resolve_operation_target_id)——设计上防串号,但也意味着未授权 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 |
| V3 类文档字符串(训练案例=普通记忆类型) | openviking/session/compressor_v3.py | SessionCompressorV3(docstring :3-11) |
| 执行记忆类型常量 | 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 |