tinyhumansai/openhuman · src/openhuman/memory_queue/

慢活全丢给队列，UI 永不阻塞

写入热路径只做"切碎、快速打分、入库、入队"，所有烧资源的事——LLM 实体抽取、封桶、re-embed——都被推进一个持久化的 SQLite 作业队列，由后台 worker 池跑。本页基于 memory_queue/{types,worker}.rs 拆解：六种作业、worker 池怎么认领、全局 LLM 信号量怎么节流、失败怎么退避与延后。

6
作业类型
4
并发 worker
1
全局 LLM 信号量
5s
轮询 + notify 唤醒

六种作业类型

mem_tree_jobs 表里每行存一个字符串判别符（discriminator），对应 JobKind 枚举的六种当前作业：

🔬
ExtractChunk

对单个 chunk 跑 LLM 实体抽取并决定准入（admission）。是写入流水线里最核心的 LLM 作业。

📥
AppendBuffer

把内容追加进缓冲层。非 LLM 作业，便宜，不占信号量。

📦
Seal

封掉恰好一个缓冲层；级联时再 enqueue 一个后续 Seal——封桶被拆成一步一作业，而不是一次性递归。

🧹
FlushStale

遍历陈旧缓冲，为任何超过 age cap 的层 enqueue Seal 作业。非 LLM 作业。

🧲
ReembedBackfill

为缺少当前 embedding 签名向量的 chunk/summary 有界批量重嵌入（换模型后回填）。

📄
SealDocument

封一个文档级单元。与 Seal 同属封桶家族的 LLM 作业。

四种是 LLM-bound（is_llm_bound() 返回 true）：ExtractChunk / Seal / ReembedBackfill / SealDocument——它们要排队等 LLM 信号量；AppendBuffer / FlushStale 不占信号量、随便跑。parse() 还对已退役的旧作业（topic_route / digest_daily，来自被移除的 global/topic 树）做容错：撞上就报"retired JobKind"而非崩溃——一个把历史队列残留优雅吞掉的设计。

mem_tree_jobs：状态机与原子认领

作业有五个状态 ready / running / done / failed / cancelled。worker 通过 claim_next 原子认领一个 job——一条 UPDATE 同时设 locked_until_ms + status='running'，于是多个 worker 不会抢到同一个 job。除了"成功/失败"，作业还有第三种归宿——Defer：

  • mark_done / mark_failed_typed / mark_deferred 三个收尾动作。
  • JobOutcome::Defer 让一个作业（比如在等某个上游就绪）带一个唤醒时间重新入队，而不被标记为失败——把"现在还不能做"和"做失败了"在状态机层面分开。
  • recover_stale_locks() 把 locked_until_ms 已过期却仍 running 的 job 重新置 ready，交给别的 worker——防止某个 worker 崩了把 job 永久锁死。
  • claim_ready_extract_batch：ExtractChunk 支持批量认领，一次取一批待抽取 chunk，摊薄认领开销。

4 个 worker + notify 驱动的轮询

worker 池是 WORKER_COUNT = 4 个 tokio::spawn 出来的并发 task（不是"三协程"——真实数字是 4），外加一个每日 scheduler。每个 worker 是一个 loop：认领 → 跑 → 收尾 → 再认领。空闲时它不傻等，而是 tokio::select! 在"POLL_INTERVAL = 5s 定时"和"WORKER_NOTIFY 被唤醒"之间二选一：

memory_queue/worker.rs — 空闲等待
// 有新作业入队就 wake_workers()，让空闲 worker 立刻重新轮询，
// 而不必干等满 5 秒
tokio::select! {
    _ = notify.notified()                       => {}  // 被唤醒
    _ = tokio::time::sleep(POLL_INTERVAL)        => {}  // 兜底轮询
}

这是经典的"低延迟 + 有兜底"组合：wake_workers() 在入队时戳一下 Notify，让新作业几乎即时被捡起；而 5s 的定时轮询保证即使 notify 漏了（或作业是定时到期的 Defer），也总会被扫到。

全局 LLM 信号量：唯一的真相源

