数据截至 2026-07-28(上游 commit 538b61f24529)

Rust 核心引擎:同步嵌入 + 有界后台 worker + 跨语言绑定

30 秒导读: Memori 把最重、最要命的两件事——把文本变成向量(嵌入) 和 在后台异步跑记忆生成/落库——都下沉到一个 Rust crate engine-orchestrator。Python SDK、Node SDK、TypeScript SDK 三家共用同一份引擎,各自只写一层"翻译 JSON、抛本语言异常"的薄适配。这一章讲这个引擎内部怎么转、以及它如何被三种语言复用。

---

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

一句话定义: engine-orchestrator 是 Memori 的"发动机舱"——一个纯 Rust 库,对外暴露少数几个方法(嵌入、检索、提交记忆生成、优雅关闭),对内藏着一个嵌入模型和两个后台工作池。

它解决什么问题。 Memori 是个记忆库(见 index),它要反复做两类脏活累活:

• 嵌入:把一句话变成一个几百维的浮点向量,这样才能算"语义相似"。这一步跑的是神经网络(ONNX 模型),CPU 密集、慢。
• 后台处理:把一段对话炼成结构化记忆(见 02-augmentation)、再写进数据库。这一步要发网络请求、要写库,慢且不能卡住主线程。

如果每个语言的 SDK 各写一遍这些逻辑,就会有三份不一致、三份 bug。Memori 的选择是:写一遍 Rust,三家共用。

给谁用。 直接使用者是 Memori 自己的三个 SDK(Python / Node / TS),不是终端开发者。终端开发者调的是 memori.recall(...) 这种高层 API,底下才转到这个引擎。

用起来什么样(以 Python 为例)。 引擎被包成一个 Python 类,调用像这样:

# 示意,非源码 —— 展示引擎对外的手感
from memori_python import EngineHandle

engine = EngineHandle(model_name, fetch_embeddings_cb, fetch_facts_cb, write_batch_cb)
engine.embed_texts(["hello world"])          # 同步:立刻拿到向量
engine.submit_augmentation(payload_json)     # 异步:塞进后台队列,立刻返回一个 job_id
engine.wait_for_augmentation(timeout_ms=5000) # 等后台把这批活干完

一句话直觉。 把引擎想成一家快餐店的后厨:

• 前台点单(embed)——你站着等,当场出餐(同步)。
• 外卖单(submit_augmentation)——丢进出单口就走,后厨有固定几个灶台(并发上限)慢慢做;出单口格子有限(队列有界),满了就直接告诉你"稍后再来"(拒单,而不是无限堆积)。
• 打烊流程(shutdown)——不再收新单,但把手上的单做完再关灯(优雅关闭)。

本节不出现代码细节;记住这三个手感即可。

---

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

怎么读下面这张图: 上层是三种语言的 SDK,中间那道虚线是"薄适配层",线以下全是同一个 Rust crate。数据都以 JSON 字符串穿过语言边界。

   Python SDK          Node SDK           TS SDK
       │                  │                  │
 ┌─────┴──────┐   ┌───────┴───────┐   ┌──────┴───────┐
 │ PyO3 适配  │   │ napi-rs 适配  │   │ 复用 napi    │   ← 薄:只翻译 JSON / 抛异常
 │ memori_    │   │ node-bindings │   │ (同一个 .node)│
 │ python     │   │               │   │              │
 └─────┬──────┘   └───────┬───────┘   └──────┬───────┘
       └──────────────────┼──────────────────┘
                          ▼
        ┌───────────────────────────────────────┐
        │   engine-orchestrator (Rust 核心)      │
        │                                         │
        │   EngineOrchestrator  ← 一个把手        │
        │     ├─ embedder ......... 同步嵌入      │
        │     ├─ postprocess_rt ... 后台 worker①  │
        │     ├─ augmentation_rt .. 后台 worker②  │
        │     └─ storage_bridge ... 回调宿主的库   │
        └───────────────────────────────────────┘
                          │
              ┌───────────┴───────────┐
              ▼                       ▼
      fastembed / ONNX        MemoriClient(HTTP)
      (向量模型)               (记忆生成 API)

部件一句话职责:

