跳到主要内容

Memori — 架构与原理

30 秒导读: Memori 是给 AI agent 用的记忆中间件。你把已有的 LLM 客户端(OpenAI、 Anthropic、Gemini…)交给它 register 一次,之后你的代码一行不改——每次 LLM 调用前它 自动把"关于这个用户的相关记忆"塞进 prompt,调用后又在后台把这轮对话炼成结构化记忆 (事实、偏好、关系……)存起来。检索用稠密向量余弦 + BM25 关键词的混合重排,重活由一个 Rust 核心引擎干。记忆可以托管在 Memori Cloud,也可以落进你自带的数据库(Postgres / MySQL / SQLite / MongoDB / Oracle / TiDB / OceanBase)。

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

一句话定义: Memori 给"本来会失忆的 LLM 应用"装上长期记忆,而且是无侵入的—— 它劫持你的 LLM 客户端,把"记忆的读和写"变成调用的副作用。

解决什么问题 / 给谁用。 LLM 本身没有跨会话的记忆:上一轮对话结束,它就忘了你是谁、 喜欢什么。做 agent / 客服 bot / 编码助手的人,通常要自己写一套"存对话 + 检索历史 + 拼进 prompt"的管线。Memori 把这套管线做成了一个中间件,让你几乎不用碰它。

它能做什么:

  • 每次 LLM 调用前,自动召回与当前问题相关的记忆并注入 prompt。
  • 每次调用后,自动在后台把对话炼成结构化记忆(事实 / 偏好 / 关系 / 技能……)。
  • LLM 无关(OpenAI / Anthropic / Gemini / xAI / Bedrock / DeepSeek)、框架无关(Agno / LangChain / Pydantic AI)、数据库无关(Cloud 或自带 7 种数据库)。
  • 支持流式 / 非流式、同步 / 异步。

用起来什么样。 关键只有一行 register——注意下面没有任何"手动存记忆 / 手动查记忆" 的代码,全部自动发生:

# 示意,改编自 README 快速上手
from memori import Memori
from openai import OpenAI

client = OpenAI()
mem = Memori().llm.register(client)          # ← 这一步劫持了 client
mem.attribution(entity_id="user_123", process_id="support_agent")  # 记忆归属给谁

# 第一轮:说出一个事实
client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "My favorite color is blue."}],
)  # 后台悄悄把"喜欢蓝色"炼成记忆存下

# 第二轮:换个问题
client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "What's my favorite color?"}],
)  # 调用前自动召回"喜欢蓝色"并注入,于是模型答得出来

一句话直觉/类比。 把 Memori 想成 LLM 客户端和真正 API 之间的一层**"记忆代理"**: 出站时它往包裹里塞一张"关于你的小抄",入站时它把这轮对话读一遍、记下值得记的东西。 你(调用方)全程感觉不到它在。

一个关键前置条件: 必须先 attribution(...) 指定记忆归属的 entity(实体,像"用户") 和可选的 process(进程,像"你的 agent");不设置归属,Memori 就不会为你造记忆 (README「Attribution」;memori/__init__.py:188 attribution)。

本节不谈底层。记住一件事:register 一次,读写记忆变成 LLM 调用的自动副作用。

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

先看一次 LLM 调用在 Memori 里的完整生命周期。这张图从上到下是时间顺序,左边是你的 调用,右边是记忆系统在背后做的两件事(召回在调用前、增强/写入在调用后)。

        你的代码                         Memori 中间件                       记忆后端
   client...create(messages)
            │
            ▼
   ┌─────────────────────┐   ① 调用前:召回并注入
   │  被劫持的 create()   │──────────────┐
   │  (Invoke.invoke)    │              ▼
   └─────────────────────┘     inject_recalled_facts
            │                  · 提取用户问题
            │                  · 检索相关记忆 ───────────►  Cloud API / Rust 核心 / DB
            │                  · 过阈值、拼成 <memori_context>
            │                  · 塞进 system / messages
            │  (prompt 已被悄悄改写)
            ▼
   ┌─────────────────────┐
   │  真正的 LLM API     │  ← 原始 create,拿到回答
   └─────────────────────┘
            │
            ▼
   ┌─────────────────────┐   ② 调用后:炼记忆并写入(后台,不加延迟)
   │  handle_post_response│──────────────┐
   └─────────────────────┘              ▼
            │                  handle_augmentation
            │                  · 对话打包 ───────────────►  后台:调"增强"服务
   把原始回答原样返回给你      · LLM 抽出 facts/关系/偏好…      → 结构化记忆写入后端
            ▼
       response(和没装 Memori 时一模一样)

