跳到主要内容

数据截至 (上游 commit 85532420387c)

图谱式明文记忆(下):检索召回、重排与推理

30 秒导读: 上一章讲了记忆怎么"写进"图谱(04)。这一章讲反向的一半: 一条用户 query 进来,系统怎么把散落在图里的相关记忆找回来、排好序、去掉重复,最后交给对话模型。 主角是一个叫 Searcher 的类,它把检索拆成"解析 → 多路并行召回 → 多源融合 → 重排 → 排序裁剪"五步流水线。

本章只讲读取一侧。写入(抽取、组织、去重、冲突消解)见 03 和 04; 记忆的形态与容器见 01;上层内核 MOSCore 见 02。

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

一句话定义: 检索管线 = 给定一句话,从记忆图谱里"捞出"最相关的一批记忆条目,排好序返回。

它解决什么问题。 记忆库越写越大,几千上万条记忆散落在图谱的节点里。对话时不可能把全部记忆塞进 上下文窗口——太贵、也会淹没重点。所以每轮对话前要做一次精准的召回:只取"和这次问题相关"的十几条。

一个直觉类比。 把它想成图书馆找书:

  • 你只说一句"我上次说的那个项目截止日期"(query,模糊)。
  • 图书馆员先听懂你要什么(任务解析:关键词=项目、截止日期)。
  • 然后兵分几路同时找:按卡片目录找(图谱结构)、按"意思相近"找(向量)、按字面词找(BM25/全文)。
  • 各路把候选堆到一起,去掉重复,按相关度排好序,只把最上面几本递给你(重排 + 裁剪)。

用起来什么样。 上层调用极简——MOSCore.search("我周五的会议改到几点了?"),内部就跑完整条管线:

# 示意,非源码:一次检索的对外观感
results = mos.search(
    query="我周五的会议改到几点了?",
    top_k=10,
    mode="fine",              # fine=慢而准(用大模型解析),fast=快而糙
    internet_search=False,    # 是否允许联网补充
)
# results 是一批 TextualMemoryItem,已按相关度排好序

为什么要"多路"而不是只用向量? 因为单一召回各有盲区:向量擅长"意思相近"但对精确的专有名词 (人名、编号)不敏感;字面匹配(BM25/全文)擅长专名但抓不住语义;图谱结构召回擅长"同一主题/标签 的邻居"但依赖抽取时打好的标签。并行跑、再融合,才能既召得全又召得准。

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

怎么读下面这张图: 从上到下是一次 search() 的时间顺序;中间那层"六路召回"是并行的(同时发起), 其余步骤串行。命中即汇入同一个候选池。

用户 query
   │
   ▼
┌─────────────────────────────────────────────┐
│ ① 任务解析  _parse_task / TaskGoalParser       │  query → 关键词 keys / 标签 tags /
│    (fast=分词; fine=大模型拆解)                 │         改写 query / 是否联网 / 向量 embedding
└─────────────────────────────────────────────┘
   │  parsed_goal + query_embedding
   ▼
┌─────────────────────────────────────────────┐
│ ② 多路并行召回  _retrieve_paths (线程池)         │
│   ├ A 工作记忆   _retrieve_from_working_memory  │
│   ├ B 长期+用户  _retrieve_from_long_term...    │  每一路内部又并行跑:
│   ├ C 联网       _retrieve_from_internet        │    图谱召回 / 向量召回 /
│   ├ (可选)关键词 _retrieve_from_keyword         │    BM25召回 / 全文召回
│   ├ (可选)工具   _retrieve_from_tool_memory     │  (GraphMemoryRetriever.retrieve)
│   ├ (可选)技能   _retrieve_from_skill_memory    │
│   └ (可选)偏好   _retrieve_from_preference...   │
└─────────────────────────────────────────────┘
   │  每一路各自 _maybe_rerank 后汇总成一个大 list
   ▼
┌─────────────────────────────────────────────┐
│ ③ 后处理  post_retrieve                        │
│   ├ 去重  _deduplicate_results (按记忆文本)     │
│   └ 排序裁剪 _sort_and_trim (按分数, 分类型 top_k)│
└─────────────────────────────────────────────┘
   │
   ▼
最终 list[TextualMemoryItem](已排好序)