部件	干什么	在哪(相对 core/)
EngineOrchestrator	顶层把手:持有下面所有资源,对外暴露 embed/retrieve/submit/shutdown	src/lib.rs:86
EmbeddingEngine	只做嵌入的轻量把手(不建后台池)	src/lib.rs:57
SentenceTransformersEmbedder	包住 fastembed/ONNX 模型,懒加载	src/embeddings/models.rs:11
embed_texts	嵌入流水线:分块→批推理→池化→逐级降级	src/embeddings/api.rs:159
WorkerRuntime<J>	通用后台工作池:有界队列 + 并发上限 + 生命周期	src/runtime/worker.rs:90
MemoriClient	调 Memori 云端记忆生成 API 的 HTTP 客户端	src/network/client.rs:26
StorageBridge	让引擎回调宿主语言的数据库(BYODB)	src/storage/bridge.rs:6
OrchestratorError	统一错误类型,带稳定 status_code()	src/error.rs:5

主线走一遍(高层,不进代码):

1. SDK 构造引擎 → EngineOrchestrator::new_with_storage(src/lib.rs:104)加载嵌入模型、建两个后台池、建 HTTP 客户端。
2. 要向量 → embed(texts) 同步返回,当场算完(src/lib.rs:134)。
3. 要存记忆 → submit_augmentation(input) 把活塞进队列立刻返回(src/lib.rs:201);后台 worker 稍后调 API → 生成记忆 → 嵌入 → 写库。
4. 要等后台跑完 → wait_for_augmentation(src/lib.rs:213)。
5. 收工 → shutdown() 优雅关闭两个池(src/lib.rs:234)。

一条铁律(架构约束)。 依赖是单向的:核心 crate → 绑定适配,绝不反向;适配层"必须是薄的转换层,不得绕过引擎直接碰运行时内部"(core/docs/architecture.md)。这条约束是本章一切设计的底色。

---

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

3.1 EngineOrchestrator:一个把手,Arc 共享,克隆廉价

它要解决的小问题。 引擎里有些资源很贵——嵌入模型占内存、后台池占线程。SDK 可能要在多处传递引擎句柄。怎么让"传引擎"这件事既安全又便宜?

思路。 把每样贵资源都放进 Arc(原子引用计数指针),EngineOrchestrator 只持有这些 Arc。于是克隆一个 orchestrator ≈ 复制几个指针 + 加引用计数,底层的模型和线程池是共享的,不会被复制。

真实结构(src/lib.rs:85):

#[derive(Clone)]
pub struct EngineOrchestrator {
    embedder: Arc<SentenceTransformersEmbedder>,
    postprocess_runtime: WorkerRuntime<PostprocessJob>,     // 内部也是 Arc
    augmentation_runtime: WorkerRuntime<AugmentationJob>,
    storage_bridge: Option<Arc<dyn StorageBridge>>,
}

注释直接点破:"Cloning is cheap — the underlying resources are shared via Arc"(src/lib.rs:82-84)。WorkerRuntime 自己也是 #[derive(Clone)] 且内部只有一个 Arc<Inner<J>>(src/runtime/worker.rs:89),所以整个句柄克隆全程零重活。

两种把手,各取所需。 如果调用方只要嵌入,不需要后台池,就用更轻的 EmbeddingEngine(src/lib.rs:57)——它"deliberately does not create postprocess or augmentation worker runtimes"(src/lib.rs:53-55),只揣一个 Arc<embedder>。Python 的独立 embed_texts 函数走的就是它(见 §4.1)。

3.2 同步嵌入:一次批推理 + 三级降级 + 零向量兜底

它要解决的小问题。 把一批文本变成向量。听着简单,但有三个坑:文本可能太长(超过模型上下文窗口)、模型推理可能偶发失败、而调用方必须拿到一个形状正确的输出缓冲区(否则上层崩)。

思路——先把形状讲清楚。 embed_texts 返回 (Vec<f32>, [rows, cols]):一个扁平的浮点大缓冲 + 一个 [行数, 维度] 形状。扁平化是为了跨语言边界(FFI)好传——不用传"数组的数组"。

流水线怎么走(src/embeddings/api.rs:159 embed_texts):