怎么读这张图: ① 和 ② 都是自动副作用;你拿到的 response 和不装 Memori 时完全一样, Memori 只是"顺路"读了记忆、写了记忆。召回同步做(会稍微改 prompt),增强异步做(几乎零延迟)。

主要部件一句话职责:

部件干什么在哪(相对克隆根)
Memori 入口SDK 门面:配置、attribution、session、recall;决定走 Cloud 还是 BYODBmemori/__init__.py
LLM 劫持层register 把客户端的 create/parse/… 换成 Invoke().invoke 包装memori/llm/clients/direct.py、memori/llm/_base.py
Invoke 管线每次调用的编排:先注入召回、调真方法、再触发增强memori/llm/invoke/invoke.py
召回注入提取问题→检索记忆→过阈值→拼 <memori_context>→塞进 promptmemori/llm/pipelines/recall_injection.py
增强(写入)把对话交给"增强"服务炼成结构化记忆,后台落库memori/memory/augmentation/_handler.py
检索核心稠密余弦选候选 + BM25 词法打分 + 加权重排memori/search/_core.py、memori/search/_lexical.py
Rust 核心同步嵌入 + 两个有界后台 worker + 检索管线,跨 Py/Node 复用core/src/lib.rs
存储层Cloud 走 HTTP;BYODB 走多方言驱动(7 种库)memori/storage/、core/src/storage/

主线走一遍(高层): 你调 create() → Invoke 先跑 inject_recalled_facts(把相关记忆 拼进 prompt)→ 调真正 LLM 拿回答 → handle_post_response 把这轮对话丢给增强流程 → 增强服务(远端 API)抽出结构化记忆 → 写进后端(Cloud 或你的数据库)。下次提问,新记忆就 能被召回。

3. 阅读地图(建议顺序)​

这是一个多子系统项目,按"由浅入深"分成 5 章。建议顺序:

  1. 拦截与自动捕获:register 如何劫持 LLM 调用 — 一切的起点。 register 怎么把客户端方法换成包装器、Invoke.invoke 的三段式编排、同步/异步/流式怎么分派。
  2. 记忆生成:Advanced Augmentation 如何把对话炼成结构化记忆 — 写路径。 一轮对话如何变成 facts / 语义三元组 / 进程属性 / 会话摘要,以及"后台不加延迟"是怎么做到的。
  3. 回忆与检索:dense 余弦 + BM25 混合重排 — 读路径。 候选召回、余弦相似度、BM25 词法分、短查询偏词法的加权公式、阈值过滤与注入格式。
  4. Rust 核心引擎:同步嵌入 + 有界后台 worker + 跨语言绑定 — 底层引擎。 EngineOrchestrator、有界队列 + 限并发的 WorkerRuntime、PyO3/napi 绑定、回调式存储桥。
  5. 存储层与两条路:Cloud vs BYODB、StorageBridge 与多方言驱动 — 落地。 Cloud 与自带数据库两条路怎么分叉、StorageBridge 抽象、7 种方言驱动与 4 种适配器。

想快速判断相关性:改 prompt 注入 → 看第 3 章;造记忆/写库 → 看第 2、5 章; 性能/并发/嵌入 → 看第 4 章;接入新 LLM/框架 → 看第 1 章。

4. 巧妙之处(带走的精华)​

① 无侵入靠"方法替换"(monkeypatch),不是继承也不是代理对象。 register 直接把 client.chat.completions.create 这样的方法替换成包装器的 .invoke,并把原方法备份到 client.chat._completions_create;还用 client._memori_installed 做幂等,重复 register 不会 叠加(memori/llm/clients/direct.py:155-207 OpenAi.register;memori/llm/_base.py:35-84 _wrap_method)。因此调用方代码零改动。

② 一个入口点,三段式编排。 无论同步/异步/流式,核心都是 inject_conversation_messages(inject_recalled_facts(...)) → 调真方法 → handle_post_response 三步,只是按响应类型分派到不同迭代器(memori/llm/invoke/invoke.py:25-65 Invoke.invoke)。

③ 召回按 provider 注入到"对的地方"。 同一份 <memori_context>,对 Anthropic/Bedrock 拼到 system,对 Google 拼到 system instruction,对 OpenAI Responses 拼到 instructions,否则拼到 messages 的 system 消息——一段逻辑吃下所有家族的 prompt 形态 (memori/llm/pipelines/recall_injection.py:201-224)。

④ 混合重排,且"短查询更信关键词"。 检索先用稠密余弦选候选池,再用 BM25 词法分,按 w_cos·cos + w_lex·lex 加权;查询 ≤2 个词时把词法权重从 0.15 提到 0.30,因为极短查询里 精确词命中是高信号(memori/search/_core.py:116-150 _rank_candidates; memori/search/_lexical.py:127-148 dense_lexical_weights)。

