soul-protocol · 记忆系统 · 2026-06-05

soul-protocol 的记忆设计:不是把聊天塞进向量库,而是一条分层的认知管线

这篇项目讲解基于本地 soul-protocol 源码,拆解它的记忆系统如何从 observe 触发写入、按显著性和事实抽取分层存储、用去重与 supersede 处理冲突,再通过 BM25/ACT-R、图谱增强和可选 LLM rerank 完成召回。

聚焦
记忆写入、存储、冲突更新、召回与检索机制
源码
assets/projects/soul-protocol
查看仓库

先给结论

soul-protocol 的记忆系统不是“每轮聊天 embedding 后写进向量库”。它更像一条认知管线:交互结束后由 Soul.observe() 触发学习,MemoryManager.observe() 判断这轮交互是否值得变成情节记忆,事实抽取器再把稳定信息变成语义记忆,冲突更新用 superseded_by 保留旧事实的审计线索,召回时再用文本候选、图谱扩展、ACT-R activation 和可选 LLM rerank 逐层筛选。

soul-protocol 记忆系统的主干:observe 负责分层写入,recall 负责跨 store 检索、图谱扩展和 activation 排序;向量与 LLM rerank 是可选增强。
soul-protocol 记忆系统的主干:observe 负责分层写入,recall 负责跨 store 检索、图谱扩展和 activation 排序;向量与 LLM rerank 是可选增强。

默认路径

observe 后处理一轮交互;运行时写入 episodic、semantic、procedural、social 等内存 store;save/export 时把 MemoryManager.to_dict() 写成本地 .soul 或 JSON 布局;召回默认走 BM25/text candidates + ACT-R activation。

可选增强

VectorSearchStrategy 可以替换候选检索策略;smart_recall 可以在候选池上做 LLM rerank;domain、layer、visibility、scope 让同一套记忆进入更细的隔离和权限路径。

源码证据地图

当前机制边界是运行时 memory lifecycle:dream、importer、evaluation 和 soul evolution 会消费或补充记忆状态,但不参与 observe 到 recall 的默认主链路。核心证据来自这些文件:

text
1# 1 assets/projects/soul-protocol/src/soul_protocol/runtime/soul.py2# 2 assets/projects/soul-protocol/src/soul_protocol/runtime/types.py3# 3 assets/projects/soul-protocol/src/soul_protocol/runtime/memory/manager.py4# 4 assets/projects/soul-protocol/src/soul_protocol/runtime/memory/episodic.py5# 5 assets/projects/soul-protocol/src/soul_protocol/runtime/memory/semantic.py6# 6 assets/projects/soul-protocol/src/soul_protocol/runtime/memory/dedup.py7# 7 assets/projects/soul-protocol/src/soul_protocol/runtime/memory/recall.py8# 8 assets/projects/soul-protocol/src/soul_protocol/runtime/memory/activation.py9# 9 assets/projects/soul-protocol/src/soul_protocol/runtime/memory/strategy.py10# 10 assets/projects/soul-protocol/src/soul_protocol/runtime/embeddings/vector_strategy.py11# 11 assets/projects/soul-protocol/src/soul_protocol/runtime/storage/file.py

宏观架构

读者最容易卡住的问题是:为什么这里叫 memory,但不是一张 memories 表?答案是,soul-protocol 把不同“记住方式”拆成不同生命周期。

从宏观看,soul-protocol 把“记忆”拆成几类不同生命周期的东西,而不是统一塞进一个 memory 表。MemoryType 里有 coreepisodicsemanticproceduralsocial 五类,MemoryEntry 又补了 importancesomaticsignificanceaccess_timestampssuperseded_byabstractsaliencevisibilityscopeuser_idlayerdomain 等字段。

这些字段不是装饰。importance 决定基本重要性,significancesomatic 来自 observe 时的心理信号,access_timestamps 支撑“越常想起越容易被想起”,superseded_by 让事实更新变成软替换,visibility/scope/user_id/domain/layer 则把召回变成带权限和命名空间的检索。