输入 texts
  │  prepare_text_inputs  ── 丢掉纯空白/控制字符/零宽空格的串
  ▼
  plan_text(每条)         ── 超过 chunk_size 个 token? 
  │                           否→Single(原串,零拷贝)  是→Multi(切成多块)
  ▼
  build_flat_and_metas    ── 把所有块摊平成一个 flat 批,记住每条文档占哪几行
  ▼
  run_batch               ── 一次性 embed_batch(flat)  ← 正常路径
  │      失败↓
  fallback_sequential     ── 每条文档单独 embed
  │      失败↓
  fallback_one_by_one     ── 每个块逐个 embed
  │      失败↓
  zero_vectors            ── 最后兜底:返回全零向量,保证形状合法
  ▼
  多块文档 mean_pool_l2   ── 多个块向量取平均再 L2 归一,合回一条
  ▼
  拼成扁平 flat_out + [rows, dim]

为什么要三级降级? 注释说得很直白:降级链存在,是为了"callers always receive a well-formed buffer"(src/embeddings/api.rs:156-158)。哪怕模型彻底挂了,也宁可返回全零向量(zero_vectors,src/embeddings/utils.rs:15)让流水线继续,也不把异常抛给上层。上层拿到全零向量后,retrieve 会识别出来并报 ModelError(src/lib.rs:188-192)。

分块的门槛怎么定? 模型的 chunk_size = 最大序列长度 − 2(留给特殊 token),取自 HuggingFace 的配置文件,拿不到就默认 256(src/embeddings/models.rs:64-65)。分块本身按 token 而非字符切(chunk_text_by_tokens,src/embeddings/chunking.rs:10),切完还原成文本;如果切出来 ≤1 块就当没切、回退到原串。

多块怎么合回一条? mean_pool_l2(src/embeddings/api.rs:31):把一条长文档的各块向量逐维求平均,再 L2 归一化(除以模长)。归一后余弦相似度就退化成点积,检索时算得快——这一步是给 03-recall 的 dense 检索铺路。

模型是懒加载的。 SentenceTransformersEmbedder 里模型放在 Mutex<Option<TextEmbedding>>(src/embeddings/models.rs:14),第一次 embed 才真正建 ONNX 会话(with_model,src/embeddings/models.rs:87)。注释解释:"so engine startup does not require loading ONNX runtime"(src/embeddings/models.rs:12-13)——引擎构造要快,别一上来就吃掉几百 MB。

FFI 存取的两个小工具。 向量写库时用小端 f32 打包成字节(format_embedding_for_db,src/embeddings/utils.rs:20),读回时 chunks_exact(4) 解包(parse_embedding_from_db)——尾部凑不满 4 字节的直接丢弃,防脏数据。

3.3 有界后台 worker:队列 + 信号量 + 生命周期

这是全章工程含量最高的一支。WorkerRuntime<J> 是个通用后台工作池(泛型 J = 任务类型),Memori 建了两个实例:一个跑 postprocess、一个跑 augmentation(src/lib.rs:244、src/lib.rs:274)。

它要解决的小问题。 "提交一个后台任务"必须满足四条:

1. 不阻塞提交方(submit 立刻返回)。
2. 不无限堆积(内存爆炸)——队列要有界,满了要能拒。
3. 不打爆下游——同时在跑的任务数要有上限。
4. 能优雅收工——关闭时把手上的活做完。

四个部件对应四条需求:

需求	机制	代码
有界、满了拒单	tokio::mpsc 有界 channel + try_send	src/runtime/worker.rs:171
并发上限	tokio::Semaphore(max_concurrency)	src/runtime/worker.rs:68、:316
知道"还有几个活没干完"	AtomicUsize outstanding + OutstandingGuard	src/runtime/worker.rs:19、:174
状态机(未启动/运行/关闭中/已停)	Lifecycle(AtomicU8)	src/runtime/state.rs:26

提交路径怎么走(submit,src/runtime/worker.rs:160):

match tx.try_send(job) {
    Ok(()) => { self.inner.outstanding.fetch_add(1, SeqCst); Ok(()) }   // 计数+1
    Err(TrySendError::Full(job)) => Err(SubmitError::QueueFull(job)),    // 满 → 把 job 还回去
    Err(TrySendError::Closed(_)) => Err(SubmitError::ShuttingDown),
}

注意一个巧思: 队列满时,错误里把 job 原样还给调用方(QueueFull(job)),让调用方自己决定重试还是丢弃——引擎不替你做主。orchestrator 侧则把它映射成 OrchestratorError::QueueFull(map_submit_error,src/lib.rs:363)。

执行路径——分发器 + 信号量(run_dispatcher,src/runtime/worker.rs:309):

