数据入口与缓冲区(为什么先攒再处理)

30 秒导读: 你往 Memobase 塞的每一段对话(blob),不会立刻触发一次昂贵的 LLM 抽取。它先原样落进数据库,再挂进一个按用户分组的缓冲区;只有当这个用户攒够了(token 超过阈值),系统才把这一批一次性喂给记忆管线。这样把"每次调 LLM 的固定成本"摊薄到很多条对话上——这就是本章要讲清的省钱设计。

本章只讲数据怎么进来、怎么被缓冲、批处理怎么省钱;真正把一批 blob 变成用户画像的三次 LLM 调用留给 02-extract-merge-pipeline.md。全景与阅读地图见 index.md。

---

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

一句话定义: blob(二进制大对象在这里泛指"一段原始输入",主要是一段聊天记录)是 Memobase 的最小摄入单位;缓冲区(buffer)是一个按用户攒 blob 的暂存队列,攒够才批量处理。

它要解决的问题: 记忆系统的贵活儿是 LLM 抽取——从对话里提炼"这个用户喜欢什么、是谁"。如果每来一条消息就调一次 LLM,那 90% 的成本花在"启动一次 LLM 调用"的固定开销上,而单条消息往往信息量很小。

直觉类比: 像拼车。一个人打车(每条 blob 单独处理)最贵;等同路的人凑一车再走(攒够一批再处理),每个人分摊的车费就低得多。缓冲区就是那个"等人上车"的站台。

用起来什么样: 调用方(SDK / HTTP)往某个 user_id 插入一段对话:

# 示意,非源码:一次 blob 插入
POST /api/v1/blobs/insert/{user_id}?wait_process=false
{
  "blob_type": "chat",
  "blob_data": {"messages": [
    {"role": "user", "content": "我最近在学做意大利菜"},
    {"role": "assistant", "content": "那你可以从番茄意面入手"}
  ]}
}
# 返回一个 blob id;至于要不要立刻抽取,由缓冲区攒够没决定

关键点:插入接口立即返回,是否真的跑 LLM,取决于这个用户的缓冲区是否被这条 blob 撑满。wait_process 决定"如果确实要处理,你是等它跑完(同步)还是让它在后台跑(异步)"。

---

2. 顶层全景(一条 blob 从进入到被缓冲)

先看一条 blob 进来后走的路。这张图从上到下是时间顺序,注意最后的分叉:大多数时候缓冲区没满,请求到"挂进缓冲区"就结束了。

调用方 insert_blob 请求
        │
        ▼
① api_layer/blob.py:insert_blob   ← HTTP 入口
        │   先做 billing 校验(token 用超了直接拒)
        ▼
② controllers/blob.py:insert_blob ← 原始 blob 落库 GeneralBlob 表
        │   拿到 blob_id
        ▼
③ controllers/buffer.py:insert_blob_to_buffer ← 在 BufferZone 表加一行(status=idle)
        │
        ▼
④ controllers/buffer.py:detect_buffer_full_or_not ← 这个用户 idle 的 blob token 加起来超阈值没?
        │
   ┌────┴─────────────┐
   │ 没超              │ 超了(要 flush)
   ▼                  ▼
 直接返回 blob_id   ⑤ flush_buffer_by_ids
 (最常见)         ┌──────┴───────┐
                  │ wait_process │ wait_process
                  │  = true      │  = false
                  ▼              ▼
              同步跑完再返回   丢进后台任务,立即返回
              (buffer.py)   (buffer_background.py)

各部件一句话职责:

部件	干什么	文件:符号
HTTP 入口	校验 billing、编排"落库→入缓冲→判满→flush"	api_layer/blob.py:insert_blob
blob 落库	把原始 blob 写进 GeneralBlob 表	controllers/blob.py:insert_blob
入缓冲区	在 BufferZone 表加一行、记该 blob 的 token 数	controllers/buffer.py:insert_blob_to_buffer
判满	累加该用户 idle blob 的 token,过阈值就返回待处理 id 列表	controllers/buffer.py:detect_buffer_full_or_not
flush(同步)	把一批 blob join 出来喂给管线,跑状态机	controllers/buffer.py:flush_buffer_by_ids
flush(后台)	加锁 + Redis 队列,后台批量跑 flush	controllers/buffer_background.py:flush_buffer_by_ids_in_background

---

3. 核心原理

3.1 入口:billing 校验 + 同步/异步两条路

它要解决的小问题: 入口既要挡住欠费请求,又要让"处理"这件慢事不阻塞调用方(除非调用方明确要等)。

先校验 billing。 拿到项目的 billing,token_left 为负直接拒绝,连 blob 都不落库:

# api_layer/blob.py:insert_blob(节选)
if billing.token_left is not None and billing.token_left < 0:
    return Promise.reject(CODE.SERVICE_UNAVAILABLE, ...)  # 超额,拒绝