部件一句话职责:

部件干什么在哪个文件
Searcher检索总指挥,编排整条管线retrieve/searcher.py:46
TaskGoalParser把 query 解析成关键词/标签/改写句/是否联网retrieve/task_goal_parser.py:18
GraphMemoryRetriever单一 scope 内的四路底层召回retrieve/recall.py:61
BaseReranker(BGE/cosine)对候选打相关度分、排序reranker/base.py:12
MemoryReasoner用大模型二次筛选/合成(可选深链路)retrieve/reasoner.py:11
AdvancedSearcherSearcher 子类,多阶段"深检索"retrieve/advanced_searcher.py:25
联网检索器把网页搜索结果转成记忆条目retrieve/xinyusearch.py 等

一个容易混淆的点:实际用的搜索器是子类。 tree.py 里是 from ... import AdvancedSearcher as Searcher (tree.py:25),所以生产实例是 AdvancedSearcher;但普通 .search() 走的仍是父类 Searcher.search 的流水线,AdvancedSearcher 只是额外提供了 deep_search(见 §6.2)。

主线走一遍(高层,不进代码): MOSCore.search 遍历用户可访问的每个 Cube,对每个 Cube 的 text_mem.search() 发起检索(mem_os/core.py:618)→ TreeTextMemory.search new 一个搜索器并调其 .search(tree.py:214)→ 进入本章的五步流水线 → 各 Cube 结果汇总回 MOSCore。

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

3.1 任务解析:把模糊的 query 变成可检索的目标​

要解决的小问题。 用户说的话是自然语言,而底层召回需要结构化的抓手:该用哪些关键词做字面匹配? 哪些标签做图谱过滤?要不要联网?原句要不要改写得更适合检索?

思路。 用 TaskGoalParser 产出一个 ParsedTaskGoal。它有两档:

  • fast 模式——不调大模型,几乎"原样透传"。快,适合高并发。
  • fine 模式——调一次大模型,把 query 拆成结构化的 keys/tags/memories,还判断 internet_search 并给出 rephrased_query(改写句)。慢但准。

真实实现。 入口 parse 按 mode 分流(task_goal_parser.py:30):

# task_goal_parser.py:46 —— parse() 的分流
if mode == "fast":
    return self._parse_fast(task_description, context=context, **kwargs)
elif mode == "fine":
    return self._parse_fine(task_description, context, conversation, **kwargs)

fast 模式默认连分词都不做,keys/tags 直接留空、rephrased_query 就是原句(_parse_fast, task_goal_parser.py:55);只有开了 use_fast_graph 才用 tokenizer.tokenize_mixed 分词填充。 fine 模式把 query + 上下文 + 历史对话套进 TASK_PARSE_PROMPT 交给大模型,再把 JSON 解析成 ParsedTaskGoal(_parse_fine / _parse_response,task_goal_parser.py:83 / :107)。

关键细节:解析结果如何回流到检索。 Searcher._parse_task(searcher.py:283)拿到 parsed_goal 后: 用 rephrased_query 覆盖原 query;若 parsed_goal.memories 非空,就把 query 和这些 memories 一起 embed,得到一批 query_embedding(不是一个向量,是多个,后续向量召回会各走一遍)。

# searcher.py:353 —— 改写句覆盖 + 多向量嵌入
query = parsed_goal.rephrased_query or query
if parsed_goal.memories:
    embed_texts = list(dict.fromkeys([query, *parsed_goal.memories]))
    query_embedding = self.embedder.embed(embed_texts)   # 返回 list[list[float]]

3.2 单 scope 内的四路召回:GraphMemoryRetriever​

要解决的小问题。 给定一个记忆范围(如 LongTermMemory)和解析好的目标,怎么从图数据库里 把候选节点捞出来?

思路:一个 scope 内也不止一种召回。 GraphMemoryRetriever.retrieve(recall.py:81)在同一个 scope 里 并行跑最多四种召回,最后按节点 id 合并去重:

GraphMemoryRetriever.retrieve(scope=LongTermMemory)
   ├─ _graph_recall     结构召回:按 keys 精确匹配 / tags 重叠≥2
   ├─ _vector_recall    向量召回:query_embedding 逐个做相似度检索
   ├─ _bm25_recall      (若配了 bm25) 字面 BM25 打分
   └─ _fulltext_recall  (若 use_fast_graph) 图库全文索引
        │
        ▼   combined = {item.id: item}  ← 按 id 去重合并
   list[TextualMemoryItem]