while let Some(job) = rx.recv().await {
    let permit = sem.clone().acquire_owned().await? ;   // 拿不到许可就在这等(并发闸门)
    tokio::spawn(async move {
        let _guard = OutstandingGuard { .. };           // 任务结束时 outstanding-1
        let _permit = permit;                           // 任务结束时归还许可
        handler(job).await;
    });
}

分发器是个单线程循环:每收到一个 job,先向信号量要一张许可(max_concurrency 张,orchestrator 里设成 2),要不到就 await 挂起——这就是并发上限。拿到许可才 spawn 出去跑。permit 和 OutstandingGuard 都靠 Drop 自动释放:任务一结束,许可归还、计数减一。

"等所有活干完"怎么实现(flush,src/runtime/worker.rs:183): flush 阻塞在一个条件变量上,循环检查 outstanding == 0。谁来叫醒它?OutstandingGuard::drop 里,当计数从 1 减到 0 时(prev == 1)调 wake_waiters()(src/runtime/worker.rs:23-29)——最后一个任务干完,唤醒所有等待者。flush_for 是带超时版,超时返回 FlushError::Timeout(src/runtime/errors.rs:29)。

优雅关闭(shutdown,src/runtime/worker.rs:225)——一套精细的时序:

1. 抢 shutdown_stage 锁,判重(已关/关闭中/未启动都短路)——保证 shutdown 幂等
2. 状态置 ShuttingDown,把 job_tx 置 None → channel 关闭
3. channel 一关,dispatcher 的 rx.recv() 返回 None → 跳出收单循环
4. dispatcher 进入尾部循环:等 outstanding 归零(drain 完手上的活)
5. 主线程 block_on 等 dispatcher join 完,再 shutdown_timeout(30s) 拆掉 tokio 运行时
6. 状态置 Stopped,唤醒所有 shutdown 等待者

关闭策略目前只有 Drain(把接下的活全做完再停,src/runtime/config.rs:36)。shutdown 幂等——用 ShutdownStage 三态(NotShutdown/InProgress/Done)+ 条件变量保证:多次调用、并发调用都安全,后来者会等第一个关完(src/runtime/worker.rs:227-253)。

一个必须知道的坑(死锁)。 模块文档明确警告:不要在一个 job 内部(它正跑在该 runtime 的 worker 线程上)去调本 runtime 的 flush/shutdown(src/runtime/mod.rs:19-22)——会自己等自己,死锁。

配置与共享 tokio。 RuntimeConfig(src/runtime/config.rs:5)有 queue_capacity、max_concurrency、worker_threads,以及一个重要选项 tokio_handle:若宿主已有 tokio 运行时,传进它的 Handle,worker 就复用而不是再套一个嵌套运行时(src/runtime/worker.rs:130)——推荐做法,见模块文档(src/runtime/mod.rs:6-8)。orchestrator 目前用的是独立运行时(worker_threads: Some(1),src/lib.rs:249)。构造时 validate() 会拒掉非法配置(容量/并发 <1,src/runtime/config.rs:44)。

3.4 命令校验与状态码:前置守卫 + 稳定错误码

它要解决的小问题。 引擎是给三种语言远程调的,输入未必干净。校验必须在引擎侧做一次(而非指望每个 SDK 都记得校验),错误还得能跨语言稳定表达。

前置校验都是纯函数,进队列前就拦。 三个守卫:

• execute_command(src/lib.rs:383):目前只认 "ping" → "pong",空串报 InvalidInput,其余报 UnsupportedCommand。
• validate_postprocess_payload(src/lib.rs:397):payload 不能全空白。
• validate_augmentation_input(src/lib.rs:407):entity_id 必填,且至少要有一条非空的对话消息或一段 content——两者皆空直接拒。

这些校验在 submit 之前跑(src/lib.rs:150、:205),脏输入根本进不了队列。

错误码为什么要稳定? OrchestratorError(src/error.rs:5)每个变体有一个固定的 status_code()(src/error.rs:26):InvalidInput=1、UnsupportedCommand=2、QueueFull=3……绑定层就靠这个数字把 Rust 错误映射成本语言的异常类型(见 §4.1)。测试 status_codes_are_distinct 专门守着"所有码互不相同"(src/lib.rs:537)——防止有人加变体时撞码、破坏映射。

---

4. 跨语言绑定(深入):同一个核心,三种语言的皮