见 api_layer/blob.py:insert_blob(第 33-40 行)。这道闸在落库之前,避免欠费用户继续写入。

再编排三步: 落库 → 入缓冲 → 判满。三步任一失败都提前返回,见 api_layer/blob.py:insert_blob(第 43-60 行)。

最后的分叉是本节重点。 只有 detect_buffer_full_or_not 返回了非空的待处理 id 列表,才真的去 flush;而 flush 有同步/异步两条路,由 wait_process 这个 query 参数决定:

wait_process	走哪条路	行为	引用
true	同步	当场 await flush_buffer_by_ids,把 chat_results 塞进响应返回	api_layer/blob.py:insert_blob:67-73
false(默认)	异步	background_tasks.add_task(...flush_buffer_by_ids_in_background),立即返回	api_layer/blob.py:insert_blob:76-82

所以默认情况下,即使触发了 flush,调用方也不用等 LLM——处理在 FastAPI 的后台任务里跑。

3.2 落库:原始 blob 进 GeneralBlob

它要解决的小问题: 缓冲区里只放"指针 + token 数",真正的对话内容得有个地方存。那个地方就是 GeneralBlob 表。

controllers/blob.py:insert_blob 先把 BlobData 解析成具体类型(chat / doc / summary),失败即 BAD_REQUEST;成功就写一行:

# controllers/blob.py:insert_blob(节选)
blob_db = GeneralBlob(
    blob_type=blob_parsed.type,
    blob_data=blob_parsed.get_blob_data(),   # 对话内容,存 JSONB
    additional_fields=blob_parsed.fields,
    user_id=user_id, project_id=project_id,
)
session.add(blob_db); session.commit()
return Promise.resolve(IdData(id=blob_db.id))   # 返回新 blob 的 id

见 controllers/blob.py:insert_blob(第 14-25 行)。注意: 这一步只是原始存档,和"记忆抽取"无关。这条 blob 会不会被处理、什么时候被处理,完全由后面的缓冲区决定。

3.3 缓冲区:按用户攒 blob,靠 token 阈值触发

这是本章的心脏。分三小步讲。

第一步:入缓冲区。 insert_blob_to_buffer 在 BufferZone 表加一行,关键是把这条 blob 的 token 数算好存下来,状态设为 idle:

# controllers/buffer.py:insert_blob_to_buffer(节选)
buffer = BufferZone(
    user_id=user_id, blob_id=blob_id, blob_type=blob_data.type,
    token_size=get_blob_token_size(blob_data),   # 预先算好 token,判满时直接加
    project_id=project_id, status=BufferStatus.idle,
)
session.add(buffer); session.commit()

见 controllers/buffer.py:insert_blob_to_buffer(第 33-47 行)。token_size 存下来是为了判满时不用重新 tokenize,直接把这个用户所有 idle 行的 token_size 相加即可。

第二步:判满。 detect_buffer_full_or_not 查出该用户、该 blob_type、status=idle 的所有缓冲行,把 token_size 求和,超过 CONFIG.max_chat_blob_buffer_token_size(默认 1024,见 env.py:Config:86)就返回这批 id;否则返回空列表:

# controllers/buffer.py:detect_buffer_full_or_not(节选)
buffer_zone = session.query(BufferZone.id, BufferZone.token_size).filter_by(
    user_id=user_id, blob_type=str(blob_type),
    project_id=project_id, status=BufferStatus.idle).all()
buffer_ids = [row.id for row in buffer_zone]
buffer_token_size = sum(row.token_size for row in buffer_zone)
if buffer_token_size and buffer_token_size > CONFIG.max_chat_blob_buffer_token_size:
    return Promise.resolve(IdsData(ids=buffer_ids))   # 满了,交出这批 id
return Promise.resolve(IdsData(ids=[]))               # 没满,先攒着

见 controllers/buffer.py:detect_buffer_full_or_not(第 66-94 行)。这就是"拼车"的发车条件:攒到 1024 token 才发车,把一次 LLM 抽取的固定成本摊到这一整批对话上。

第三步:flush 时按 token 再截断。 发车后进入管线,还有一道 truncate_chat_blobs:从最新的 blob 往回累加,只保留 token 总量不超过 max_chat_blob_buffer_process_token_size(默认 16384,env.py:Config:87)的那些,防止一次喂给 LLM 的量过大:

# controllers/modal/chat/__init__.py:truncate_chat_blobs(节选)
for b in blobs[::-1]:                      # 从最新往旧扫
    total_token_size += len(get_encoded_tokens(get_blob_str(b)))
    if total_token_size <= max_token_size:
        results.append(b)
    else:
        break                              # 超了就丢掉更旧的
return results[::-1]