特例:工作记忆走捷径。 WorkingMemory scope 不做上面四路,直接 get_all_memory_items 全量取回、 截断 top_k(recall.py:122-131)——因为工作记忆本来就小、时效强,全取即可。

四路各自在干嘛:

召回路匹配依据关键行
_graph_recallkey ∈ parsed_goal.keys,或 tags 重叠 ≥ 2recall.py:242
_vector_recall每个 query 向量做 search_by_embedding,按分数并集去重recall.py:370
_bm25_recall先按 scope 元数据取候选节点,再 BM25Okapi 打分recall.py:534
_fulltext_recall图库原生全文索引 search_by_fulltextrecall.py:572

图谱召回的巧妙约束——"标签至少重叠 2 个"。 只要有一个标签相同就召回,噪声太大;所以要求 tags 交集 ≥ 2 才保留,把"沾边"过滤成"确有共性":

# recall.py:265 —— 结构召回的保留判据
if parsed_goal.keys and node_key in parsed_goal.keys:
    keep = True          # key 精确命中
elif parsed_goal.tags:
    overlap = len(set(node_tags) & set(parsed_goal.tags))
    if overlap >= 2:     # 标签至少重叠 2 个才算相关
        keep = True

向量召回的"双路"设计。 _vector_recall 内部还分 path A(无优先级)和 path B(带 search_priority 偏好过滤)并发跑,再把两路命中按"同 id 取高分"合并(recall.py:419-467)——让"优先级偏好"能加成 但不至于漏掉普通命中。

3.3 多路召回的编排:_retrieve_paths​

要解决的小问题。 §3.2 是"一个 scope 内"的召回。但一次检索要覆盖多种记忆类型 + 多种来源 (工作/长期/用户/工具/技能/偏好/联网),这些怎么组织?

思路:用一个线程池同时发起 A~F 六条路径。 Searcher._retrieve_paths(searcher.py:349)按开关把 若干条路径提交进 ContextThreadPoolExecutor,全部并行,最后 t.result() 汇总:

路径方法触发条件
A 工作记忆_retrieve_from_working_memory总是(scope 匹配)
B 长期+用户_retrieve_from_long_term_and_user总是
C 联网_retrieve_from_internet有 retriever 且 parsed_goal.internet_search
关键词_retrieve_from_keyworduse_fulltext 开
D 工具_retrieve_from_tool_memorysearch_tool_memory
E 技能_retrieve_from_skill_memoryinclude_skill_memory
F 偏好_retrieve_from_preference_memoryinclude_preference_memory

共同套路。 每条路径都是"调 GraphMemoryRetriever.retrieve 拿候选 → 立刻 _maybe_rerank"。 路径 B 的长期记忆和用户记忆还各自并发(searcher.py:761),并对结果做两道清洗: _deduplicate_rawfile_results(删掉被 RawFile 指向的 summary 节点,searcher.py:1282)和 _filter_intermediate_content(滤掉含 "File URL:/File ID:/Filename:" 的中间产物,searcher.py:1330)。

每一路都可能带 CoT 增强(见 §3.5)。 路径 B/D/E/F 若开了 vec_cot,会先把 query 拆成子问题、 各自嵌入,和原始 query 向量拼在一起再送去向量召回,扩大召回面。

3.4 关键词召回:_retrieve_from_keyword 的加权抽词​

要解决的小问题。 全文/关键词检索需要"从一句话里挑出最该用来匹配的几个词",挑多了引噪声, 挑少了漏关键。

思路:按语言 + 长度动态决定抽几个词,并给英文词打分排序。核心是 _extract_weighted_keyword_terms(searcher.py:603):

  • 先 detect_lang,再 _keyword_extract_top_k 按 query 长短决定抽 1~3 个词(searcher.py:565)。
  • 中文用 jieba.analyse.extract_tags(带词性白名单 KEYWORD_ALLOW_POS)。
  • 英文用 _rank_english_keyword_terms:按"出现次数×3 + 长度权重 + 含数字加成"打分(searcher.py:590)。
  • 全程用停用词表 StopwordManager 过滤、去重。