这一节讲那道虚线上面——薄适配层如何把 Rust 引擎变成 Python 类、Node 类、TS 对象。三家的共同套路:输入 JSON 字符串 → 反序列化成引擎类型 → 调引擎 → 结果序列化回 JSON / 转本语言类型 → 用本语言的异常表达错误。

4.1 Python(PyO3)

PyO3 crate memori_python(core/bindings/python/src/lib.rs)开头就自陈身份:"This crate is a thin adapter... All business logic lives in the root engine-orchestrator crate"(:4-6)。它导出三个 #[pyclass]:

Python 类	揣着什么	干什么
MemoriEngine(:110)	EngineOrchestrator(无存储)	execute / embed / postprocess
NativeEmbedder(:147)	轻量 EmbeddingEngine	只嵌入
EngineHandle(:167)	EngineOrchestrator(带存储桥)	全功能:retrieve/recall/augmentation

三个关键手法:

① 释放 GIL 再干重活。 嵌入是 CPU 密集的,阻塞时若攥着 Python 全局锁(GIL)会卡死整个解释器。所以每个重方法都用 py.detach(|| ...) 释放 GIL 再进 Rust(:143、:210、:217)。

② 错误码 → Python 异常。 orchestrator_error_to_py_err(:321)读 status_code():1/2 → PyValueError(用户输入错),3 → PyRuntimeError("queue is full"),其余 → PyRuntimeError。这正是 §3.4 稳定错误码的用武之地。

③ 扁平缓冲 → 嵌套列表。 引擎给的是 (flat, [rows, cols]),reshape_embedding_result(:307)按 dim 切成 Vec<Vec<f32>> 再交给 Python;rows/dim 任一为 0 就返回空。模块级 embed_texts 函数还带一个按模型名缓存的轻量引擎池(cached_embedding_engine,:287),避免重复加载模型。

存储回调怎么过界(BYODB 关键)。 PythonStorageBridge(:29)实现 Rust 的 StorageBridge trait,但它没有 Rust 数据库——它揣着三个 Python 回调(fetch_embeddings / fetch_facts_by_ids / write_batch)。每次引擎要读写库,就把请求序列化成 JSON,在一个新线程里回调 Python(call_json_callback,:38),带 30 秒超时。为什么开新线程 + 超时?因为回调要重新 attach GIL,直接在引擎线程上调容易和 GIL 打结;超时则防宿主回调卡死后台 worker。存储层的宿主侧实现见 05-storage。

Python 侧的粘合层。 Rust 扩展不是裸用的,memori/native/ 下有 Python 包装:

• _onnxruntime.py(_ensure_onnxruntime_dylib,:305):Rust 的 fastembed 需要 ONNX Runtime 动态库。这个模块按平台查表下载对应的 ORT 二进制(带 SHA256 校验、文件锁防并发下载、多镜像重试),再设 ORT_DYLIB_PATH 环境变量让 Rust 找到它(_configure_onnxruntime_env,:296)。
• _adapter.py(RustCoreAdapter,:44):真正被 Python SDK 用的门面。maybe_create 只在 BYODB + 启用 rust core 时才建(:52);_create_engine(:68)才真正 import EngineHandle 并注入那三个存储回调;引擎懒建且用锁保护、失败会缓存异常(_active_engine,:89)。它把 retrieve/recall/submit_augmentation 都翻译成"拼 JSON → 调引擎 → 解析 JSON"。

4.2 Node(napi-rs)

Node 绑定 node-bindings(core/bindings/node/src/)导出一个 #[napi] 的 MemoriEngine(engine.rs:14)。和 Python 同构,但异步模型不同——它大量用 spawn_blocking 把阻塞活挪出 Node 事件循环(embed_texts,engine.rs:128;retrieve,:152)。

两个 Node 特有的精细处:

① 打破引用环。 存储管理器要能嵌入(写库前给记忆算向量),嵌入能力又在引擎里,于是形成环:引擎 → storage_manager → embedder → 引擎。解法是用 Arc::downgrade 弱引用注入嵌入闭包(engine.rs:47-59),环被打断,引擎能正常析构。