LLM 并发控制不归 worker 自己管，而是委托给 scheduler_gate 的一个全局信号量（LlmPermit）。worker 只调一句 wait_for_capacity() 拿许可。关键是这个信号量是跨子系统唯一的一份——worker、语音清理、自动补全、triage、reflection 全都从它取号：

memory_queue/worker.rs — 取号
// 门控许可：所有要花算力的子系统共享同一个全局信号量
let gate_permit = scheduler_gate::wait_for_capacity().await;
// 仅 LLM-bound 作业再额外占一个 LLM permit
let llm_permit = if job.kind.is_llm_bound() { … } else { None };

好处是不会出现"记忆队列和语音清理各开各的并发、合起来把 LLM 端点打爆"——整个进程对 LLM 的并发压力有一个集中的、可调的闸门。非 LLM 作业（AppendBuffer / FlushStale）压根不取 LLM permit，不会被这道闸拦住。

云端放 permit、本地保 permit

同一套代码要同时伺候"本地 Ollama"和"云端 API"两种后端，它们的并发约束正好相反，于是 worker 在认领后会按后端区别对待 LLM permit：

  • 本地 Ollama：每个并发调用常驻约 1.3 GB 显存/内存，多开会 OOM——所以保住 permit，让 LLM 作业串行，4 个 worker 在有 LLM 作业在飞时只能彼此并行跑非 LLM 部分。
  • 云端后端：瓶颈是网络往返而非本地资源——所以认领后立即丢掉全局 LLM permit，让全部 4 个 worker 都能并行打云端调用。这也是注释里说"bump WORKER_COUNT 在云端才真正提速"的原因。

分级退避：按错误类别决定睡多久

worker 的错误处理不是"统一退避 N 秒"，而是按错误类别分级——它认得 SQLite 的具体错误码：

  • SQLite busy/locked（多 worker + scheduler 抢同一个 DB 的正常争用）→ 退避 1s，下一轮重试，不上报 Sentry。
  • 瞬态 I/O 错误（IOERR_TRUNCATE 1546、-shm 家族 4618/4874/5386、IN_PAGE 8714、CANTOPEN 14）或熔断 → 退避 30s，给文件系统/WAL 喘息（曾因不退避 4 天刷了 ~19K 事件，见 #2206）。

配合状态机层的 Defer（带唤醒时间重新入队），整套重试语义是分层的：瞬态资源争用靠 worker 循环里的睡眠退避，"上游还没就绪"靠 Defer 把作业挪到未来再试，而真失败才走 mark_failed_typed。三类被刻意区分，避免把"暂时做不了"误记成"做坏了"。

UI 永不阻塞：热路径只入队

把上面拼起来，"UI 永不阻塞"就有了具体含义：一条新记忆进来时，前台热路径只做最便宜的几件事——切碎成 chunk、快速打分、写库、往 mem_tree_jobs 入队，然后 wake_workers() 戳一下队列就返回。所有要等 LLM、要重嵌入、要级联封桶的慢活，都变成队列里的行，由后台 4 个 worker 异步消化。

加上存储层用了 WAL（读不阻塞写），前台的读查询和后台 worker 的写入可以并发；Notify 让入队到开工几乎零延迟；Defer + 定时轮询让"还不能做"的作业自动等到能做的时刻——用户侧永远是"立即返回"，重活在看不见的地方慢慢追上。

意义：把"会长大"的代价摊进一个 SQLite 队列

一棵会自己长大的记忆树，意味着持续不断的抽取、嵌入、封桶——这些如果同步做，UI 会卡死、LLM 端点会被打爆。OpenHuman 的答案是一个朴素但完备的持久化作业系统：六种作业把工作切成可独立重试的小步，mem_tree_jobs 状态机 + 原子认领让 4 个 worker 安全并发，scheduler_gate 的全局信号量给整个进程的 LLM 并发一道集中闸门，permit 的云/本地差异化榨干两种后端，分级退避 + Defer 把"争用/未就绪/真失败"分得清清楚楚。它把"长大"的全部代价，摊进了一个可崩溃恢复、可观测、可调速的后台队列里——这正是前台敢承诺"永不阻塞"的底层理由。