python
1# assets/projects/soul-protocol/src/soul_protocol/runtime/types.py2class MemoryType(StrEnum):3    CORE = "core"4    EPISODIC = "episodic"5    SEMANTIC = "semantic"6    PROCEDURAL = "procedural"7    SOCIAL = "social"8 9class MemoryEntry(BaseModel):10    type: MemoryType11    content: str12    importance: int = Field(default=5, ge=1, le=10)13    access_timestamps: list[datetime] = Field(default_factory=list)14    significance: float = 0.015    superseded_by: str | None = None16    abstract: str | None = None17    salience: float = Field(default=0.5, ge=0.0, le=1.0)18    visibility: MemoryVisibility = MemoryVisibility.BONDED19    user_id: str | None = None20    layer: str = ""21    domain: str = "default"

这段代码说明 MemoryEntry 同时承担内容、生命周期、召回排序、冲突更新和权限过滤的状态。比如用户说“我搬到 Amsterdam 了”,新事实不会简单覆盖旧事实;旧事实会保留,但 superseded_by 会指向新事实或标记 raw-text contradiction,默认召回时旧事实被过滤掉。

写入什么时候触发

自动学习的入口不是 recall(),也不是模型生成回复的中途,而是每轮 user-agent exchange 结束后的 Soul.observe()。这个入口接收 Interaction,再交给 MemoryManager.observe() 做分层处理。

python
1# assets/projects/soul-protocol/src/soul_protocol/runtime/soul.py2async def observe(3    self,4    interaction: Interaction,5    *,6    user_id: str | None = None,7    domain: str = "default",8) -> None:9    """Soul observes an interaction and learns from it."""10    result = await self._memory.observe(interaction, user_id=user_id, domain=domain)

机制上可以把 observe() 理解成“事后学习 hook”。它的输入是一轮完整交互,输出不是一条单独 memory,而是一组运行结果:是否写了 episodic,抽出了哪些 facts,识别了哪些 entities,是否发生 contradictions,以及这些结果怎样影响 soul 的 bond、skills、evaluation 和 evolution。

observe 写入管线:先看情绪和显著性,再决定是否保存情节;语义事实单独抽取,低显著性但有事实的交互会被提升为值得保存的情节。
observe 写入管线:先看情绪和显著性,再决定是否保存情节;语义事实单独抽取,低显著性但有事实的交互会被提升为值得保存的情节。
1交互结束

调用者把 Interaction 传给 Soul.observe。

2检测情绪和显著性

MemoryManager 生成 somatic marker 和 significance score。

3写 episodic

显著交互写入 EpisodicStore;低显著但抽到事实时也会被提升。

4抽 semantic facts

Cognitive processor 从输入输出里抽取 MemoryType.SEMANTIC。

5去重与冲突更新

dedup、prefix conflict 和 contradiction detector 标记旧事实。

6实体和 self-model 深处理

没有被短路时继续抽实体、更新 graph 和 self_model。

写入什么东西

observe() 不是无条件把整轮聊天写进长期记忆。它先算情绪和显著性;只有显著交互才写 episodic。episodic 的内容是完整 user/agent 片段,所以它保留上下文和时间感。

python
1# assets/projects/soul-protocol/src/soul_protocol/runtime/memory/episodic.py2content = f"User: {interaction.user_input}\nAgent: {interaction.agent_output}"3 4entry = MemoryEntry(5    id=memory_id,6    type=MemoryType.EPISODIC,7    content=content,8    importance=importance,9    emotion=somatic.label if somatic else None,10    significance=significance,11    access_timestamps=[interaction.timestamp],12)

事实写入走另一条路。MemoryManager.observe() 会调用 cognitive extractor 生成 semantic facts;如果某轮对话一开始显著性不高,但抽到了事实,它会反向创建 episodic。这个分支很关键:系统不是“低显著就什么都不学”,而是把“有事实可学”当作晋升信号。

python
1# assets/projects/soul-protocol/src/soul_protocol/runtime/memory/manager.py2facts = await self._cognitive.extract_facts(3    interaction,4    self._semantic.facts(),5    significance=sig_score,6)7 8if not significant and facts:9    significant = True10    episodic_id = await self._episodic.add_with_psychology(11        interaction,12        somatic=somatic,13        significance=sig_value,14    )

一个具体例子是:用户只是顺手说“以后代码示例尽量用 TypeScript”。这句话可能不算情绪强烈的大事件,但 fact extractor 能抽出偏好事实。于是 semantic store 会写入偏好,episodic 也会保留这轮上下文,后续召回时可以解释这个偏好来自哪次交流。

存储形态