② 存储桥走 ThreadsafeFunction。 NodeConnectionFactory(bridge.rs:127)通过一个 TS 侧的 storageCall(id, payloadJson) 线程安全函数(TSFN)做所有数据库操作,协议是纯 JSON(acquire/execute/begin/commit/rollback/close,协议表见 bridge.rs:113-126)。Rust 侧发起调用后用 block_in_place + block_on 等 TS 用 resolveStorageCall(id, resultJson) 回填(Inner::call,bridge.rs:51),带 30 秒超时。shutdown 会先丢掉 TSFN、再唤醒所有卡在等待里的 Rust 线程,避免它们干等到超时(bridge.rs:172-190)。

类型桥。 types.rs 用 #[napi(object)] 定义 NapiRetrievalRequest/NapiAugmentationInput/NapiRecallObject 等,Either<i64, String> 表达"fact id 可能是数字也可能是字符串"这种跨库差异(types.rs:24)。

4.3 TS SDK:复用同一个 .node

TypeScript SDK(memori-ts/)不另建 Rust——它直接加载 Node 绑定产出的原生模块。NativeEngine(memori-ts/src/core/engine.ts:14)是对 napi MemoriEngine 的 TS 包装:构造时把 StorageManager 的 handleStorageCall 接到 Rust 的 storageCall 回调上(engine.ts:56-71),再把 snake_case ↔ camelCase 来回翻(retrieve,engine.ts:116)。它还处理进程退出时的优雅关闭(beforeExit 里逐个 shutdown,engine.ts:31-39)。

一句话: TS 是 Node 绑定的"上层 TypeScript 皮",两者共享同一个 .node 二进制、同一份 Rust 核心。

4.4 网络客户端 MemoriClient

后台 augmentation worker 要调 Memori 云端的记忆生成 API,走的是 MemoriClient(src/network/client.rs:26)。要点:

• 从环境变量配置(new,:35):base URL、X-Memori-API-Key、可选 bearer token,还有 MEMORI_TEST_MODE 切 staging;默认值对齐 Python SDK。
• 指数退避重试(post_async,:95):5xx 服务器错误和网络错误都重试,最多 5 次,退避 2^attempt 秒(do_backoff,:218);连续 SSL/TLS 关键字的错误单独归为 ApiError::Ssl。
• 按状态码分错(handle_response,:228):429→QuotaExceeded、422→Validation、433→Rejected,其余→Client。这些错误经 OrchestratorError::ApiError(码 6)透传。

业务上"这些 API 返回什么、如何炼成记忆"属于 02-augmentation;这里只讲传输机制。

---

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

• 克隆即共享,零重活。 顶层把手和 WorkerRuntime 都只裹一个 Arc,#[derive(Clone)] 让"传引擎"变成"加引用计数"(src/lib.rs:85、src/runtime/worker.rs:89)。
• 降级链 + 兜底,永不把异常甩给上层。 嵌入从批处理逐级退到逐条,最后返回零向量保证形状合法(src/embeddings/api.rs:159)。这是"宁可退化、不可崩溃"的工程哲学。
• 满队列把 job 还给你。 SubmitError::QueueFull(job) 把被拒的任务原样交还,让背压决策留给调用方(src/runtime/worker.rs:177)。
• Drop 即释放。 并发许可(Semaphore permit)和在途计数(OutstandingGuard)都靠 Rust 的 Drop 语义自动归还,无需手写清理、无泄漏(src/runtime/worker.rs:19-29、:322-328)。
• 稳定错误码当跨语言 ABI。 status_code() 把语义错误压成小整数,绑定层据此映射本语言异常;测试守着码不重复(src/error.rs:26、src/lib.rs:537)。
• 弱引用破环。 Node 侧用 Arc::downgrade 切断"引擎↔存储↔嵌入"的循环引用(core/bindings/node/src/engine.rs:47)。
• 懒加载贵资源。 ONNX 模型第一次用才建,引擎构造保持轻快(src/embeddings/models.rs:12-14)。

---

6. 边界与局限(诚实)

