主动记忆整合：Praxis中睡眠通过模式的实现

主动记忆整合：Praxis中睡眠通过模式的实现

Praxis v0.28 引入了一个后台“睡眠”通道，它重放、连接并压缩智能体记忆，形成跨场景洞见——使用 BM25 检索，无需嵌入模型，并作为只读风险操作进行治理。

大多数智能体记忆系统是被动的：将文本块嵌入一次后存储，后续按相似度检索。这对文档有效，但对连续多天运行数百轮的智能体却失效了，因为原始轨迹无限增长，而信号——模式、决策、交叉引用——从未被提炼。

Praxis v0.28 本周推出了不同的设计。不是对不断增长的堆做被动 RAG，而是一个后台整合通道定期重放近期记忆、提取结构化元数据、找到表面不相关的条目之间的连接、合成一个跨场景洞见，并重新评估重要性。这个比喻借鉴了 GCP "Always-On Memory Agent“ 研究：这是智能体在睡眠时的大脑活动。它以五个切片在 feat(memory) 提交 5a98771 到 1b695e0 中发布。

这篇文章将带你走过实际的实现——层级模型、整合算法、治理门控、检索路径以及诚实的工程权衡。每个代码块均来自 v0.28.3 的 hybridagent/。

问题：记忆囤积

一个朴素的智能体记忆是只追加的。每一轮都写入一条摘要。一周后你会得到数百条情节性条目，其中大部分已过时，且互不相连。召回返回与查询最相似的条目，但这很少是最相关的——重要的决策发生在三天前，被四十条常规工具调用日志淹没。

这个失败模式在 Praxis 代码库中有一个名字：summarize-not-hoard。Memory.add_durable() 方法通过一个硬限制来强制执行它：

`python
# hybridagent/memory.py
_MAX_DURABLE_CHARS = 280

def add_durable(self, text: str, kind: str, provenance: str,
salience: float = 1.0,
expires_at: float | None = None) -> MemoryItem:
# Enforce summarize-not-hoard: durable entries are concise.
clipped = text if len(text) <= _MAX_DURABLE_CHARS else text[:_MAX_DURABLE_CHARS] + "…"
item = MemoryItem(text=clipped, tier=Tier.DURABLE, kind=kind,
provenance=provenance, salience=salience,
expires_at=expires_at)
`

280 个字符。一条持久记忆是一条推文长度的事实，不是一份转录。私有正文从不持久存储——只有简洁的摘要加来源。这是一个故意的约束：它迫使智能体（或操作者）在持久化之前进行压缩，并使持久层级可以在几秒钟内被人类扫描。

三个层级，一条召回路径

Praxis 将记忆分为三个层级，每个有不同的保留策略：

`python
class Tier(str, Enum):
WORKING = "working"
EPISODIC = "episodic"
DURABLE = "durable"
`

关键设计决策：只有一条召回路径。Memory.recall(query) 同时对持久记忆和情节记忆进行排序。整合产生的洞见不在独立命名空间中——它们是 kind=”insight“ 的持久记忆，通过与其他任何事实相同的 recall() 调用被呈现。

`python
def recall(self, query: str, k: int = 5) -> list[MemoryItem]:
"""Rank durable+episodic memory by BM25 relevance to query, then
re-rank by salience, recency, and prior access."""
now = time.time()
pool = [it for it in (self.durable + self.episodic)
if not (it.expires_at and it.expires_at <= now)]
if not pool:
return []
index = BM25Index.build((str(i), it.text) for i, it in enumerate(pool))
relevance = dict(index.search(query, k=len(pool)))
candidates: list[tuple[float, int, MemoryItem]] = []
for i, item in enumerate(pool):
rel = relevance.get(str(i), 0.0)
if rel <= 0.0:
continue # shares no query term -> not a match
age_days = max(0.0, (now - item.ts) / 86400.0)
freshness = 1.0 / (1.0 + age_days / 30.0)
score = rel + item.salience + (0.1 * item.access_count) + freshness
candidates.append((score, i, item))
candidates.sort(key=lambda t: (-t[0], t[1]))
...
`

最终分数是 BM25_relevance + salience + 0.1 × access_count + freshness。BM25 处理词汇相关性；salience 是一个治理信号（整合会提升它）；access_count 奖励智能体反复访问的条目；freshness 是一个以 30 天窗口的软衰减。没有嵌入、没有向量数据库、没有 GPU。

在智能体自身记忆上使用 BM25

检索骨干是 hybridagent/bm25.py 中一个 97 行纯标准库的 BM25 实现。它之所以存在，是因为智能体的记忆语料库很小——可能只有几百个条目——而且 BM25 在小语料库上对精确词和罕见词的匹配很强，不需要嵌入模型这种过度杀伤，也不会有向量索引的操作开销。

