fix(memory): 增量向量化 + 单分片逐条隔离，召回请求数降一个数量级（M61）
M15（2026-10-13）登记的两条放大源落在记忆层，本次一并收口： 1. 整理轮/入库审查**只嵌增量**：新加 vector.Index.Vectors（SQLite/PG 实现， 按 id 取已存向量、维度不符当未命中），Store.embedIncremental 先复用索引里的 向量、只嵌缺的那些，并把手算出的向量回填索引 ⇒ 同一分区第二轮起零请求。 判据写明为「向量存在性 + 维度一致」（正文在本仓不可就地改写，故无需哈希）。 维度与当前 embedder 不一致时禁用复用并 WARN（不混两种表示）。 2. 嵌入链对单个慢分片有韧性：embedIsolated 整批失败后逐条隔离，坏的那几条 记缺口、其余照常返回；一条都没拿到才报错降级。请求数上界 1+B（B=8 ⇒ 9）， 服务整体挂死时一次即收手，不用「请求放大」换韧性。 3. 缺口/复用/回填条数进日志（缺口走 WARN），review 的被审条目拿不到向量时 如实返回错误 ⇒ 走既有词项余弦降级（不把「算不出来」说成「没有重复」）。 请求数（单测内假 OpenAI 兼容服务端计数，生产 API embedder + 真 SQLite 索引）： 24 条存量：修复前每轮 3 → 第一轮 3、第二轮 0 12 条分区整理轮：第一轮 2 → 第二轮 0 20 条候选的入库审查：26（201÷8）→ 第一条 3、第二条 0 一条坏正文：保住 7/8 条，代价 9 个请求；整体性故障 9 个请求后收手
This commit is contained in:
AskaEth committed
1 parent
9d28477e3a
commit
589e26c887
15 files changed
+1343
-48
No files matched your search
@@ -0,0 +1,282 @@
|
||||
package memory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory/vector"
|
||||
)
|
||||
|
||||
// 本文件是「增量向量化」：召回链只嵌**索引里还没有**的行。
|
||||
//
|
||||
// ## 背景（M15，2026-10-13 登记的放大源）
|
||||
//
|
||||
// 整理轮每轮把整个分区重新向量化（user 分区一轮 ≈ 250 个请求）、入库审查每条把
|
||||
// 200 条既有候选重新嵌一遍（26 个请求）—— 而 `vec_items` 里本来就有这些行的向量
|
||||
// （写入侧 store_vector.go 每条 Add 都写一条）。同一段正文在下一轮从零再算一遍是
|
||||
// 纯浪费，也是 M15 记的「请求放大」在**记忆层**的那一半（另一半是 embedding 包的
|
||||
// 切片形状，已由 M15 把 4 条/片改成 8 条/片）。
|
||||
//
|
||||
// ## 增量判据：向量存在性 + 维度一致（不用内容哈希）
|
||||
//
|
||||
// 复用条件是「`vec_items` 里该 item.ID 的向量存在，且长度等于索引维度」。不用
|
||||
// 内容哈希，理由是可证的一条：**记忆正文在本仓不存在就地改写** —— `Store.Add` 是
|
||||
// 唯一 INSERT 路径、merge 是软删除（`status`+`merged_into`）、reclassify 只改分类列、
|
||||
// 结构性改写一律「新行 + 旧行置为被合并」（design/23 §15 D3-6）。正文只增不改 ⇒
|
||||
// 「有向量」等价于「那个向量就是这段正文的向量」，哈希比对不提供额外信息。
|
||||
//
|
||||
// ⚠️ 已知边界（既存缺口，不在本切片内）：`vec_items` 不带**模型标签**。若换了
|
||||
// embedding 模型而维度恰好相同，复用会把两种表示混进同一轮的余弦里。可用的判据
|
||||
// 只有维度（本层用的就是它）——要根治得给向量表加一列 model，属迁移级改动。
|
||||
//
|
||||
// ## 算完回填索引（否则「增量」只会增到同样的那几条）
|
||||
//
|
||||
// 只复用不填，第二轮还是复用到同样那几条：索引的覆盖面本来只有「索引接入之后
|
||||
// 写入的条目 + 一次性 ≤512 条回填」（retrieve_index.go），一个 2500 条的分区因此
|
||||
// 长期大部分没有向量。所以本层在算完之后把向量 upsert 回索引（与
|
||||
// store_vector.indexItem 同一件事、同一条 id），使**第二轮起**同一分区的请求数
|
||||
// 趋近于 0。回填失败只计数 + 一条 WARN，绝不让召回失败（召回已经在手上的向量上
|
||||
// 跑完了）。
|
||||
|
||||
// ErrNilEmbedder 在「要增量向量化却没有 embedder」时返回。生产调用点都先判了
|
||||
// nil（未配置语义 embedding 就走词项余弦），因此它只防内部装配错误。
|
||||
var ErrNilEmbedder = errors.New("memory: nil embedder")
|
||||
|
||||
// embedReuseStats 是一轮召回**实际向远端发了什么**的计数。它进日志而不是进
|
||||
// 候选/留痕表：请求数是运维观测，不改变任何判定结果。
|
||||
type embedReuseStats struct {
|
||||
// Requested 是参与召回的条目数（本轮需要向量的条目）。
|
||||
Requested int
|
||||
// Reused 是索引里已有向量、一次远端都没发的条目数。
|
||||
Reused int
|
||||
// Sent 是真正送进 embedder 的**条数**（≈ 请求数 × 批上限）。
|
||||
Sent int
|
||||
// Dropped 是逐条隔离重试后仍失败、本轮没有向量的条目数（缺口）。
|
||||
Dropped int
|
||||
// Filled 是本轮回填进索引的向量数。
|
||||
Filled int
|
||||
// FillFailed 是回填失败的条数（不影响召回）。
|
||||
FillFailed int
|
||||
}
|
||||
|
||||
// approxRequests 估算本轮发出的 HTTP 请求数：条数 ÷ 批上限（向上取整）。
|
||||
// 它是**观测用的估算**（embedding 包内部还按同一个批上限切片，因此两者一致），
|
||||
// 不是任何判据。
|
||||
func (s embedReuseStats) approxRequests(batch int) int {
|
||||
if s.Sent <= 0 {
|
||||
return 0
|
||||
}
|
||||
if batch <= 0 {
|
||||
batch = tidyEmbedBatch
|
||||
}
|
||||
return (s.Sent + batch - 1) / batch
|
||||
}
|
||||
|
||||
// logEmbedReuse 如实记一笔本轮「复用 / 真发请求 / 缺口」。缺口必须看得见
|
||||
// （绝不静默），正常的复用只记 Debug —— 它每轮每分区都会发生，不该刷屏。
|
||||
func logEmbedReuse(log *slog.Logger, caller string, stats embedReuseStats, batch int) {
|
||||
if log == nil || stats.Requested == 0 {
|
||||
return
|
||||
}
|
||||
attrs := []any{
|
||||
"caller", caller,
|
||||
"items", stats.Requested,
|
||||
"reused", stats.Reused,
|
||||
"sent", stats.Sent,
|
||||
"requests", stats.approxRequests(batch),
|
||||
"dropped", stats.Dropped,
|
||||
"filled", stats.Filled,
|
||||
}
|
||||
if stats.Dropped > 0 {
|
||||
log.Warn("memory: semantic recall ran with embedding gaps", attrs...)
|
||||
return
|
||||
}
|
||||
log.Debug("memory: semantic recall reused indexed vectors", attrs...)
|
||||
}
|
||||
|
||||
// embedIncremental 返回与 items 对齐的向量（拿不到的位置为 nil）：索引里已有的
|
||||
// 直接复用，只有缺的才送进 embedder。batch 是单次 Embed 的条数上限（生产传
|
||||
// tidyEmbedBatch，它 = embedding.DefaultBatchLimit）。
|
||||
//
|
||||
// 失败语义（design/23 §15.3 已登记的韧性口径：**一片失败不丢整条**）：
|
||||
// - 单批内部失败由 embedIsolated **逐条隔离**，仍失败的那几条留下**缺口**，其余
|
||||
// 条目照常带向量返回；
|
||||
// - 整批一条都没拿到（整体性故障）时停止后续批，已拿到的向量照常返回；
|
||||
// - 只有「一条向量都没拿到」才返回错误 —— 调用方据此走既有的词项余弦降级。
|
||||
func (s *Store) embedIncremental(ctx context.Context, emb embedding.Embedder, items []MemoryItem, batch int) ([]embedding.Vector, embedReuseStats, error) {
|
||||
stats := embedReuseStats{Requested: len(items)}
|
||||
vecs := make([]embedding.Vector, len(items))
|
||||
if len(items) == 0 {
|
||||
return vecs, stats, nil
|
||||
}
|
||||
if emb == nil {
|
||||
return nil, stats, ErrNilEmbedder
|
||||
}
|
||||
idx := s.vectorIndex()
|
||||
if idx != nil && emb.Dim() > 0 && emb.Dim() != idx.Dim() {
|
||||
// 🔴 维度不一致说明索引与当前 embedder 不是同一套表示（换过模型 / 换过维度）。
|
||||
// 此时**既不复用也不回填**：复用会把两种维度的向量混进同一轮余弦（长度不符
|
||||
// ⇒ 相似度恒 0，静默降质），回填则必然逐条 ErrDimMismatch。退回「全量嵌入」，
|
||||
// 并如实留一条 WARN，让「索引该重建了」这件事看得见。
|
||||
s.log.Warn("memory: embedder dim differs from vector index dim, incremental reuse disabled",
|
||||
"embedder_dim", emb.Dim(), "index_dim", idx.Dim())
|
||||
idx = nil
|
||||
}
|
||||
if idx != nil {
|
||||
reused, err := s.reuseIndexedVectors(ctx, idx, items, vecs)
|
||||
if err != nil {
|
||||
// 索引读失败**不**降级语义召回：退回「全量嵌入」（改动前的行为），代价是
|
||||
// 这一轮又放大成 ⌈N/批⌉ 个请求 —— 如实记 WARN，让请求数异常有据可查。
|
||||
s.log.Warn("memory: read indexed vectors failed, embedding the whole partition", "error", err)
|
||||
vecs = make([]embedding.Vector, len(items))
|
||||
} else {
|
||||
stats.Reused = reused
|
||||
}
|
||||
}
|
||||
missing := missingVectorIndexes(vecs)
|
||||
if len(missing) == 0 {
|
||||
return vecs, stats, nil
|
||||
}
|
||||
stats.Sent = len(missing)
|
||||
if err := s.embedMissing(ctx, emb, items, missing, vecs, &stats, batch); err != nil {
|
||||
return nil, stats, err
|
||||
}
|
||||
if idx != nil {
|
||||
s.fillIndexVectors(ctx, idx, items, missing, vecs, &stats)
|
||||
}
|
||||
return vecs, stats, nil
|
||||
}
|
||||
|
||||
// reuseIndexedVectors 把索引里已有的向量摊到 vecs 的对应位置，返回复用条数。
|
||||
// 只读：它不写 vecs 之外的任何东西，也不产生 embedding 请求。
|
||||
func (s *Store) reuseIndexedVectors(ctx context.Context, idx vector.Index, items []MemoryItem, vecs []embedding.Vector) (int, error) {
|
||||
ids := make([]string, 0, len(items))
|
||||
positions := make([]int, 0, len(items))
|
||||
for i, it := range items {
|
||||
if it.ID == "" {
|
||||
continue
|
||||
}
|
||||
ids = append(ids, it.ID)
|
||||
positions = append(positions, i)
|
||||
}
|
||||
if len(ids) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
stored, err := idx.Vectors(ctx, ids)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
reused := 0
|
||||
for k, pos := range positions {
|
||||
if k >= len(stored) || len(stored[k]) == 0 {
|
||||
continue
|
||||
}
|
||||
vecs[pos] = embedding.Vector(stored[k])
|
||||
reused++
|
||||
}
|
||||
return reused, nil
|
||||
}
|
||||
|
||||
// missingVectorIndexes 返回还没有向量的那些条目的下标（顺序即条目顺序，
|
||||
// 入库审查的新条目因此在最前）。
|
||||
func missingVectorIndexes(vecs []embedding.Vector) []int {
|
||||
missing := make([]int, 0, len(vecs))
|
||||
for i, v := range vecs {
|
||||
if len(v) == 0 {
|
||||
missing = append(missing, i)
|
||||
}
|
||||
}
|
||||
return missing
|
||||
}
|
||||
|
||||
// embedMissing 只嵌入缺向量的那些条目，把结果写回 vecs 的对应位置。
|
||||
//
|
||||
// 批级失败（整体性故障）当场收手：对已经挂掉的服务继续发后续批就是 M15 记的
|
||||
// 那种放大。缺口条数进 stats，绝不静默。
|
||||
func (s *Store) embedMissing(ctx context.Context, emb embedding.Embedder, items []MemoryItem,
|
||||
missing []int, vecs []embedding.Vector, stats *embedReuseStats, batch int) error {
|
||||
if batch <= 0 {
|
||||
batch = tidyEmbedBatch
|
||||
}
|
||||
var firstErr error
|
||||
for start := 0; start < len(missing); start += batch {
|
||||
end := min(start+batch, len(missing))
|
||||
idxs := missing[start:end]
|
||||
texts := make([]string, len(idxs))
|
||||
for k, i := range idxs {
|
||||
texts[k] = items[i].Content
|
||||
}
|
||||
got, err := embedIsolated(ctx, emb, texts)
|
||||
if err != nil {
|
||||
if firstErr == nil {
|
||||
firstErr = err
|
||||
}
|
||||
break
|
||||
}
|
||||
for k, i := range idxs {
|
||||
if k >= len(got) || len(got[k]) == 0 {
|
||||
continue
|
||||
}
|
||||
vecs[i] = got[k]
|
||||
}
|
||||
}
|
||||
stats.Dropped = stats.Sent - countVectorsAt(vecs, missing)
|
||||
if stats.Sent == stats.Dropped {
|
||||
if firstErr != nil {
|
||||
return firstErr
|
||||
}
|
||||
return ErrEmbedUnavailable
|
||||
}
|
||||
if stats.Dropped > 0 {
|
||||
s.log.Warn("memory: embedding left gaps in semantic recall",
|
||||
"missing", len(missing), "dropped", stats.Dropped, "error", firstErr)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// countVectorsAt 数 idxs 这些位置里已有向量的条数。
|
||||
func countVectorsAt(vecs []embedding.Vector, idxs []int) int {
|
||||
n := 0
|
||||
for _, i := range idxs {
|
||||
if i < len(vecs) && len(vecs[i]) > 0 {
|
||||
n++
|
||||
}
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// fillIndexVectors 把本轮**算出来的**向量 upsert 回索引（与 store_vector.indexItem
|
||||
// 同一件事、同一条 id），使下一轮不再重算同一批正文。
|
||||
//
|
||||
// 失败只计数 + 一条 WARN：召回已经在本轮拿到的向量上跑完了，回填失败不该让它
|
||||
// 变成一次失败。维度不符（索引与当前 embedder 不是同一套表示）时当场收手 ——
|
||||
// 继续逐条试只会刷出上千条同样的失败。
|
||||
func (s *Store) fillIndexVectors(ctx context.Context, idx vector.Index, items []MemoryItem,
|
||||
idxs []int, vecs []embedding.Vector, stats *embedReuseStats) {
|
||||
mismatch := false
|
||||
for _, i := range idxs {
|
||||
if items[i].ID == "" || len(vecs[i]) == 0 {
|
||||
continue
|
||||
}
|
||||
err := idx.Add(ctx, items[i].ID, vector.Vector(vecs[i]))
|
||||
if err == nil {
|
||||
stats.Filled++
|
||||
continue
|
||||
}
|
||||
stats.FillFailed++
|
||||
if errors.Is(err, vector.ErrDimMismatch) {
|
||||
mismatch = true
|
||||
break
|
||||
}
|
||||
}
|
||||
switch {
|
||||
case mismatch:
|
||||
s.log.Warn("memory: vector index dim mismatch, index fill stopped for this round",
|
||||
"filled", stats.Filled, "failed", stats.FillFailed)
|
||||
case stats.FillFailed > 0:
|
||||
s.log.Warn("memory: vector index fill failed for some items",
|
||||
"filled", stats.Filled, "failed", stats.FillFailed)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user