安全细节:防注入。 抽出的词在拼进 to_tsquery 前会被引号包裹并转义单引号,避免用户输入里的 操作符被当成查询语法(searcher.py:659)。

3.5 CoT 查询增强:把复杂问题拆成子问题​

要解决的小问题。 "我上个月在杭州出差时定的那家酒店叫什么、离会场多远?"——一句里其实塞了两三个 子问题,单向量召回抓不全。

思路。 _cot_query(searcher.py:1379)用大模型判断 query 是否"复杂";复杂就拆成 sub_questions (最多 split_num=3 个),对每个子问题各嵌入一遍,和原向量一起送召回。

# searcher.py:1412 —— CoT 拆解的核心判断
assert "is_complex" in response_json
if not response_json["is_complex"]:
    return [query]                       # 简单问题,不拆
else:
    return response_json["sub_questions"][:split_num]   # 复杂,拆成子问题

接线处。 在路径 B 里(searcher.py:753):vec_cot 开时先 _cot_query 得到子问题、embed 成 cot_embeddings,再 extend(query_embedding) 把原向量并进去,一起喂 graph_retriever.retrieve。 拆解失败会 except 兜底成 [query],不影响主流程。

3.6 重排:给候选打相关度分​

要解决的小问题。 六路召回堆出一大池候选,彼此分数不可比(向量分、BM25 分、全文分口径都不同), 必须用统一标尺重新打分。

思路:抽象出 BaseReranker 接口,多实现可插拔。 Searcher._maybe_rerank(searcher.py:80)是唯一入口, reranker 为空或关闭时退化为"原序 + 0 分":

# searcher.py:89 —— 重排开关的兜底
if not enabled or self.reranker is None:
    return [(item, 0.0) for item in graph_results[:top_k]]
return self.reranker.rerank(query=query, graph_results=graph_results, top_k=top_k, **kwargs)

三种生产实现(reranker/factory.py:26 按 backend 选):

backend实现打分方式
cosine_localCosineLocalReranker本地算 query 向量与候选 embedding 的余弦,乘层级权重
http_bgeHTTPBGEReranker把 (query, 文档) 发给远端 BGE cross-encoder 打分
http_bge_strategyHTTPBGERerankerStrategy同上,但文档组装交给"策略"(见下)
noopNoopReranker不重排

本地余弦重排的细节。 CosineLocalReranker.rerank(reranker/cosine_local.py:63):先滤掉没 embedding 的候选;没向量就退化给 0.5 分;有向量就算余弦相似度,再乘 level_weights(topic/concept/fact 结构层级权重), 排序取 top_k,不足则用 -1.0 分的剩余项补齐。

注:retrieve/reranker.py 里还有一个 MemoryReranker(reranker.py:39),逻辑与 CosineLocalReranker 几乎一致(余弦 + 层级权重),是同一思路的模块内版本;生产链路上 Searcher.reranker 由 RerankerFactory 注入,走 reranker/ 包这套。

BGE 远端重排的两个巧思:

  • 降级永不炸。 HTTPBGEReranker.rerank 用 @timed_with_status(..., fallback=...) 兜底:HTTP 超时/异常时 返回"前 top_k 候选 + 0 分",检索不因重排服务挂掉而失败(reranker/http_bge.py:126)。
  • 偏好加成。 _apply_boost_generic(http_bge.py:287)按 search_filter 命中项给分数乘 (1+weight) (默认 user_id:0.5 / tags:0.2 / session_id:0.3),让"本人/本会话/同标签"的记忆浮上来,并 clip 到 [0,1]。

重排策略(strategy)。 http_bge_strategy 把"候选如何拼成待打分文档"抽成 RerankerStrategyFactory (reranker/strategies/factory.py)可选:single_turn(把对话 source 两两拼成 user/assistant 对)、 concat_background、concat_docsource、singleturn_outmem。这样同一个 BGE 服务能适配不同记忆形态。

4. 深入实现:后处理 post_retrieve​

六路召回各自重排后,retrieve 返回的是一个混杂的 (item, score) 大列表——同一条记忆可能被多路 召回、多种类型混在一起。post_retrieve(searcher.py:157)做最后两步收敛。