一个细节很重要：Praxis 使用的是 Lucene 风格的 IDF 变体，而不是经典的 Okapi：

`python
def _idf(self, term: str) -> float:
df = self._df.get(term, 0)
return math.log(1 + (self.n - df + 0.5) / (df + 0.5))
`

经典 Okapi IDF 是 log((N - df + 0.5)/(df + 0.5))，当某个词出现在超过一半的文档中时会变成负数。在智能体的小语料库中，如果常用词占主导，那就会把好的匹配清零。log(1 + ...) 形式总是非负的，所以即使你的全部记忆只有 40 个条目且其中 25 个包含 ”praxis“，排序器也能合理运行。

索引每次查询构建都很廉价——BM25Index.build() 在每次 recall() 中都是新调用的。没有失效、没有持久化、没有后台重新索引。代价是 O(N) 每查询；收益是零活动部件。

整合通道：四个操作，一份报告

hybridagent/consolidation.py 有 349 行，暴露了一个公共方法：MemoryConsolidator.run()。一次通道按顺序做四件事：

1. 提取元数据——每个条目的实体和主题，写入 memory_items 表。
2. 寻找连接——窗口内的成对关系，上限为 max_connections（默认 5）。
3. 合成一条洞见——一个没有任何单条记忆能单独表述的跨模式模式，写入为 kind=”insight" 的持久条目。
4. 重新评估重要性——对连接和被近期召回的条目进行有界、单调的提升。

`python
def run(self, re_consolidate_after: float | None = None) -> ConsolidationReport:
report = ConsolidationReport()
window = self._select_window(re_consolidate_after)
report.items_reviewed = len(window)
if len(window) < self.min_items:
report.skipped_reason = f"too few unconsolidated ({len(window)} < {self.min_items})"
return report

ts = time.time()
ids = [w["id"] for w in window]

if self.extract_metadata:
self._extract_metadata(window)
conns = self._find_connections(window)
report.connections_made = len(conns)
insight_id = self._synthesize_insight(window, conns, ts)
if insight_id is not None:
report.insights_written = 1
if self.rerate_salience:
report.salience_rerated = self._rerate(window, conns)
self.store.mark_consolidated(ids, ts=ts)
return report
`

窗口是有界的（window_size 默认 20——GCP 的参考用了 10；20 对噪声更大的智能体轨迹更安全）且只拉取尚未整合的条目（list_unconsolidated）。一个最小阈值（min_items 默认 3——GCP 用了 2；3 减少了微小通道的噪声）会跳过那些会为了两个条目而消耗一次 LLM 调用提取元数据的运行。

洞见合成提示

洞见步骤是“睡眠”比喻的用武之地。提示要求 LLM 找到一个没有任何单条记忆能单独表述的模式：

`python
def _insight_prompt(self, window: list[dict], conns: list[_Conn]) -> str:
lines = [f"[#{w['id']}] {w['text']}" for w in window]
conn_lines = [f"- #{c.from_id} -> #{c.to_id}: {c.relationship}"
for c in conns] or ["(no connections found)"]
return (
"Given these memories and the connections found between them, "
"synthesize ONE cross-cutting insight — a pattern, theme, or "
"implication that no single memory states alone. Be concrete, "
"not generic. 1-2 sentences max.\n\n"
"Memories:\n" + "\n".join(lines) + "\n\nConnections:\n"
+ "\n".join(conn_lines) + "\n\nInsight:"
)
`

生成的洞见被写入为一条正常的持久记忆，kind="insight"，salience=0.8，并且产生它的连接通过 insight_id 回溯：

`python
item = self.memory.add_durable(
text, kind="insight", provenance=f"consolidation:{ts:.0f}",
salience=0.8,
)
insight_id = getattr(item, "id", None)
if insight_id is not None:
for c in conns:
self.store.add_memory_connection(
c.from_id, c.to_id, c.relationship,
insight_id=insight_id, created_at=ts)
`

这是关键的重用决策：洞见通过与其他任何持久事实相同的 recall() 路径流回。没有新的检索命名空间，没有特殊的“洞见查询”端点。当智能体下次为一个目标召回上下文时，合成的洞见与人工编写的事实平等竞争——按 BM25 + salience + 新鲜度排序。

重要性重新评估是有界且单调的

_rerate() 方法刻意保守。它只提升重要性，从不降低，且提升幅度很小：