运行时,memory 先存在多个 store 中:EpisodicStore._memoriesSemanticStore._factsProceduralStore._proceduresSocialStore._entries,再加上 graph、self_model、archives 和 custom_layers。MemoryManager.to_dict() 把它们合成一个可持久化结构。

python
1# assets/projects/soul-protocol/src/soul_protocol/runtime/memory/manager.py2return {3    "core": self._core_manager.get().model_dump(),4    "episodic": [entry.model_dump(mode="json") for entry in self._episodic.entries()],5    "semantic": [6        fact.model_dump(mode="json")7        for fact in self._semantic.facts(include_superseded=True)8    ],9    "procedural": [proc.model_dump(mode="json") for proc in self._procedural.entries()],10    "social": [entry.model_dump(mode="json") for entry in self._social.entries()],11    "graph": self._graph.to_dict(),12    "self_model": self._self_model.to_dict(),13    "archives": [archive.model_dump(mode="json") for archive in self._archival.all_archives()],14    "custom_layers": custom_dump,15}

持久化发生在 Soul.save()save_local()export() 等路径里。Soul._build_storage_memory_data() 先拿到 memory dict,再叠加 trust chain 和 keys;文件存储层只负责把这个 dict 写到稳定布局。

python
1# assets/projects/soul-protocol/src/soul_protocol/runtime/soul.py2def _build_storage_memory_data(self, *, include_keys: bool) -> dict:3    memory_data = self._memory.to_dict()4    memory_data["trust_chain"] = self._trust_chain_manager.to_dict()5    memory_data["keys"] = self._keystore.to_archive_files(include_private=include_keys)6    return memory_data

默认 domain 和内置层可以走扁平布局;出现自定义 domain 或 custom layer 时,storage/file.py 会写成 memory/<layer>/<domain>/entries.json。这说明文件系统持久化是默认落点,向量索引不是事实的唯一来源。

python
1# assets/projects/soul-protocol/src/soul_protocol/runtime/storage/file.py2for domain, group in grouped.items():3    ddir = mem_dir / tier / domain4    ddir.mkdir(parents=True, exist_ok=True)5    (ddir / "entries.json").write_text(json.dumps(group, indent=2, default=str))

冲突怎么更新

冲突更新要解决的问题是:长期 agent 会持续听到“我搬家了”“我换工作了”“我现在不喜欢 X 了”。如果直接覆盖,历史解释能力会消失;如果什么都不更新,召回会把旧事实当真。soul-protocol 的选择是软替换:旧事实保留,但通过 superseded_by 退出默认召回。

第一层是去重。dedup.py 用 Jaccard 加 containment 系数判断新事实是 SKIPMERGE 还是 CREATE

python
1# assets/projects/soul-protocol/src/soul_protocol/runtime/memory/dedup.py2if best_similarity > 0.85 and best_match is not None:3    return ("SKIP", best_match.id)4 5if best_similarity >= 0.6 and best_match is not None:6    return ("MERGE", best_match.id)7 8return ("CREATE", None)

这段代码处在 semantic fact 写入之前。SKIP 表示新旧事实几乎重复,不再写;MERGE 表示新事实是旧事实的补充版,会让旧 fact 指向新 fact;CREATE 才真正新增。containment 的作用是避免 “User likes Python” 和 “User likes Python for backend automation” 因为后者更长而被误判成完全不同事实。

第二层是事实冲突。MemoryManager._resolve_fact_conflicts() 会按事实前缀找同类旧事实,例如 “User lives in ...” 和新的 “User lives in ...” 不同,就把旧 fact 的 superseded_by 指向新 fact。

python
1# assets/projects/soul-protocol/src/soul_protocol/runtime/memory/manager.py2def _find_conflict(self, new_content: str, existing_facts: list[MemoryEntry]) -> MemoryEntry | None:3    for prefix in _FACT_PREFIXES:4        if new_content.startswith(prefix):5            for fact in existing_facts:6                if fact.superseded_by is not None:7                    continue8                if fact.content.startswith(prefix) and fact.content != new_content:9                    return fact10    return None

这里的“前缀”是事实模板的简化版本,例如 User lives inUser works atUser prefers。它不是通用逻辑推理,而是对常见个人资料槽位做启发式更新。好处是便宜、可解释;边界是复杂时间事实和多实体事实仍可能需要更结构化的 schema。

