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 逐层筛选。
默认路径
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 的默认主链路。核心证据来自这些文件:
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 里有 core、episodic、semantic、procedural、social 五类,MemoryEntry 又补了 importance、somatic、significance、access_timestamps、superseded_by、abstract、salience、visibility、scope、user_id、layer、domain 等字段。
这些字段不是装饰。importance 决定基本重要性,significance 和 somatic 来自 observe 时的心理信号,access_timestamps 支撑“越常想起越容易被想起”,superseded_by 让事实更新变成软替换,visibility/scope/user_id/domain/layer 则把召回变成带权限和命名空间的检索。
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() 做分层处理。
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。
调用者把 Interaction 传给 Soul.observe。
MemoryManager 生成 somatic marker 和 significance score。
显著交互写入 EpisodicStore;低显著但抽到事实时也会被提升。
Cognitive processor 从输入输出里抽取 MemoryType.SEMANTIC。
dedup、prefix conflict 和 contradiction detector 标记旧事实。
没有被短路时继续抽实体、更新 graph 和 self_model。
写入什么东西
observe() 不是无条件把整轮聊天写进长期记忆。它先算情绪和显著性;只有显著交互才写 episodic。episodic 的内容是完整 user/agent 片段,所以它保留上下文和时间感。
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。这个分支很关键:系统不是“低显著就什么都不学”,而是把“有事实可学”当作晋升信号。
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._memories、SemanticStore._facts、ProceduralStore._procedures、SocialStore._entries,再加上 graph、self_model、archives 和 custom_layers。MemoryManager.to_dict() 把它们合成一个可持久化结构。
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 写到稳定布局。
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。这说明文件系统持久化是默认落点,向量索引不是事实的唯一来源。
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 系数判断新事实是 SKIP、MERGE 还是 CREATE。
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。
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 in、User works at、User prefers。它不是通用逻辑推理,而是对常见个人资料槽位做启发式更新。好处是便宜、可解释;边界是复杂时间事实和多实体事实仍可能需要更结构化的 schema。
第三层是 contradiction detector。它处理否定、实体属性冲突、以及 location/employer/role 这类动词事实冲突。尤其是 raw-text fallback 很关键:如果抽取器没有抽出新 fact,系统仍会拿原始用户输入去扫旧语义事实,捕捉 “I moved to Amsterdam” 这种更新。
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。
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。源码里只有 enabled 或 settings.smart_recall_enabled 为真,并且存在 engine,且候选数超过 limit 时才 rerank。
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]怎么检索和排序
默认 RecallEngine 的 strategy 是 BM25SearchStrategy。这意味着默认候选生成更接近 lexical/BM25 search,而不是向量数据库查询。
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 生成候选。
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 合成最终分数。一个经常被访问、情绪强、显著性高的记忆,可能超过一条文本更相似但低价值的记忆。
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_id、layer、domain 过滤时扩大候选池。原因很实际:如果先按小 limit 取候选,再过滤 domain,结果可能被过滤到不够用。
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,而不是从某个外部向量数据库读取全部事实。
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 是一条闭环,而不是一个单点功能。
例如用户说 I moved to Amsterdam。
MemoryManager 尝试写 episodic,并抽取 User lives in Amsterdam。
旧的 User lives in NYC 被 superseded_by 标记或 raw-text contradiction 标记。
MemoryManager.to_dict 输出 semantic、episodic、graph 等状态,storage 写入本地布局。
下一次 query 问用户所在地时,调用者显式调用 Soul.recall。
旧事实被 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 决策还可以更细。
因此,这篇源码最值得带走的不是某个具体算法,而是一条设计原则:记忆系统要先定义生命周期,再选择检索技术。 向量搜索可以增强召回,但不能替代写入门控、冲突更新、持久化边界和可解释排序。