⑤ 写记忆"零延迟"靠有界后台 worker。 增强(调远端服务 + 嵌入 + 落库)全部丢进后台执行; Rust 侧是两个**有界队列(容量 512)、限并发(2)**的 WorkerRuntime,队列满就明确报 QueueFull 而不是无限堆积(core/src/lib.rs:244-285 init_*_runtime; memori/memory/augmentation/_handler.py:157-201)。

⑥ Rust 引擎不碰数据库,靠回调"反向调用"宿主。 Rust 核心通过 StorageBridge trait 拿数据, Python 侧把 fetch_embeddings / fetch_facts_by_ids / write_batch 三个回调注入进去——于是同一个 Rust 引擎能配任意 Python DB 驱动(core/src/storage/bridge.rs:6-30; memori/native/_adapter.py:215-297)。

⑦ 增强是"数据驱动"的 WriteBatch。 增强服务返回的 JSON 里有 facts 就发 entity_fact.create,有 semantic_triples 就发 knowledge_graph.create,有 attributes、 summary 就分别发对应 op——一份响应拆成若干写操作,后端只认 op 类型 (core/src/augmentation/pipeline.rs:100-165 build_write_batch_from_response)。

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

用符号名跳转比行号更抗漂移;下表每行是"读这个项目该打开的关键锚点"。

主题文件路径(相对克隆根)符号名
SDK 入口 / Cloud vs BYODB 分叉memori/__init__.pyMemori.__init__、_get_default_connection
归属 / 会话 / 直接召回memori/__init__.pyMemori.attribution、new_session、Memori.recall
注册总入口(直连 vs 框架)memori/llm/_registry.pyregister_llm、Registry.client
方法替换 / 幂等劫持memori/llm/clients/direct.pyOpenAi.register、Anthropic.register、Google.register
包装辅助(备份 + async 检测)memori/llm/_base.pyBaseClient._wrap_method
单次调用三段式编排memori/llm/invoke/invoke.pyInvoke.invoke、InvokeAsync、InvokeStream
召回注入(过阈值 + 按 provider 注入)memori/llm/pipelines/recall_injection.pyinject_recalled_facts、format_recalled_fact_lines
调用后触发写入memori/llm/pipelines/post_invoke.pyhandle_post_response、format_payload
增强分派(Cloud / Rust / Python)memori/memory/augmentation/_handler.pyhandle_augmentation、_submit_rust_augmentation_background
检索核心(候选池 + 重排)memori/search/_core.pysearch_entity_facts_core、_rank_candidates
BM25 词法分 + 权重memori/search/_lexical.pylexical_scores_for_ids、dense_lexical_weights
Rust 引擎顶层core/src/lib.rsEngineOrchestrator、retrieve、submit_augmentation
有界后台 workercore/src/runtime/worker.rsWorkerRuntime、submit、flush
Rust 检索管线core/src/retrieval/pipeline.rsrun_retrieval、dynamic_candidate_limit
余弦相似度 + top-Kcore/src/search/similarity.rscosine_similarity、find_similar_embeddings
增强 → WriteBatchcore/src/augmentation/pipeline.rsrun_advanced_augmentation、build_write_batch_from_response
存储桥抽象core/src/storage/bridge.rsStorageBridge
Rust↔Python 回调适配memori/native/_adapter.pyRustCoreAdapter、_fetch_embeddings_cb、_write_batch_cb
Python 存储管理 / 方言memori/storage/_manager.pyManager.start、Registry

跨语言与多后端一览(结构关系,不是表数据):

memori (Python SDK)  ─┐
                      ├──►  Rust 核心 engine-orchestrator  ──►  StorageBridge 回调
@memorilabs/memori    ─┘         (fastembed 嵌入 +               │
   (Node SDK, memori-ts/)         2×有界 WorkerRuntime +         ▼
                                  余弦+BM25 检索)          你的 DB 驱动:
                                                          postgresql / mysql / sqlite /
                                                          mongodb / oracle / tidb / oceanbase
    (或者不走 Rust/DB,直接走 Cloud:HTTP → Memori Cloud API)

说明:Rust 核心只在 BYODB 且启用 rust core 时创建(memori/native/_adapter.py:52-66 RustCoreAdapter.maybe_create);默认 Cloud 模式下,召回/增强走 HTTP 到 Memori Cloud,不启用 本地引擎(memori/__init__.py:169-186)。两条路的完整分叉见 第 5 章。