第三层是 contradiction detector。它处理否定、实体属性冲突、以及 location/employer/role 这类动词事实冲突。尤其是 raw-text fallback 很关键:如果抽取器没有抽出新 fact,系统仍会拿原始用户输入去扫旧语义事实,捕捉 “I moved to Amsterdam” 这种更新。

python
1# assets/projects/soul-protocol/src/soul_protocol/runtime/memory/manager.py2raw_cresults = await self._contradiction_detector.detect_heuristic(3    raw_text, all_semantic_raw4)5for cr in raw_cresults:6    if not cr.is_contradiction or not cr.old_memory_id:7        continue8    for existing_fact in all_semantic_raw:9        if existing_fact.id == cr.old_memory_id:10            existing_fact.superseded = True11            existing_fact.superseded_by = "raw-text-contradiction"

这个 fallback 解决的是抽取失败后的旧事实污染。比如旧事实是 User lives in NYC,用户说 I moved to Amsterdam,但 fact extractor 没有成功产出标准 fact。raw-text scan 至少能把旧 location fact 标记过期,避免它继续被默认召回。

召回什么时候触发

召回是调用者显式触发的,不是 observe 后自动把记忆塞进下一轮 prompt。普通入口是 Soul.recall(),智能入口是 Soul.smart_recall()。前者负责过滤、候选生成、排序和 trace;后者先取更大的候选池,再根据配置选择是否 LLM rerank。

python
1# assets/projects/soul-protocol/src/soul_protocol/runtime/soul.py2results = await self._memory.recall(3    query=query,4    limit=limit,5    types=types,6    min_importance=min_importance,7    requester_id=requester_id,8    bond_strength=effective_bond,9    progressive=progressive,10    user_id=user_id,11    layer=layer,12    domain=domain,13)

smart_recall() 不是默认所有召回都会调用 LLM。源码里只有 enabledsettings.smart_recall_enabled 为真,并且存在 engine,且候选数超过 limit 时才 rerank。

python
1# assets/projects/soul-protocol/src/soul_protocol/runtime/soul.py2candidates = await self.recall(query, limit=candidate_pool)3effective_enabled = (4    enabled if enabled is not None else self._memory.settings.smart_recall_enabled5)6 7if effective_enabled and self._engine and len(candidates) > limit:8    results = await rerank_memories(candidates, query, self._engine, limit)9else:10    results = candidates[:limit]
召回链路:调用者显式请求 recall,MemoryManager 处理 user/domain/layer 过滤,RecallEngine 生成候选并按 activation 排序,smart_recall 再可选 LLM rerank。
召回链路:调用者显式请求 recall,MemoryManager 处理 user/domain/layer 过滤,RecallEngine 生成候选并按 activation 排序,smart_recall 再可选 LLM rerank。

怎么检索和排序

默认 RecallEngine 的 strategy 是 BM25SearchStrategy。这意味着默认候选生成更接近 lexical/BM25 search,而不是向量数据库查询。

python
1# assets/projects/soul-protocol/src/soul_protocol/runtime/memory/recall.py2class RecallEngine:3    def __init__(4        self,5        episodic,6        semantic,7        procedural,8        strategy=None,9    ):10        self._strategy = strategy if strategy is not None else BM25SearchStrategy()

store 层负责候选,activation 层负责排序。以 semantic store 为例,它会跳过已经 superseded_by 的旧事实,再按文本 relevance 生成候选。

python
1# assets/projects/soul-protocol/src/soul_protocol/runtime/memory/semantic.py2for fact in self._facts.values():3    if fact.superseded_by is not None:4        continue5    if fact.importance < min_importance:6        continue7    score = relevance_score(query, fact.content)8    if score > 0.0:9        scored.append((score, fact))

排序不是只看文本相似度。activation.py 把 base activation、spreading activation、emotion、significance、salience 和 personality modulation 合成最终分数。一个经常被访问、情绪强、显著性高的记忆,可能超过一条文本更相似但低价值的记忆。

python
1# assets/projects/soul-protocol/src/soul_protocol/runtime/memory/activation.py2activation = (3    (W_BASE * base)4    + (W_SPREAD * spread)5    + (W_EMOTION * emo)6    + sig_boost7    + salience_boost8    + personality_boost9)

MemoryManager.recall() 还会在有 user_idlayerdomain 过滤时扩大候选池。原因很实际:如果先按小 limit 取候选,再过滤 domain,结果可能被过滤到不够用。