见 controllers/modal/chat/__init__.py:truncate_chat_blobs(第 23-35 行),由 process_blobs(第 42 行)在处理开头调用。

两个 token 阈值别混淆: max_chat_blob_buffer_token_size(1024)是发车线——攒够就 flush;max_chat_blob_buffer_process_token_size(16384)是上车限载——真喂给 LLM 前的截断上限。前者管"何时处理",后者管"一次处理多少"。

3.4 flush 的状态机与失败回滚

它要解决的小问题: 一批 blob 从"待处理"到"处理完",中途可能失败(LLM 挂了、DB 出错)。得有个状态字段记录每行走到哪一步,失败了能标记出来而不是丢失。

BufferZone.status 就是这个状态机(取值见 env.py:BufferStatus,第 59-63 行):

   插入时
     │
     ▼
  ┌──────┐   detect 判满 + flush 开始   ┌────────────┐
  │ idle │ ─────────────────────────▶ │ processing │
  └──────┘                            └─────┬──────┘
                                            │
                          管线成功 ┌─────────┴─────────┐ 管线失败/异常
                                  ▼                   ▼
                              ┌──────┐            ┌────────┐
                              │ done │            │ failed │
                              └──────┘            └────────┘

flush_buffer_by_ids 的执行顺序(见 controllers/buffer.py:flush_buffer_by_ids,第 117-249 行):

1. join 取数(第 133-156 行): 把 BufferZone 和 GeneralBlob 一次 join 出来,拿到 token 数和真正的对话内容,按 created_at 排序(保证对话时序)。
2. 置 processing(第 159-165 行): 把这批行标记为 processing,防止别人再当它是 idle。
3. 调管线(第 190 行): await BLOBS_PROCESS[blob_type](...) 跑真正的抽取(细节见 02)。
4. 成功 → done(第 205-210 行);失败 → failed。 管线返回失败时回滚为 failed(第 193-199 行);抛异常时同样在 except 里置 failed 并重抛(第 234-249 行)。

非持久化时删原始 chat blob(第 211-215 行): 处理成功后,若 blob_type == chat 且 CONFIG.persistent_chat_blobs 为 False(默认就是 False,env.py:Config:77),就把这批原始 chat blob 从 GeneralBlob 删掉:

# controllers/buffer.py:flush_buffer_by_ids(节选)
if blob_type == BlobType.chat and not CONFIG.persistent_chat_blobs:
    session.query(GeneralBlob).filter(
        GeneralBlob.id.in_(blob_ids),
        GeneralBlob.project_id == project_id,
    ).delete(synchronize_session=False)   # 记忆已抽出,原始对话不再留存

设计意图(inferred):记忆(画像、事件)已经从对话里抽出来了,原始逐字对话没必要长期占存储,默认就删——想保留原文得显式开 persistent_chat_blobs。

已知隐患(源码自带 FIXME): flush_buffer_by_ids 第 124 行有注释 # FIXME: parallel calling will cause duplicated flush。"判满(读 idle)"和"置 processing(写)"之间没有锁,并发插入同一用户时,两个请求可能查到重叠的 idle id 并各自 flush 同一批,导致重复处理。后台路径(3.5)用 Redis 锁部分缓解了这个问题,但同步路径仍暴露。

3.5 后台 flush:Redis 锁 + 队列

它要解决的小问题: 异步路径(wait_process=false)不能让多个后台任务同时处理同一用户的缓冲区(既浪费 LLM 钱,又踩 3.4 的重复 flush 坑)。

flush_buffer_by_ids_in_background(见 controllers/buffer_background.py,第 40-98 行)的做法:

1. 先置 processing(第 49-72 行): 只挑仍是 idle 的行标为 processing,拿到 actual_buffer_ids。
2. 入 Redis 队列(第 74-84 行): 把这批 id 打包成串 rpush 进该用户的队列。
3. 抢锁再消费(flush_buffer_background_running,第 101 行起): 用 SET nx ex 抢一把用户级 Redis 锁(第 122-125 行);抢不到就直接返回(说明已有任务在处理该用户)。抢到的那个循环 lpop 队列、逐批调 flush_buffer_by_ids,并带 select_status=BufferStatus.processing(第 188-194 行,因为第 1 步已经置过 processing 了)。

这里有几道自我保护上限(都是 flush_buffer_background_running 的默认参数):

保护	值	作用
max_iterations	200	单次运行最多处理 200 批就退出
max_processing_time_s	15 分钟	总处理超时就 break
process_interval_s	5 分钟	锁的 TTL,循环里续期
max_consecutive_errors	5	连续 5 次失败就停,避免死循环烧钱

锁在 finally 里用 Lua 脚本"值匹配才删"释放(第 253-269 行),避免误删别人续上的锁。

---

4. 数据结构:两张表

缓冲机制落在两张表上(models/database.py)。