第一步:去重。 _deduplicate_results(searcher.py:1149)以记忆文本为键,同文本只留最高分那条:

# searcher.py:1157 —— 按记忆文本去重,保留最高分
for item, score in results:
    if item.memory not in deduped or score > deduped[item.memory][1]:
        deduped[item.memory] = (item, score)

(可通过 dedup="no" 跳过去重。)

第二步:分类型排序裁剪。 _sort_and_trim(searcher.py:1157)不是简单"全体排序取 top_k",而是 按记忆类型分桶,各桶用各自的 top_k:

  • 工具记忆(ToolSchemaMemory / ToolTrajectoryMemory)各取 tool_mem_top_k;
  • 技能记忆取 skill_mem_top_k;偏好记忆取 pref_mem_top_k;
  • 其余"文本型"(Working/LongTerm/User/Outer/RawFile)合在一起排序取 top_k。

每条结果会把分数写进 metadata.relativity,并重建成带 SearchedTreeNodeTextualMemoryMetadata 的 TextualMemoryItem 返回(searcher.py:1271)。plugin 模式下 0 分项会被丢弃。

为什么要分桶? 因为工具/技能/偏好和普通事实记忆不该抢同一个 top_k 名额——它们服务不同用途, 各留固定配额,才能保证每类都进得了最终上下文。

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

  • 并行是一等公民。 从"六路径并行"(searcher.py:383)到"单 scope 内四路并行"(recall.py:133) 再到"向量召回 A/B 双路并行"(recall.py:447),层层用线程池铺开——检索延迟被压到"最慢那一路"。

  • 多向量而非单向量。 解析产出的多条 memories + CoT 子问题都各自嵌入,向量召回逐个走一遍再并集去重 (searcher.py:349、_cot_query)。用"多把钥匙开锁"换召回率。

  • 标签重叠阈值 ≥ 2。 把"沾一个标签就算相关"收紧成"至少两个共性",廉价地压噪声(recall.py:311)。

  • 重排服务失败即降级,不阻断主流程。 BGE 远端超时就退化为原序 0 分(http_bge.py:126), 可用性优先于精排。

  • 偏好加成 + 上限裁剪。 命中 user/session/tag 时分数乘性加成再 clip 到 [0,1],既提权又不越界 (_apply_boost_generic,http_bge.py:287)。

  • 分类型配额。 工具/技能/偏好各留独立 top_k,避免不同用途的记忆互相挤占(_sort_and_trim)。

  • 联网结果被"记忆化"。 网页搜索结果不是直接拼字符串,而是转成 TextualMemoryItem (xinyusearch.py:138 的 retrieve_from_internet),从而能和本地记忆走同一套重排/去重/排序。

6. 两条增强链路​

6.1 推理增强:MemoryReasoner​

它做什么。 在拿到排好序的记忆后,再用大模型做一次"读题选材":把候选记忆列成 [id] key: memory 喂给大模型,让它挑出真正该用的那几条 id,只返回被选中的记忆 (reasoner.py:20 的 reason)。选中 id 的解析支持 JSON 的 selected_ids,也支持从文本里 正则抓 UUID 兜底(_parse_selected_ids,reasoner.py:50)。

这是"召回负责多、推理负责精"的分工——召回宁滥勿缺,推理再把噪声筛掉。

它做什么。 面向"一次召回答不全"的复杂问题,做多轮迭代检索(advanced_searcher.py:232):

deep_search
  1. 先 retrieve + post_retrieve 拿一批记忆
  2. 循环 thinking_stages 轮:
       stage_retrieve → 大模型判断"够不够回答?"
         ├ 够 → 结束,(必要时)重写增强后返回
         └ 不够 → 吐出新的 retrieval_phrases,逐个再 retrieve,
                  合并 → memory_recreate_enhancement 重写记忆 → 下一轮
  3. 最后一轮 judge_memories 终判,返回增强后的记忆