`python
def _rerate(self, window: list[dict], conns: list[_Conn]) -> int:
"""+0.1 per connection received, +0.05 per access since last pass.
Monotonic (never down-rates). Hard cap at 1.0."""
...
for w in window:
mid = w["id"]
old_salience = float(w.get("salience", 1.0))
bump = 0.0
if conn_by_id.get(mid, 0):
bump += 0.1 * conn_by_id[mid]
if w.get("access_count", 0) > 0 and (
last_consolidated is None or
w.get("last_access_ts", 0) > last_consolidated):
bump += 0.05
if bump > 0:
self.store.update_memory_salience(mid, old_salience + bump)
bumped += 1
return bumped
`

+0.1 每个连接（被整合器链接到其他条目的条目可能很重要），+0.05 如果自上次通道后被召回（智能体反复访问的条目可能很重要）。存储会钳位到 1.0。这里没有衰减——衰减由 decay_episodic() 和 purge_expired() 单独处理，这是操作者控制的显式保留策略方法。整合只强化；它不会默默遗忘。

治理：只读风险，默认关闭

这是大多数“智能体记忆”文章跳过的一步。整合在后台对智能体自身的记忆运行 LLM。这是一个治理决策，而不仅仅是一个功能开关。

Praxis 将整合通道归类为只读风险——它读取记忆，写入本地洞见和连接行，没有外部影响。它不会经过发送/破坏性操作审批关口。但它在 v0.28.0 中仍然默认关闭：

`python
# hybridagent/config.py
_CONSOLIDATION_DEFAULTS = {
"enabled": False, # v0.28.0 ships off-by-default; flip after dogfood
"intervalMinutes": 30,
"windowSize": 20, # max memories per pass (GCP used 10; 20 is safer)
"minItemsToConsolidate": 3, # GCP used 2; 3 reduces noise from tiny passes
"rerateSalience": True, # Gap D - bump connected/recalled items
"extractMetadata": True, # Gap C - entities/topics on the window
"maxConnections": 5,
}
`

驱动它的守护进程嘀嗒在禁用时是一个廉价的无操作——一次配置读取，一次时间戳检查，然后返回：

`python
# hybridagent/daemon.py
def _consolidation_tick(self) -> None:
"""Gated by agents.consolidation.enabled (default off in v0.28.0).
When off, this method is a cheap no-op."""
try:
from .config import get_consolidation_config
cc = get_consolidation_config()
except Exception:
return
if not cc.get("enabled", False):
return
...
`

该嘀嗒内部还有两个保护：

- 当有未完成的工作时延迟执行。如果任务队列中有待处理或重试项，该嘀嗒会重新安排在 60 秒后执行。整合永远不会饿死面向用户的循环。
- 出错时退避。一个失败的通道（LLM 超时、格式错误的 JSON、存储错误）重新安排到 min(interval, 300s)，这样失败的通道不会立即重试并消耗 token。

`python
# Don't starve the task queue — defer if work is pending/running.
if self.manager is not None and (
self.manager.list(status="pending") or self.manager.list(status="retry")
):
self._next_consolidation_ts = now + 60.0
return
`

操作者用一个 CLI 命令打开它，extractMetadata 的运行时翻转在下一个嘀嗒时生效，无需重启守护进程：

`
praxis consolidation enable
praxis consolidation status
praxis consolidation run # trigger one pass immediately
`

诚实失败是设计约束

整合器中的每个 LLM 调用都被包装为永远不会引发异常。格式错误的 JSON、网络超时、空响应——所有情况都会产生带有 skipped_reason 设置和日志行的报告，然后通道继续执行它获得的任何内容。测试明确固定了这一点：

`
test_consolidation_malformed_metadata_json_does_not_block
test_consolidation_malformed_connections_json_still_writes_insight
test_consolidation_empty_insight_response_skips_insight
test_consolidation_rerates_salience_bounded_monotonic
test_consolidation_connection_to_self_rejected
test_consolidation_respects_max_connections_cap
test_consolidation_json_parser_strips_fences_and_prose
`