python
1# assets/projects/soul-protocol/src/soul_protocol/runtime/memory/manager.py2any_filter = user_id is not None or layer is not None or domain is not None3fetch_limit = limit * 3 if any_filter else limit4 5results = await self._recall_engine.recall(6    query=query,7    limit=fetch_limit,8    types=engine_types,9    min_importance=min_importance,10)

向量能力到底是什么

VectorSearchStrategy 接收 embedding provider,对 query 和候选内容算 cosine similarity。它的输入仍然是 candidates,而不是从某个外部向量数据库读取全部事实。

python
1# assets/projects/soul-protocol/src/soul_protocol/runtime/embeddings/vector_strategy.py2class VectorSearchStrategy:3    def search(self, query: str, candidates: list[Any], limit: int = 10) -> list[Any]:4        query_vec = self._embedder.embed(query)5        for candidate in candidates:6            content = candidate.content if hasattr(candidate, "content") else str(candidate)7            vec = index_map.get(content) or self._embedder.embed(content)8            sim = cosine_similarity(query_vec, vec)

所以更准确的说法是:soul-protocol 把 vector search 放在检索策略扩展层,把文件/JSON 和 .soul 放在持久化层。vector storage、vector search、lexical/BM25 search、LLM rerank 是四个不同概念:默认存储不是向量库;默认检索策略是 BM25;向量策略可替换候选生成;LLM rerank 只在 smart recall opt-in 时参与最后重排。

检索策略

BM25SearchStrategy 和 VectorSearchStrategy 都是在候选列表上计算相关性。它们决定哪些 memory 可能相关。

持久化后端

MemoryManager.to_dict() 和 storage/file.py 决定 memory 如何落盘。默认是本地文件结构,不要求外部向量数据库。

端到端路径

把写入、存储、冲突和召回串起来,soul-protocol 的 memory lifecycle 是一条闭环,而不是一个单点功能。

1用户说出新信息

例如用户说 I moved to Amsterdam。

2observe 抽取事实

MemoryManager 尝试写 episodic,并抽取 User lives in Amsterdam。

3dedup/contradiction 更新旧事实

旧的 User lives in NYC 被 superseded_by 标记或 raw-text contradiction 标记。

4save/export 固化状态

MemoryManager.to_dict 输出 semantic、episodic、graph 等状态,storage 写入本地布局。

5recall 显式请求上下文

下一次 query 问用户所在地时,调用者显式调用 Soul.recall。

6activation 与可选 rerank 返回结果

旧事实被 semantic search 过滤,新事实按 activation 排序进入返回结果。

这个闭环的工程意义是:长期记忆的难点不只是“搜得准”,还包括写入时别把噪声永久化,更新时别让旧事实继续污染上下文,存储时别把运行态绑死在某个数据库,召回时还要考虑权限、关系、domain 和使用频率。

可复用设计

  • 把回复生成和学习后处理分开observe() 作为交互后的学习 hook,让主回复路径不必背负所有记忆成本。
  • 把记忆拆成不同生命周期:episodic 保存发生过什么,semantic 保存抽出的事实,procedural 保存做法,social 保存关系状态,core 保存常驻身份。
  • 事实更新用软替换superseded_by 让旧事实退出默认召回,但仍然可导出、审计和解释历史。
  • 召回排序不要只看相似度:ACT-R activation 把近期、频率、情绪、显著性、人格调制一起纳入,更符合长期 agent 的使用场景。
  • 昂贵增强做成 opt-in:向量策略和 LLM rerank 都存在,但默认路径不依赖它们,降低了最小可运行成本。

边界

这套实现的边界也很清楚。第一,事实冲突主要靠模板前缀、启发式 contradiction 和 raw-text fallback;复杂时间事实、多实体事实、计划/假设/历史状态仍然可能被抽得过粗。第二,默认本地文件持久化适合 portable soul 和本地库,但大规模多租户产品还需要外部数据库、索引和权限系统。第三,召回链路越丰富,调试一次“为什么没想起来”越需要可观测性;源码已经有 retrieval trace,但每条候选的过滤原因、分数组成和 rerank 决策还可以更细。

因此,这篇源码最值得带走的不是某个具体算法,而是一条设计原则:记忆系统要先定义生命周期,再选择检索技术。 向量搜索可以增强召回,但不能替代写入门控、冲突更新、持久化边界和可解释排序。