GeneralBlob(原始存档,models/database.py:285):

字段	类型	说明
id	UUID	主键(与 project_id 组成复合主键)
user_id	UUID	属于哪个用户
blob_type	VARCHAR(16)	chat / doc / summary
blob_data	JSONB	原始内容(对话消息等)
project_id	VARCHAR(64)	多租户隔离

BufferZone(暂存队列,models/database.py:345):

字段	类型	说明
id	UUID	主键
blob_id	UUID	指向 GeneralBlob.id(外键,第 402-407 行)
blob_type	VARCHAR(16)	同 blob 类型
token_size	Integer	该 blob 的 token 数,判满时求和用
status	VARCHAR(16)	状态机:idle/processing/done/failed(默认 idle,第 364-366 行)
user_id / project_id	UUID / VARCHAR	分组键

关键索引 idx_buffer_zones_user_id_blob_type(第 389-395 行)覆盖 (user_id, project_id, blob_type, status)——正好是 detect_buffer_full_or_not / get_unprocessed_buffer_ids 的过滤条件,让"查某用户某类型的 idle 缓冲"走索引。

status 的四个字面量在 env.py:BufferStatus(第 59-63 行)集中定义,数据库列本身是普通 VARCHAR、不是 DB 枚举,靠应用层约束。

---

5. 巧妙之处(可借鉴)

• token 数在入缓冲时就算好存下来(insert_blob_to_buffer:41),判满(detect_buffer_full_or_not)只做整数求和,不重复 tokenize——把贵的 tokenize 摊到写入时、且只做一次。
• 缓冲区只存指针(blob_id + token_size),内容留在 GeneralBlob,判满这种高频操作只扫一张窄表。
• join 一次取全(flush_buffer_by_ids:135-156):把 buffer 元数据和 blob 内容一次 join 出来,避免 N+1 查询;顺带按 created_at 排序保证喂给 LLM 的对话时序正确。
• 默认删原始 chat(persistent_chat_blobs=False):记忆抽完就丢原文,把存储成本压到画像/事件层。
• 后台路径的多重刹车:Redis 用户锁 + 连续错误熔断 + 总时长上限,防止一个坏用户把后台 worker 拖死或烧钱。

---

6. 边界与局限

• 同步 flush 的并发重复处理(源码 FIXME,buffer.py:124):同步路径无锁,并发插入可能重复 flush 同一批。后台路径靠 Redis 锁缓解,但同步路径未防护。
• 发车靠 token 阈值,不靠时间:detect_buffer_full_or_not 只看 token 是否超 1024。一个用户如果长期只发零星短消息、始终不到阈值,这批 blob 就一直躺在 idle 不被处理(存在按时间兜底的 buffer_flush_interval=1 小时配置和独立的 flush_buffer,但 insert_blob 主路径不主动按时间触发)。
• failed 状态无自动重试:flush 失败置为 failed 后,主入口路径不会自动重跑;需要外部机制重新捞 failed 行(本章引用的代码里未见自动重试逻辑)。
• billing 只在入口挡一次:insert_blob 只在写入前查 token_left < 0;一次 flush 内部 LLM 的实际消耗不在这条路径里做二次预算校验。

---

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

主题	文件路径	符号名
HTTP 入口 / billing 校验 / 同步·异步分叉	src/server/api/memobase_server/api_layer/blob.py	insert_blob
原始 blob 落库	src/server/api/memobase_server/controllers/blob.py	insert_blob
入缓冲区(记 token)	src/server/api/memobase_server/controllers/buffer.py	insert_blob_to_buffer
判满(token 阈值触发)	src/server/api/memobase_server/controllers/buffer.py	detect_buffer_full_or_not
flush 状态机 / join / 删原始 chat	src/server/api/memobase_server/controllers/buffer.py	flush_buffer_by_ids
取未处理 buffer id	src/server/api/memobase_server/controllers/buffer.py	get_unprocessed_buffer_ids
后台 flush(锁 + 队列)	src/server/api/memobase_server/controllers/buffer_background.py	flush_buffer_by_ids_in_background / flush_buffer_background_running
处理前按 token 截断	src/server/api/memobase_server/controllers/modal/chat/__init__.py	truncate_chat_blobs / process_blobs
blob token 计算	src/server/api/memobase_server/utils.py	get_blob_token_size / pack_blob_from_db
表结构	src/server/api/memobase_server/models/database.py	GeneralBlob / BufferZone
状态枚举 / token 阈值配置	src/server/api/memobase_server/env.py	BufferStatus / Config.max_chat_blob_buffer_token_size

下一章: 一批 blob 发车后,BLOBS_PROCESS[chat] 内部的三次 LLM 调用(抽取 → 合并 → 组织)见 02-extract-merge-pipeline.md;抽出来的画像与事件怎么存见 03-storage-model.md。