JSON 解析器容忍 LLM 输出的现实世界失败模式——`json fences、末尾 prose、list-vs-dict 混淆：

`python
@staticmethod
def _parse_json_list(raw: str) -> Any:
"""Strict JSON parse with fence-stripping. Returns the parsed value,
or raises ValueError (caller catches and skips)."""
text = raw.strip()
if text.startswith("`"):
inner = text.split("`", 2)
if len(inner) >= 2:
body = inner[1]
if body.lower().startswith("json"):
body = body[4:]
text = body.strip()
start = text.find("[")
end = text.rfind("]")
if start == -1 or end == -1 or end < start:
raise ValueError("no JSON array found in LLM response")
return json.loads(text[start:end + 1])
`

与自身的连接（from_id == to_id）被拒绝。指向窗口外 ID 的连接被拒绝。重复的 (from_id, to_id, relationship) 三元组被 SQLite 中的 UNIQUE 约束拒绝。删除一条记忆会级联到其连接（FK ON DELETE CASCADE）；删除一条洞见会将 insight_id 列设置为 NULL 而不是级联（你保留连接，但丢失来源）。这些不是边缘情况——它们是经过测试固定的行为。

完整循环，端到端

总而言之，Praxis 智能体循环现在是：

感知 → 规划 → 治理 → 执行/起草 → 反思 → 整合
↑
recall() 将持久和情节记忆
注入系统提示中

整合是第六步，而不是第五步。它在反思之后运行，在后台按定时器，默认关闭。当它触发时：

1. 选择窗口（≤20 个未整合条目，如果 <3 则跳过）
2. 提取每个条目的实体+主题 → memory_items 元数据
3. 寻找成对连接（≤5） → memory_connections 表
4. 合成一条洞见（1-2 句） → 持久，kind="insight"，salience 0.8
5. 重新评估重要性（+0.1 每连接，+0.05 如果被召回）
6. 将窗口标记为已整合（幂等）

下次智能体处理目标时，recall_context() 将 top-k 记忆前置到系统提示中——包括整合产生的任何洞见。智能体不知道哪些事实是人工编写的，哪些是合成的。它们在相同的排序上竞争。

权衡，诚实陈述

这种模式并非免费。成本如下：

最大的诚实限制：我还没有在生产中运行它数月。v0.28.3 默认关闭它，带有注释 # flip after dogfood。默认值保守，因为实际工作负载的正确节奏、窗口大小和连接上限未知。GCP 参考数字（window=10, min=2）是为他们的工作负载调优的，不是我们的。请将默认值视为起点，而非推荐。

这不是什么

这不是向量数据库的替代品。Praxis 有 hybridagent/rag.py 和 hybridagent/embeddings.py 用于对外部知识库进行语义检索——那是一个不同的问题，有不同的规模（数千份文档，而非数百条记忆条目）。整合操作在智能体自身的记忆上，其语料库足够小，以至于 BM25 是合适的工具，而嵌入会增加操作成本而没有明确收益。

这也不是显式保留策略的替代品。purge_expired()（用于具有 expires_at 的条目）、decay_episodic()（用于低重要性的旧条目）和 forget_by_provenance()（用于被遗忘权）都是操作者有意调用的独立方法。整合强化信号；它不决定遗忘什么。

自己验证

整个事情可以在没有 LLM 密钥的情况下测试。整合器接受一个 _LLMLike 协议（LLMClient 的鸭子类型），因此测试传递返回固定 JSON 的假对象：

`
cd ~/smf-praxis
python3 -m pytest tests/test_consolidation.py -q # 969 lines, ~45 tests
python3 -m pytest tests/test_memory_search.py -q # BM25 recall
python3 -m hybridagent.cli eval # 40/40 capability + safety
`

模式添加（memory_connections 表，entities/topics/last_consolidated_at 列）是附加性的——无需迁移，在翻转 enabled=true 之前不会改变现有记忆的行为。

给构建者的要点

如果你正在构建一个运行超过几轮的智能体：

- 对只追加日志的被动 RAG 会退化。堆在增长；信号没有。一个重放、连接和压缩的整合通道是结构性的修复。
- BM25 是智能体自身记忆的正确检索骨干。语料库很小，精确词匹配很重要，而且你避免了嵌入模型的操作开销。将嵌入留给外部知识库。
- 合成的洞见应该与人工编写的事实在同一排序上竞争。独立的“洞见命名空间”意味着智能体永远不会呈现它们。将洞见作为带有 kind 标签的正常持久记忆写入，让 recall() 排序。
- 整合是一个治理决策。它在后台对智能体记忆运行 LLM。分类风险，在标志后门控，默认关闭，并使成本可见。
- 诚实失败是不可协商的。通道中的每个 LLM 调用都必须被包装为永不引发异常。格式错误的 JSON 跳过该步骤；通道用已有的内容继续。用测试固定这一点。
- 有界、单调的重要性提升。只强化，从不默默遗忘。让衰减成为操作者控制的显式操作。

代码在 github.com/smfworks/smf-praxis 下 hybridagent/consolidation.py、hybridagent/memory.py 和 hybridagent/bm25.py。界定五个切片的阶段计划是 praxis-consolidation-phase-plan.md 在仓库中。启发了这项工作的 GCP 研究在 consolidation 模块 docstring 中引用了。