• 不是分布式,是进程内。 WorkerRuntime 是单进程的后台池,不是跨机队列;队列容量、并发都是进程内的内存/线程数(src/runtime/mod.rs:1-5)。
• 关闭策略只有 Drain。 ShutdownPolicy 目前唯一变体是 Drain(标注为 v1),没有"立刻丢弃在途任务"的选项(src/runtime/config.rs:34-40)。
• job 内不能 flush/shutdown 自己的 runtime,否则死锁(src/runtime/mod.rs:19-22)。
• 存储回调有 30 秒硬超时。 Python 与 Node 桥都是 30s(bindings/python/src/lib.rs:36、bindings/node/src/bridge.rs:13);宿主数据库回调若更慢会被判超时,而非无限等。
• execute 命令面几乎是空的。 目前只支持 ping(src/lib.rs:391),是个占位/健康检查接口,真正的功能都走专用方法。
• 嵌入彻底失败时静默退化成零向量。 好处是流水线不崩,代价是除非下游显式检查(如 retrieve 检查全零,src/lib.rs:188),否则可能悄悄用了无意义的向量。
• ONNX Runtime 靠运行时下载。 Python 侧按平台联网拉 ORT 二进制(memori/native/_onnxruntime.py:305);离线环境需预置 ORT_DYLIB_PATH 或关掉自动下载(MEMORI_ORT_AUTO_DOWNLOAD=0)。

---

7. 横向对比(同 shelf 兄弟)

本章讲的是 Memori 内部的引擎机制与多语言复用,与同组其它章的边界:

• 后台 worker 具体跑什么业务(把对话炼成结构化记忆)→ 02-augmentation。
• 嵌入向量最终怎么被检索(dense 余弦 + BM25 混合重排)→ 03-recall。
• 存储桥另一端宿主怎么落库(Cloud vs BYODB、多方言驱动)→ 05-storage。
• 引擎入口如何被 SDK 劫持挂上(register 拦截 LLM 调用)→ 01-capture。

在 ai-agent-reference 这个书架里,Memori 代表"把记忆基础设施下沉到原生核心、多语言共用一份实现"的取舍——用 Rust 换性能与一致性,用 JSON + trait 回调换语言无关的存储/网络边界。

---

8. 代码地图(导航索引)

主题	文件(相对 core/)	关键符号
顶层把手 / 生命周期	src/lib.rs	EngineOrchestrator、new_with_storage、embed、submit_augmentation、wait_for_augmentation、shutdown
轻量嵌入把手	src/lib.rs	EmbeddingEngine
后台池初始化	src/lib.rs	init_postprocess_runtime、init_augmentation_runtime
命令 / 校验 / 错误映射	src/lib.rs	execute_command、validate_postprocess_payload、validate_augmentation_input、map_submit_error、map_flush_error
统一错误 + 状态码	src/error.rs	OrchestratorError、status_code
嵌入流水线	src/embeddings/api.rs	embed_texts、mean_pool_l2、fallback_sequential、fallback_one_by_one
ONNX 模型封装	src/embeddings/models.rs	SentenceTransformersEmbedder、with_model、embed_batch
分块	src/embeddings/chunking.rs	chunk_text_by_tokens
输入清洗 / FFI 打包	src/embeddings/utils.rs	prepare_text_inputs、zero_vectors、format_embedding_for_db、parse_embedding_from_db
后台工作池	src/runtime/worker.rs	WorkerRuntime、start、submit、flush、shutdown、run_dispatcher、OutstandingGuard
运行时配置	src/runtime/config.rs	RuntimeConfig、ShutdownPolicy、validate
运行时错误	src/runtime/errors.rs	SubmitError、RuntimeError、FlushError
生命周期状态机	src/runtime/state.rs	Lifecycle、LifecycleState
HTTP 客户端	src/network/client.rs	MemoriClient、post_async、do_backoff、handle_response
存储桥 trait	src/storage/bridge.rs	StorageBridge
Python 绑定	bindings/python/src/lib.rs	MemoriEngine、NativeEmbedder、EngineHandle、PythonStorageBridge、orchestrator_error_to_py_err、cached_embedding_engine
Node 绑定	bindings/node/src/engine.rs	MemoriEngine(napi)、embed_texts、弱引用注入
Node 存储桥	bindings/node/src/bridge.rs	NodeConnectionFactory、NodeConnection、Inner::call
Node 类型桥	bindings/node/src/types.rs	NapiRetrievalRequest、NapiAugmentationInput、NapiRecallObject
TS 引擎包装	memori-ts/src/core/engine.ts	NativeEngine
Python ORT 引导	memori/native/_onnxruntime.py	_ensure_onnxruntime_dylib、_configure_onnxruntime_env
Python 适配门面	memori/native/_adapter.py	RustCoreAdapter、maybe_create、_create_engine
架构约束	core/docs/architecture.md	单向依赖、薄适配层