每轮用 stage{N}_expand_retrieve 提示词让模型给出"还差什么、该再搜哪些短语",带重试与异常兜底 (stage_retrieve,advanced_searcher.py:71)。这是"检索—反思—再检索"的 agentic 检索范式。

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

  • fast 模式召回偏字面。 fast 下 keys/tags 默认为空,图谱结构召回基本失效,主要靠向量; 想要结构召回得开 fine 或 use_fast_graph。

  • 标签重叠 ≥ 2 的双刃。 抽取阶段标签打得稀疏的记忆,可能因为凑不满 2 个交集而被结构召回漏掉 (recall.py:311)——召回质量强依赖上游 04 的标签质量。

  • 关键词/全文路径依赖特定后端。 _retrieve_from_keyword 需要 user_name 且面向 PolarDB 全文 (_require_keyword_user_name,searcher.py:543),并非所有部署都启用。

  • _update_usage_history 当前是空实现。 使用历史回写的逻辑整段被注释掉(searcher.py:1342), 即代码里看不到"检索会更新记忆热度"这一行为已生效。

  • deep_search 成本高。 多轮大模型调用 + 反复检索,延迟和 token 开销显著,只适合离线/高价值查询。

8. 横向对比​

同货架的记忆系统里,本管线的取舍偏"多路召回 + 可插拔重排 + 分类型配额"。相较仅靠单一向量库 (embedding 一把梭)的做法,MemOS 用并行的图谱/向量/字面多路来对冲各自盲区,代价是编排复杂、 延迟依赖最慢路径。写入侧的组织质量(04)直接决定这里结构召回的上限; 上层由 02 的 MOSCore.search 跨 Cube 汇总。

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

主题文件路径符号
检索总编排src/memos/memories/textual/tree_text_memory/retrieve/searcher.pySearcher
对外入口同上Searcher.search
任务解析接线同上Searcher._parse_task
召回主流程同上Searcher.retrieve
六路并行编排同上Searcher._retrieve_paths
路径A 工作记忆同上Searcher._retrieve_from_working_memory
路径B 长期+用户同上Searcher._retrieve_from_long_term_and_user
路径C 联网同上Searcher._retrieve_from_internet
关键词/全文路径同上Searcher._retrieve_from_keyword
加权抽词同上Searcher._extract_weighted_keyword_terms
路径D 工具同上Searcher._retrieve_from_tool_memory
路径E 技能同上Searcher._retrieve_from_skill_memory
路径F 偏好同上Searcher._retrieve_from_preference_memory
多 Cube 召回同上Searcher._retrieve_from_memcubes
CoT 查询拆解同上Searcher._cot_query
重排开关同上Searcher._maybe_rerank
后处理同上Searcher.post_retrieve
去重同上Searcher._deduplicate_results
排序裁剪(分桶)同上Searcher._sort_and_trim
任务目标解析器retrieve/task_goal_parser.pyTaskGoalParser / _parse_fast / _parse_fine
单 scope 四路召回retrieve/recall.pyGraphMemoryRetriever.retrieve
结构召回同上GraphMemoryRetriever._graph_recall
向量召回同上GraphMemoryRetriever._vector_recall
BM25 召回同上GraphMemoryRetriever._bm25_recall
全文召回同上GraphMemoryRetriever._fulltext_recall
BM25 工具retrieve/bm25_util.pyEnhancedBM25
重排接口src/memos/reranker/base.pyBaseReranker
本地余弦重排src/memos/reranker/cosine_local.pyCosineLocalReranker
BGE 远端重排src/memos/reranker/http_bge.pyHTTPBGEReranker / _apply_boost_generic
BGE 策略版重排src/memos/reranker/http_bge_strategy.pyHTTPBGERerankerStrategy
重排策略工厂src/memos/reranker/strategies/factory.pyRerankerStrategyFactory
重排工厂src/memos/reranker/factory.pyRerankerFactory.from_config
模块内重排retrieve/reranker.pyMemoryReranker / batch_cosine_similarity
推理增强retrieve/reasoner.pyMemoryReasoner.reason
深检索retrieve/advanced_searcher.pyAdvancedSearcher.deep_search / stage_retrieve
联网检索(信息)retrieve/xinyusearch.pyXinyuSearchRetriever.retrieve_from_internet
联网检索工厂retrieve/internet_retriever_factory.pyInternetRetrieverFactory.from_config
Cube 层调用src/memos/memories/textual/tree.pyTreeTextMemory.search
内核层调用src/memos/mem_os/core.pyMOSCore.search(text_mem.search)