From 589e26c887747c70b66390cd0a5a3d1a0301f4fc Mon Sep 17 00:00:00 2001 From: AskaEth Date: Fri, 30 Oct 2026 08:30:06 +0800 Subject: [PATCH] =?UTF-8?q?fix(memory):=20=E5=A2=9E=E9=87=8F=E5=90=91?= =?UTF-8?q?=E9=87=8F=E5=8C=96=20+=20=E5=8D=95=E5=88=86=E7=89=87=E9=80=90?= =?UTF-8?q?=E6=9D=A1=E9=9A=94=E7=A6=BB=EF=BC=8C=E5=8F=AC=E5=9B=9E=E8=AF=B7?= =?UTF-8?q?=E6=B1=82=E6=95=B0=E9=99=8D=E4=B8=80=E4=B8=AA=E6=95=B0=E9=87=8F?= =?UTF-8?q?=E7=BA=A7=EF=BC=88M61=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 个请求后收手 --- .../runtime/memory/embed_fake_server_test.go | 163 ++++++++++ internal/runtime/memory/embed_incremental.go | 282 ++++++++++++++++++ .../runtime/memory/embed_incremental_test.go | 172 +++++++++++ internal/runtime/memory/embed_isolation.go | 109 +++++++ .../runtime/memory/embed_resilience_test.go | 162 ++++++++++ internal/runtime/memory/preinsert_config.go | 7 + internal/runtime/memory/preinsert_recall.go | 56 ++-- internal/runtime/memory/retrieve_m4_test.go | 5 + internal/runtime/memory/tidy_config.go | 6 +- internal/runtime/memory/tidy_recall.go | 34 +-- internal/runtime/memory/vector/index.go | 9 +- internal/runtime/memory/vector/lookup.go | 68 +++++ internal/runtime/memory/vector/pg.go | 74 +++++ internal/runtime/memory/vector/sqlite.go | 47 +++ .../runtime/memory/vector/vectors_test.go | 197 ++++++++++++ 15 files changed, 1343 insertions(+), 48 deletions(-) create mode 100644 internal/runtime/memory/embed_fake_server_test.go create mode 100644 internal/runtime/memory/embed_incremental.go create mode 100644 internal/runtime/memory/embed_incremental_test.go create mode 100644 internal/runtime/memory/embed_isolation.go create mode 100644 internal/runtime/memory/embed_resilience_test.go create mode 100644 internal/runtime/memory/vector/lookup.go create mode 100644 internal/runtime/memory/vector/vectors_test.go diff --git a/internal/runtime/memory/embed_fake_server_test.go b/internal/runtime/memory/embed_fake_server_test.go new file mode 100644 index 0000000..d6c28df --- /dev/null +++ b/internal/runtime/memory/embed_fake_server_test.go @@ -0,0 +1,163 @@ +package memory + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/config" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory/vector" +) + +// 本文件是「增量向量化 + 单分片韧性」的**请求计数**测试（M61，2026-10-30）。 +// +// 判据全部落在**真实的生产路径**上：假服务端是一个 OpenAI 兼容的 +// POST /v1/embeddings（应答 {data:[{index,embedding}]}），被测对象是生产用的 +// embedding.NewAPI + vector.New(SQLite) + Store.embedIncremental / Tidier / +// PreInsertReviewer —— 不是替身。因此这里数出来的请求数就是线上会发出去的请求数。 +// +// 断言的三件事： +// 1. **增量**：索引里已有向量的行不再重嵌（第二轮整理轮 = 0 个请求）； +// 2. **不放大**：同输入 N 条只发 ⌈N/批上限⌉ 个请求（批上限 = 8）； +// 3. **韧性**：一条坏/慢分片只损失它自己（缺口），其余照常；整体性故障最多 +// 3 个请求就收手（不把放大换韧性）。 + +const fakeEmbedDim = 8 + +// fakeDatum / fakeResponse 是 /embeddings 的最小应答形态。 +type fakeDatum struct { + Index int `json:"index"` + Embedding []float32 `json:"embedding"` +} + +type fakeResponse struct { + Data []fakeDatum `json:"data"` +} + +// fakeEmbedServer 是**计数的** OpenAI 兼容假服务端。failIf 命中任一输入文本时 +// 该请求回 400（模拟「一条坏正文」），failAll 时全部回 503（模拟服务挂死）。 +type fakeEmbedServer struct { + mu sync.Mutex + requests int + batches []int + failIf string + failAll bool + dim int + srv *httptest.Server +} + +// newFakeEmbedServer 起一个假服务端，并在 t.Cleanup 里关掉它。 +func newFakeEmbedServer(t *testing.T) *fakeEmbedServer { + t.Helper() + f := &fakeEmbedServer{dim: fakeEmbedDim} + f.srv = httptest.NewServer(http.HandlerFunc(f.handle)) + t.Cleanup(f.srv.Close) + return f +} + +// apiURL 返回 OpenAI 兼容的完整端点（与 .env 里 EMBEDDING_API_URL 的口径一致）。 +func (f *fakeEmbedServer) apiURL() string { return f.srv.URL + "/v1/embeddings" } + +// embedder 返回**生产实现**（embedding.NewAPI），只把端点指到假服务端。 +func (f *fakeEmbedServer) embedder() *embedding.APIEmbedder { + return embedding.NewAPI(config.EmbeddingConfig{APIURL: f.apiURL(), Model: "fake-embed"}) +} + +// count 返回到目前为止收到的请求数。 +func (f *fakeEmbedServer) count() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.requests +} + +// handle 记一笔请求并按 failIf / failAll 决定应答。 +func (f *fakeEmbedServer) handle(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(io.LimitReader(r.Body, 1<<20)) + var req struct { + Input []string `json:"input"` + } + if err := json.Unmarshal(body, &req); err != nil { + http.Error(w, `{"error":"bad request"}`, http.StatusBadRequest) + return + } + f.mu.Lock() + f.requests++ + f.batches = append(f.batches, len(req.Input)) + failAll, failIf := f.failAll, f.failIf + f.mu.Unlock() + + if failAll { + http.Error(w, `{"error":"unavailable"}`, http.StatusServiceUnavailable) + return + } + if failIf != "" { + for _, text := range req.Input { + if strings.Contains(text, failIf) { + http.Error(w, `{"error":"bad input"}`, http.StatusBadRequest) + return + } + } + } + resp := fakeResponse{Data: make([]fakeDatum, len(req.Input))} + for i, text := range req.Input { + resp.Data[i] = fakeDatum{Index: i, Embedding: fakeVector(text, f.dim)} + } + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(resp); err != nil { + http.Error(w, `{"error":"encode"}`, http.StatusInternalServerError) + } +} + +// fakeVector 由正文算一个确定性向量：同一段正文永远得到同一个向量（embedder +// 的确定性是复用索引向量的前提），不同正文仍在同一个扇区（余弦 > 召回门）。 +func fakeVector(text string, dim int) []float32 { + var h uint32 + for _, r := range text { + h = h*31 + uint32(r) + } + v := make([]float32, dim) + for i := range v { + v[i] = 1 + } + v[0] = float32(1 + h%5) + return v +} + +// fakeIndex 用**生产构造路径** vector.New 建一个真实 SQLite 索引（隔离内存库）， +// embedder 是假服务端背后的生产 API embedder。构造时会探测一次维度 ⇒ 1 个请求。 +func fakeIndex(t *testing.T, emb embedding.Embedder) vector.Index { + t.Helper() + idx, err := vector.New(config.DBConfig{SQLitePath: ":memory:"}, emb, testLogger()) + if err != nil { + t.Fatalf("vector.New: %v", err) + } + t.Cleanup(func() { _ = idx.Close() }) + return idx +} + +// reviewItems 造 n 条同分区条目并**在接入索引之前**落库（模拟历史存量：写入侧 +// 那时还没接索引，所以它们没有向量 —— 这正是整理轮每轮重嵌的那批）。 +func reviewItems(t *testing.T, store *Store, n int) []MemoryItem { + t.Helper() + base := tidyTestTime(t) + items := make([]MemoryItem, 0, n) + for i := 0; i < n; i++ { + items = append(items, seedTidyItem(t, store, MemoryItem{ + PersonaID: "persona.a", + Scope: ScopePersona, + Kind: KindRelationship, + Content: fmt.Sprintf("存量记忆正文 第%d条", i), + Confidence: 0.9, + CreatedAt: base.Add(time.Duration(i) * time.Minute), + SourceConversation: FormatSourceConversation("web", "m-1"), + })) + } + return items +} diff --git a/internal/runtime/memory/embed_incremental.go b/internal/runtime/memory/embed_incremental.go new file mode 100644 index 0000000..403345c --- /dev/null +++ b/internal/runtime/memory/embed_incremental.go @@ -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) + } +} diff --git a/internal/runtime/memory/embed_incremental_test.go b/internal/runtime/memory/embed_incremental_test.go new file mode 100644 index 0000000..ab84596 --- /dev/null +++ b/internal/runtime/memory/embed_incremental_test.go @@ -0,0 +1,172 @@ +package memory + +import ( + "context" + "testing" +) + +// 本文件断言两件事的**请求数**（M61，2026-10-30）： +// - 增量：索引里已有向量的行不再重嵌 —— 同一分区第二轮整理轮/审查是 **0** 个请求； +// - 韧性：一条坏分片只损失它自己，整体性故障最多 3 个请求就收手。 +// +// 计数来自假服务端（embed_fake_server_test.go），被测对象是生产路径 +// （embedding.NewAPI + vector.New + Store/Tidier/PreInsertReviewer），不是替身。 + +// batchRequests 是「按批上限发出去要几个请求」的算式（⌈n/batch⌉）。 +func batchRequests(n, batch int) int { + if n <= 0 || batch <= 0 { + return 0 + } + return (n + batch - 1) / batch +} + +// TestEmbedIncremental_SecondRoundMakesNoRequests 是「增量后不重嵌」的钉死： +// 第一轮把 24 条存量嵌入并**回填索引**（3 个请求 = ⌈24/8⌉），第二轮**一次都不发**。 +// +// 删掉复用实现（或把 embedIncremental 改回「每次全量嵌」）这一条立刻打红： +// 第二轮会重新发出同样的 3 个请求。 +func TestEmbedIncremental_SecondRoundMakesNoRequests(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + items := reviewItems(t, store, 24) + + srv := newFakeEmbedServer(t) + emb := srv.embedder() + // vector.New 会探测一次 embedder 维度（1 个请求），因此基线在它之后取。 + store.attachVectorIndex(fakeIndex(t, emb)) + + base := srv.count() + first, stats1, err := store.embedIncremental(ctx, emb, items, tidyEmbedBatch) + if err != nil { + t.Fatalf("embedIncremental(first): %v", err) + } + firstReqs := srv.count() - base + if want := batchRequests(len(items), tidyEmbedBatch); firstReqs != want { + t.Fatalf("第一轮请求数 = %d, want %d（⌈%d/%d⌉）", firstReqs, want, len(items), tidyEmbedBatch) + } + if stats1.Reused != 0 || stats1.Sent != len(items) || stats1.Filled != len(items) || stats1.Dropped != 0 { + t.Fatalf("第一轮 stats = %+v, want reused=0 sent=%d filled=%d dropped=0", stats1, len(items), len(items)) + } + + base = srv.count() + second, stats2, err := store.embedIncremental(ctx, emb, items, tidyEmbedBatch) + if err != nil { + t.Fatalf("embedIncremental(second): %v", err) + } + secondReqs := srv.count() - base + if secondReqs != 0 { + t.Fatalf("第二轮请求数 = %d, want 0（已有向量的行不该重嵌）", secondReqs) + } + if stats2.Reused != len(items) || stats2.Sent != 0 || stats2.Filled != 0 { + t.Fatalf("第二轮 stats = %+v, want reused=%d sent=0 filled=0", stats2, len(items)) + } + for i := range items { + if len(second[i]) == 0 { + t.Fatalf("第 %d 条复用失败：向量为空", i) + } + if len(second[i]) != len(first[i]) { + t.Fatalf("第 %d 条维度变了：%d -> %d", i, len(first[i]), len(second[i])) + } + for k := range first[i] { + if second[i][k] != first[i][k] { + t.Fatalf("第 %d 条复用到了别的向量（分量 %d）", i, k) + } + } + } + t.Logf("整理/审查同量记忆的请求数：修复前每轮 %d → 修复后第一轮 %d、第二轮 %d（%d 条，批上限 %d）", + batchRequests(len(items), tidyEmbedBatch), firstReqs, secondReqs, len(items), tidyEmbedBatch) +} + +// TestTidyRecall_SecondRoundZeroRequests 是整理轮的生产路径断言：同一分区连续两轮 +// recallPairs，第一轮按批上限发请求，第二轮**零请求**（候选召回因此不再每轮重嵌）。 +func TestTidyRecall_SecondRoundZeroRequests(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + reviewItems(t, store, 12) + + srv := newFakeEmbedServer(t) + emb := srv.embedder() + store.attachVectorIndex(fakeIndex(t, emb)) + + tidier, err := NewTidier(TidyConfig{}, TidyDeps{Store: store, Embedder: emb, Log: testLogger()}) + if err != nil { + t.Fatalf("NewTidier: %v", err) + } + partition := tidyPartition{personaID: "persona.a", scope: ScopePersona} + items, err := store.listTidyPartitionItems(ctx, partition, 500) + if err != nil { + t.Fatalf("listTidyPartitionItems: %v", err) + } + if len(items) != 12 { + t.Fatalf("分区条目数 = %d, want 12", len(items)) + } + + base := srv.count() + ledger := &recallLedger{} + if _, err := tidier.recallPairs(ctx, partition, items, &TidyStats{}, tidyBudget{}, ledger); err != nil { + t.Fatalf("recallPairs(round 1): %v", err) + } + firstReqs := srv.count() - base + want := batchRequests(len(items), tidyEmbedBatch) + if firstReqs != want { + t.Fatalf("第一轮整理召回请求数 = %d, want %d", firstReqs, want) + } + if ledger.embedding != 1 || ledger.lexical != 0 { + t.Fatalf("ledger = %+v, want 一次 embedding 召回、零次降级", ledger) + } + + base = srv.count() + if _, err := tidier.recallPairs(ctx, partition, items, &TidyStats{}, tidyBudget{}, &recallLedger{}); err != nil { + t.Fatalf("recallPairs(round 2): %v", err) + } + secondReqs := srv.count() - base + if secondReqs != 0 { + t.Fatalf("第二轮整理召回请求数 = %d, want 0（增量向量化没生效）", secondReqs) + } + t.Logf("整理轮 recallPairs 请求数：第一轮 %d → 第二轮 %d（%d 条）", firstReqs, secondReqs, len(items)) +} + +// TestReviewRecall_SecondReviewZeroRequests 是入库审查的生产路径断言：新条由写入 +// 侧落库时已建好向量，因此**第二次审查同一分区是零请求**（候选都从索引取）。 +// +// 这把 M15 那条「每条审查 26 个请求」压到「第一条审查 ⌈K/批⌉、之后 0」。 +func TestReviewRecall_SecondReviewZeroRequests(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + reviewItems(t, store, 20) + + srv := newFakeEmbedServer(t) + emb := srv.embedder() + store.attachVectorIndex(fakeIndex(t, emb)) + reviewer := newTestReviewer(t, store, emb, 0.80) + + first := addItem(t, store, MemoryItem{ + PersonaID: "persona.a", Scope: ScopePersona, Kind: KindRelationship, + Content: "新写入记忆 第一条", + }) + base := srv.count() + if _, degraded, err := reviewer.recallCandidates(ctx, first); err != nil || degraded { + t.Fatalf("recallCandidates(first) = (degraded=%v, err=%v), want (false, nil)", degraded, err) + } + firstReqs := srv.count() - base + if want := batchRequests(20, tidyEmbedBatch); firstReqs != want { + t.Fatalf("第一条审查的请求数 = %d, want %d（只有 20 条既有候选缺向量）", firstReqs, want) + } + + second := addItem(t, store, MemoryItem{ + PersonaID: "persona.a", Scope: ScopePersona, Kind: KindRelationship, + Content: "新写入记忆 第二条", + }) + base = srv.count() + if _, degraded, err := reviewer.recallCandidates(ctx, second); err != nil || degraded { + t.Fatalf("recallCandidates(second) = (degraded=%v, err=%v), want (false, nil)", degraded, err) + } + secondReqs := srv.count() - base + if secondReqs != 0 { + t.Fatalf("同分区第二次审查的请求数 = %d, want 0（增量为空 ⇒ 一个请求都不该发）", secondReqs) + } + t.Logf("入库审查请求数：修复前 26（201÷8）→ 第一条 %d、第二条 %d", firstReqs, secondReqs) +} diff --git a/internal/runtime/memory/embed_isolation.go b/internal/runtime/memory/embed_isolation.go new file mode 100644 index 0000000..70283cc --- /dev/null +++ b/internal/runtime/memory/embed_isolation.go @@ -0,0 +1,109 @@ +package memory + +import ( + "context" + "errors" + "fmt" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding" +) + +// 本文件是嵌入链对**单个慢分片**的韧性（design/23 §15.3 登记的遗留，2026-10-13）： +// `embedding.Embed` 的语义是「任一分片失败 ⇒ 整条调用失败」，于是一条慢分片会把 +// 其余已经算好的分片一起丢掉 —— 审查因此整条降级到词项余弦（92% 的 +// `recall_mode` 为空的一半成因），整理轮同理。 +// +// 修法**不在** embedding 包里改语义（那是四个调用方共享的口径，改它等于改所有 +// 人的预算推演），而是在调用方这一层做**逐条隔离**：整批先发一次；失败则把这一批 +// 拆成单条逐条重试 —— 好的照常拿到向量，坏的那几条记缺口，**整条调用不失败**。 +// +// ## 为什么是「逐条」而不是递归二分 +// +// 二分定位一条坏文本要 1+2+2+2 = 7 个请求（比逐条的 9 少 2 个），但它需要一套 +// 失败预算的簿记（哪一层还允许拆、拆到多深收手），而那套簿记本身就是「韧性变成 +// 放大」的风险点：预算给紧了会把好正文一起丢掉，给松了会在服务挂死时把请求数 +// 放大成 2^深度。逐条隔离的代价**恒定且可算**：一批最多 1+B 个请求（B=8 ⇒ 9）， +// 定位精度与二分相同（精确到单条），没有任何深度参数。 +// +// ## 三条必须同时成立的约束 +// +// 1. **定位**：只有整批失败才逐条重试，因此正常路径（绝大多数批）**零额外请求**； +// 2. **止损**：一条都没拿到 ⇒ 如实返回错误（调用方降级到词项余弦）；有拿到就 +// 只损失坏的那几条 —— 绝不把「算不出来」说成「没有相似」； +// 3. **预算**：父 ctx 已到期时不再逐条重试（预算花完了，重试只会把 worker 拖住）。 +// +// 请求数的上界（B = 批上限，生产 = 8）： +// +// 正常批 1 +// 含一条坏正文 1 + B = 9（定位到它自己） +// 服务整体挂死 1 + B = 9（一次之后 embedMissing 立刻停手，不再发后续批） + +// ErrEmbedUnavailable 在「一条向量都没拿到」时返回：调用方据此走既有的词项余弦 +// 降级（记忆一条不丢、判定规则一字不改），而不是把「算不出来」当成「没有相似」。 +var ErrEmbedUnavailable = errors.New("memory: embedder returned no vector for any input") + +// ErrEmbedCountMismatch 在 embedder 返回的向量条数与入参条数不一致时返回： +// 错位会让相似度与条目错配，因此按失败处理（随后由逐条隔离定位是哪一条出的问题）。 +var ErrEmbedCountMismatch = errors.New("memory: embedder returned a mismatched vector count") + +// embedIsolated 嵌入 texts 并返回与之对齐的向量：失败的位置为 nil，其余照常。 +// +// 返回错误的**唯一**情形是「一个向量都没拿到」（或父 ctx 已到期）；此时调用方按 +// 既有语义降级到词项余弦。 +func embedIsolated(ctx context.Context, emb embedding.Embedder, texts []string) ([]embedding.Vector, error) { + if len(texts) == 0 { + return nil, nil + } + vecs, err := embedOnce(ctx, emb, texts) + if err == nil { + return vecs, nil + } + if len(texts) == 1 { + // 单条失败：这条慢/坏分片只损失它自己。 + return []embedding.Vector{nil}, nil + } + if cerr := ctx.Err(); cerr != nil { + return nil, fmt.Errorf("memory: embed %d texts: %w (%w)", len(texts), err, cerr) + } + out := make([]embedding.Vector, len(texts)) + for i, text := range texts { + if ctx.Err() != nil { + break + } + one, oneErr := embedOnce(ctx, emb, []string{text}) + if oneErr != nil { + // 这一条就是坏的那个分片：留下缺口，别让同一批里的其它条目连坐。 + continue + } + out[i] = one[0] + } + if nonEmptyVectors(out) == 0 { + // 逐条重试也没拿到任何向量 ⇒ 故障是整体性的（服务挂死 / 地址错）， + // 如实报错让调用方降级，而不是交回一批「没有向量」的空结果。 + return nil, err + } + return out, nil +} + +// embedOnce 发一次 Embed，并要求向量条数与入参一致。它是隔离的「一次尝试」。 +func embedOnce(ctx context.Context, emb embedding.Embedder, texts []string) ([]embedding.Vector, error) { + vecs, err := emb.Embed(ctx, texts) + if err != nil { + return nil, err + } + if len(vecs) != len(texts) { + return nil, fmt.Errorf("%w (got %d, want %d)", ErrEmbedCountMismatch, len(vecs), len(texts)) + } + return vecs, nil +} + +// nonEmptyVectors 数一个对齐切片里真正拿到向量的条数。 +func nonEmptyVectors(vecs []embedding.Vector) int { + n := 0 + for _, v := range vecs { + if len(v) > 0 { + n++ + } + } + return n +} diff --git a/internal/runtime/memory/embed_resilience_test.go b/internal/runtime/memory/embed_resilience_test.go new file mode 100644 index 0000000..93f490b --- /dev/null +++ b/internal/runtime/memory/embed_resilience_test.go @@ -0,0 +1,162 @@ +package memory + +import ( + "context" + "fmt" + "testing" +) + +// 本文件是「单分片韧性」的断言（M61，2026-10-30；design/23 §15.3 已登记口径）： +// 一条慢/坏分片**不丢整条**——同一批里其它条目的向量照常返回，只损失那一条； +// 而整体性故障（服务挂死）**不放大**：最多 3 个请求就收手。 + +// TestEmbedIsolation_OneBadTextKeepsTheRest 是韧性的主断言：批里有一条坏正文时， +// 其余 7 条照常拿到向量，坏的那条记为缺口，请求数被二分定位的成本封顶。 +// +// 删掉 embedIsolated（改回「一次 Embed 定生死」）这一条会打红：整批返回错误、 +// 7 条好正文的向量一起丢掉。 +func TestEmbedIsolation_OneBadTextKeepsTheRest(t *testing.T) { + t.Parallel() + ctx := context.Background() + srv := newFakeEmbedServer(t) + srv.failIf = "毒" + emb := srv.embedder() + + texts := make([]string, 0, tidyEmbedBatch) + bad := -1 + for i := 0; i < tidyEmbedBatch; i++ { + text := fmt.Sprintf("正常正文 第%d条", i) + if i == tidyEmbedBatch-1 { + text = "毒 这一条会让整批请求失败" + bad = i + } + texts = append(texts, text) + } + + base := srv.count() + vecs, err := embedIsolated(ctx, emb, texts) + reqs := srv.count() - base + if err != nil { + t.Fatalf("embedIsolated: %v（一条坏分片不该丢掉整条调用）", err) + } + if len(vecs) != len(texts) { + t.Fatalf("向量条数 = %d, want %d（必须与入参对齐）", len(vecs), len(texts)) + } + if len(vecs[bad]) != 0 { + t.Fatalf("坏正文居然拿到了向量：%v", vecs[bad]) + } + kept := 0 + for i := range vecs { + if i == bad { + continue + } + if len(vecs[i]) == 0 { + t.Fatalf("第 %d 条好正文被坏分片连坐了（向量为空）", i) + } + kept++ + } + if kept != len(texts)-1 { + t.Fatalf("保住的条数 = %d, want %d", kept, len(texts)-1) + } + // 逐条隔离的成本上界：整批 1 次 + 逐条重试 B 次（B = 批上限 8）⇒ 9。 + if limit := 1 + tidyEmbedBatch; reqs > limit { + t.Fatalf("定位一条坏分片花了 %d 个请求，超过逐条上界 %d（韧性变成了放大）", reqs, limit) + } + t.Logf("一条坏分片：保住 %d/%d 条，代价 %d 个请求（上界 1+B = %d）", kept, len(texts), reqs, 1+tidyEmbedBatch) +} + +// TestEmbedIsolation_OutageStopsFast 是**反放大**的断言：服务整体挂死时不许把 +// 「用请求换韧性」变成请求风暴 —— 逐条重试一条都没拿到即判定整体性故障，当场 +// 返回错误，且请求数被 1+B 封顶。 +func TestEmbedIsolation_OutageStopsFast(t *testing.T) { + t.Parallel() + ctx := context.Background() + srv := newFakeEmbedServer(t) + srv.failAll = true + emb := srv.embedder() + + texts := make([]string, 0, tidyEmbedBatch) + for i := 0; i < tidyEmbedBatch; i++ { + texts = append(texts, fmt.Sprintf("正文 %d", i)) + } + base := srv.count() + vecs, err := embedIsolated(ctx, emb, texts) + reqs := srv.count() - base + if err == nil { + t.Fatalf("服务整体不可用却返回成功：%v", vecs) + } + if limit := 1 + tidyEmbedBatch; reqs > limit { + t.Fatalf("整体性故障发了 %d 个请求，want ≤ %d（整批 1 + 逐条 B）", reqs, limit) + } + t.Logf("整体性故障：%d 个请求后收手（无放大），错误 = %v", reqs, err) +} + +// TestEmbedIncremental_PartialGapKeepsGoing 是端到端的韧性：24 条里 1 条坏正文， +// 其余 23 条照常带向量返回，统计里如实记 1 个缺口（绝不静默）。 +func TestEmbedIncremental_PartialGapKeepsGoing(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + items := reviewItems(t, store, 24) + + srv := newFakeEmbedServer(t) + srv.failIf = "毒" + emb := srv.embedder() + store.attachVectorIndex(fakeIndex(t, emb)) + + // 把第 9 条换成坏正文（用一条新条目顶掉，保证它也在 missing 里）。 + items[8] = addItem(t, store, MemoryItem{ + PersonaID: "persona.a", Scope: ScopePersona, Kind: KindRelationship, + Content: "毒 这条正文每次请求都失败", + }) + + vecs, stats, err := store.embedIncremental(ctx, emb, items, tidyEmbedBatch) + if err != nil { + t.Fatalf("embedIncremental: %v（一条坏正文不该丢掉整轮召回）", err) + } + if stats.Dropped != 1 { + t.Fatalf("缺口数 = %d, want 1（缺口必须如实计数）", stats.Dropped) + } + if stats.Filled != len(items)-1 { + t.Fatalf("回填数 = %d, want %d", stats.Filled, len(items)-1) + } + if len(vecs[8]) != 0 { + t.Fatalf("坏正文居然拿到了向量") + } + for i := range vecs { + if i == 8 { + continue + } + if len(vecs[i]) == 0 { + t.Fatalf("第 %d 条被坏分片连坐了", i) + } + } +} + +// TestReviewRecall_SelfGapDegradesToLexical 是「被审的那一条自己没向量」的口径： +// 此时语义召回对它无从谈起（相似度恒 0），必须**如实降级**到词项余弦并标记 +// degraded，而不是交出一片看起来「没有重复」的空候选。 +func TestReviewRecall_SelfGapDegradesToLexical(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + reviewItems(t, store, 3) + + srv := newFakeEmbedServer(t) + srv.failIf = "毒" + emb := srv.embedder() + store.attachVectorIndex(fakeIndex(t, emb)) + reviewer := newTestReviewer(t, store, emb, 0.80) + + item := addItem(t, store, MemoryItem{ + PersonaID: "persona.a", Scope: ScopePersona, Kind: KindRelationship, + Content: "毒 被审的这条没有向量", + }) + _, degraded, err := reviewer.recallCandidates(ctx, item) + if err != nil { + t.Fatalf("recallCandidates: %v（缺口该走降级，不该整条失败）", err) + } + if !degraded { + t.Fatal("被审条目没有向量却把自己报成未降级：那会把「算不出来」说成「没有重复」") + } +} diff --git a/internal/runtime/memory/preinsert_config.go b/internal/runtime/memory/preinsert_config.go index 7eb68b2..1b91f94 100644 --- a/internal/runtime/memory/preinsert_config.go +++ b/internal/runtime/memory/preinsert_config.go @@ -60,6 +60,13 @@ const ( // // 🔴 与设计稿的对齐（如实登记）：design/23 §15 那一行原写 `Timeout=20s`（本次改动 // **之前**的口径），已按上面实测同步为 `Timeout=90s`。 + // + // 🆕 2026-10-30（M61）：链长已由**增量向量化**压到「只剩缺向量的那几条」—— + // 同分区第一条审查 ⌈K/批⌉ 个请求、之后 **0 个**（候选从 `vec_items` 取， + // 见 preinsert_recall.go 的 embedReviewTexts 与 embed_incremental.go）。因此上面 + // 的 90s 从「必须覆盖链长」变成「远大于链长的余量」；**不下调**是刻意的： + // 单请求在并发下仍会偶发越过 5s（embedding/budget.go 文件头 ④ 的实测），余量是 + // 韧性而不是浪费 —— 慢分片现在由 embedIsolated 逐条隔离，不吃满整条预算。 defaultReviewTimeout = 90 * time.Second // reviewAuditWho / reviewAuditVerb 是审查留痕的审计口径：它是**系统任务**， // 不是任何角色（design §4.1 C 的同一理由）。 diff --git a/internal/runtime/memory/preinsert_recall.go b/internal/runtime/memory/preinsert_recall.go index 4d9c966..6b54649 100644 --- a/internal/runtime/memory/preinsert_recall.go +++ b/internal/runtime/memory/preinsert_recall.go @@ -2,10 +2,16 @@ package memory import ( "context" + "errors" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding" ) +// ErrReviewSelfUnembedded 在「被审的那一条自己拿不到向量」时返回：语义召回对它 +// 无从谈起（相似度恒 0），因此如实报错让调用方走既有的词项余弦降级，而不是交出 +// 一片看起来「没有重复」的空候选（那会把「算不出来」说成「没有」）。 +var ErrReviewSelfUnembedded = errors.New("memory/review: the item under review could not be embedded") + // 本文件是入库前审查的**候选召回**（design/23 §15 D3-3：用既有 recall 路径， // 不许另造一套相似度实现）。 // @@ -44,7 +50,7 @@ func (r *PreInsertReviewer) recallCandidates(ctx context.Context, item MemoryIte // 不是降级（这是配置形态，不是故障）。 return reviewPairs(items, r.cfg.Threshold, lexicalSimFunc(items), tidyRecallLexical), false, nil } - vecs, err := embedReviewTexts(ctx, r.deps.Embedder, items) + vecs, err := r.embedReviewTexts(ctx, items) if err != nil { // 语义召回不可用 ⇒ 降级到与整理生产轮**同一个**词项余弦，并如实标记 // degraded（design §4.1 生产轮回退口径）。降级绝不改变判定规则， @@ -79,37 +85,33 @@ func reviewPairs(items []MemoryItem, threshold float64, sim func(i, j int) float return out } -// embedReviewTexts 分批把条目向量化（批次上限复用整理侧的 tidyEmbedBatch： -// 不把整个分区的文本一次性交给远端 embedding API）。单批失败即整体失败， -// 由调用方决定降级还是上报，绝不半途改用混合表示。 +// embedReviewTexts 给「新条 1 + 既有候选 K」取向量：**索引里已有的直接复用， +// 只嵌缺的**（判据与回填口径见 embed_incremental.go）。 // // 🔴 链长就是**调用方预算的输入**（2026-10-13 实测，见 embedding/budget.go 文件头）： // 条数是「新条 1 + cfg.ScanLimit（默认 200）」= 201，因此一次召回发出的 HTTP 请求数 -// = ⌈201 / tidyEmbedBatch⌉（tidyEmbedBatch = embedding.DefaultBatchLimit）： +// 曾是 ⌈201 / tidyEmbedBatch⌉（tidyEmbedBatch = embedding.DefaultBatchLimit）： // // 批次 4（改前）= 51 个请求 ⇒ 整条链 23.070 / 21.860 s -// 批次 8（改后）= 26 个请求 ⇒ 整条链 21.435 / 21.597 s +// 批次 8（M15）= 26 个请求 ⇒ 整条链 21.435 / 21.597 s +// 增量（本切片）= **1 个请求**（只有刚落库的新条没向量；既有候选从索引取）， +// 同一分区第二次审查起 = **0 个请求**（新条落库时写入侧已建好向量） // -// 两点必须一起读：① 请求数减半是本改动**唯一**买到的收益 —— 链路墙钟由 -// 「条数 × 每条成本」主导，两个批次形状都是 ~21s；② 所以单条审查的预算 -// （memory.defaultReviewTimeout）必须按**整条链**算，20s 的旧值比链长还短， -// 会让每一条候选满额的审查都超时（这正是「92% 的审查 recall_mode 为空」的根因）。 -func embedReviewTexts(ctx context.Context, emb embedding.Embedder, items []MemoryItem) ([]embedding.Vector, error) { - out := make([]embedding.Vector, 0, len(items)) - for start := 0; start < len(items); start += tidyEmbedBatch { - end := min(start+tidyEmbedBatch, len(items)) - texts := make([]string, 0, end-start) - for _, it := range items[start:end] { - texts = append(texts, it.Content) - } - vecs, err := emb.Embed(ctx, texts) - if err != nil { - return nil, err - } - if len(vecs) != len(texts) { - return nil, ErrTidyEmbedCount - } - out = append(out, vecs...) +// ⇒ 单条审查的预算（memory.defaultReviewTimeout = 90s）因此从「按链长算、勉强够」 +// 变成「按链长算、余量极大」。预算本身**不下调**：慢分片与并发仍会拉长单请求， +// 余量是韧性而不是浪费（见 preinsert_config.go 的推导）。 +// +// 返回切片恒与 items 等长（缺口为 nil）。被审的那一条（items[0]）拿不到向量时 +// 返回错误：语义召回对它无从谈起（相似度恒 0），交给调用方走既有的词项余弦降级， +// 而不是交出一片空候选。 +func (r *PreInsertReviewer) embedReviewTexts(ctx context.Context, items []MemoryItem) ([]embedding.Vector, error) { + vecs, stats, err := r.deps.Store.embedIncremental(ctx, r.deps.Embedder, items, tidyEmbedBatch) + if err != nil { + return nil, err } - return out, nil + logEmbedReuse(r.log, "review", stats, tidyEmbedBatch) + if len(vecs) == 0 || len(vecs[0]) == 0 { + return nil, ErrReviewSelfUnembedded + } + return vecs, nil } diff --git a/internal/runtime/memory/retrieve_m4_test.go b/internal/runtime/memory/retrieve_m4_test.go index 114d377..7ca286a 100644 --- a/internal/runtime/memory/retrieve_m4_test.go +++ b/internal/runtime/memory/retrieve_m4_test.go @@ -26,6 +26,11 @@ func (m *mockVectorIndex) Dim() int { func (m *mockVectorIndex) Add(_ context.Context, _ string, _ vector.Vector) error { return nil } +// Vectors 默认一律未命中：这些用例验的是检索/写入接线，不是增量复用。 +func (m *mockVectorIndex) Vectors(_ context.Context, ids []string) ([]vector.Vector, error) { + return make([]vector.Vector, len(ids)), nil +} + func (m *mockVectorIndex) Close() error { return nil } func (m *mockVectorIndex) Embed(ctx context.Context, texts []string) ([]vector.Vector, error) { diff --git a/internal/runtime/memory/tidy_config.go b/internal/runtime/memory/tidy_config.go index e23aa5c..ce0356b 100644 --- a/internal/runtime/memory/tidy_config.go +++ b/internal/runtime/memory/tidy_config.go @@ -87,9 +87,11 @@ const ( // 同样的分区 500 → **250** 个请求、审查链 51 → **26** 个请求，而单请求实测最大 // 1.782s，仍在 5s 之内（16 条/片已复现出 11.043s 的越界单请求，故不取更大）。 // - // ⚠️ 但它**不缩短整理轮的墙钟**：链路耗时由「条数 × 每条成本」主导，批次只改请求 + // ⚠️ 它**不缩短整理轮的墙钟**：链路耗时由「条数 × 每条成本」主导，批次只改请求 // 数（实测 201 条在 4 条/片与 8 条/片下都是 ~21s）。「整理轮每轮把整个分区重新 - // 向量化」这个**放大源仍然存在**（user 分区一轮 250 个请求），本切片不动它。 + // 向量化」这个放大源由**增量向量化**收口（embed_incremental.go，2026-10-30）： + // 索引里已有向量的行不再重嵌、算出来的向量回填索引，因此同一分区**第二轮起** + // 的请求数趋近 0（第一轮仍按 ⌈条数/批⌉ 发，那是把向量算出来的一次性成本）。 // // 批上限在 embedder 内部**还有一道强制**（Embed 自己切片），因此这里即使被改大 // 也不会把服务端顶住；这一行只是让调用方别去构造注定被切开的超大切片。 diff --git a/internal/runtime/memory/tidy_recall.go b/internal/runtime/memory/tidy_recall.go index aed1c2b..c462cd2 100644 --- a/internal/runtime/memory/tidy_recall.go +++ b/internal/runtime/memory/tidy_recall.go @@ -256,26 +256,24 @@ func vectorSimFunc(vecs []embedding.Vector) func(i, j int) float64 { } } -// embedItems 分批把分区内条目向量化。单批失败即整体失败（调用方决定回退到 -// 词项余弦还是如实上报），不半途改用混合表示 —— 混用会让召回范围不可预测。 +// embedItems 给分区内条目取向量：**索引里已有的直接复用，只嵌缺的** +// （判据与回填口径见 embed_incremental.go）。 +// +// 2026-10-13 M15 登记的两条放大源在这里收口： +// - 「整理轮每轮把整个分区重新向量化」（user 分区一轮 ≈250 个请求）⇒ +// 第二轮起趋近 0（第一轮把算出来的向量回填索引）； +// - 「一个分片失败 ⇒ 整条 Embed 失败」⇒ 逐条隔离（embedIsolated）， +// 一条慢/坏分片只损失它自己，其余条目照常带向量参与比对。 +// +// 返回切片恒与 items 等长，缺口位置为 nil（相似度按 0 计 ⇒ 那一对不进候选）。 +// 一条向量都拿不到时返回错误，调用方按既有语义降级到词项余弦。 func (t *Tidier) embedItems(ctx context.Context, items []MemoryItem) ([]embedding.Vector, error) { - out := make([]embedding.Vector, 0, len(items)) - for start := 0; start < len(items); start += tidyEmbedBatch { - end := min(start+tidyEmbedBatch, len(items)) - texts := make([]string, 0, end-start) - for _, it := range items[start:end] { - texts = append(texts, it.Content) - } - vecs, err := t.deps.Embedder.Embed(ctx, texts) - if err != nil { - return nil, err - } - if len(vecs) != len(texts) { - return nil, ErrTidyEmbedCount - } - out = append(out, vecs...) + vecs, stats, err := t.deps.Store.embedIncremental(ctx, t.deps.Embedder, items, tidyEmbedBatch) + if err != nil { + return nil, err } - return out, nil + logEmbedReuse(t.log, "tidy", stats, tidyEmbedBatch) + return vecs, nil } // lexicalSimilarity 是零依赖的相似度回退：两条内容词项集合的余弦。 diff --git a/internal/runtime/memory/vector/index.go b/internal/runtime/memory/vector/index.go index 4d11bf1..7292708 100644 --- a/internal/runtime/memory/vector/index.go +++ b/internal/runtime/memory/vector/index.go @@ -46,12 +46,19 @@ type Hit struct { // Index 是建立在已存 (id, vector) 对上的稠密向量相似度索引。 // // Embed 用接入的 embedder 把一批文本转成向量。Add -// 持久化一个 (id, vector) 对，id 重复时 upsert。Search 返回与 q +// 持久化一个 (id, vector) 对，id 重复时 upsert。Vectors 按 id 取回 +// 已存向量（增量向量化的读口）。Search 返回与 q // 最相似的前 topK 个 id，按相似度降序排列。Dim 报告 // 固定向量维度。Close 释放底层数据库。 type Index interface { Embed(ctx context.Context, texts []string) ([]Vector, error) Add(ctx context.Context, id string, v Vector) error + // Vectors 返回与 ids 对齐的已存向量：命中按位置给出，未命中（含维度与 + // 本索引不符的行）为 nil。ids 为空时返回同长度的空切片。 + // + // 它是「增量向量化」的读口（memory 包的 embedIncremental）：调用方据此 + // 只嵌缺的那些，而不是每轮把整个分区重新算一遍。它只读，不写任何行。 + Vectors(ctx context.Context, ids []string) ([]Vector, error) Search(ctx context.Context, q Vector, topK int) ([]Hit, error) Dim() int Close() error diff --git a/internal/runtime/memory/vector/lookup.go b/internal/runtime/memory/vector/lookup.go new file mode 100644 index 0000000..1968406 --- /dev/null +++ b/internal/runtime/memory/vector/lookup.go @@ -0,0 +1,68 @@ +package vector + +import ( + "strconv" + "strings" +) + +// 本文件是「按 id 取已存向量」的共用件：增量向量化（memory 包的 +// embedIncremental）靠它跳过已经算过的行，从而不再每轮把整个分区重新嵌一遍 +// ——那是 M15（2026-10-13）登记的请求放大源之一。 +// +// 只读、参数化：id 一律走绑定参数，查询里不出现任何拼接的取值。 + +// vectorsQueryChunk 是单条 IN 查询携带的 id 上限。SQLite 的绑定变量上限 +// （历史默认 999）之下留出余量，使 2000 条的分区不会因为一条查询太长而 +// 失败；超过它的 id 分多条查询，结果仍按入参顺序对齐。 +const vectorsQueryChunk = 256 + +// placeholders 返回 n 个占位符：encode 决定形式（SQLite 的 "?"、PostgreSQL +// 的 "$i"）。它只拼占位符，不拼任何取值。 +func placeholders(n int, encode func(i int) string) string { + var b strings.Builder + for i := 0; i < n; i++ { + if i > 0 { + b.WriteByte(',') + } + b.WriteString(encode(i)) + } + return b.String() +} + +// sqlitePlaceholders 返回 "?,?,?"。 +func sqlitePlaceholders(n int) string { + return placeholders(n, func(int) string { return "?" }) +} + +// pgPlaceholders 返回 "$1,$2,..."。 +func pgPlaceholders(n int) string { + return placeholders(n, func(i int) string { return "$" + strconv.Itoa(i+1) }) +} + +// idArgs 把 ids 转成绑定参数（[]any 是 database/sql 要求的入参形态）。 +func idArgs(ids []string) []any { + args := make([]any, len(ids)) + for i, id := range ids { + args[i] = id + } + return args +} + +// alignVectors 把「id → 向量」按入参顺序摊开：未命中的位置（含同一 id 在 +// 入参里出现多次时）一律为 nil。因此调用方拿到的切片长度恒等于 ids。 +func alignVectors(ids []string, found map[string]Vector) []Vector { + out := make([]Vector, len(ids)) + if len(found) == 0 { + return out + } + pos := make(map[string][]int, len(ids)) + for i, id := range ids { + pos[id] = append(pos[id], i) + } + for id, vec := range found { + for _, i := range pos[id] { + out[i] = vec + } + } + return out +} diff --git a/internal/runtime/memory/vector/pg.go b/internal/runtime/memory/vector/pg.go index 14fcced..c6bdfa6 100644 --- a/internal/runtime/memory/vector/pg.go +++ b/internal/runtime/memory/vector/pg.go @@ -147,6 +147,56 @@ func (p *PGVectorIndex) Add(ctx context.Context, id string, v Vector) error { return nil } +// Vectors 返回与 ids 对齐的已存向量：命中按位置给出，未命中（含维度与 +// 本索引不符的行、解析不了的行）为 nil。它是「增量向量化」的读口 —— +// 调用方据此只嵌缺的那些，而不是每轮重算整个分区。它只读，不写任何行。 +// +// 向量列显式 ::text 取出：pgvector 的 OID 不在 database/sql 的标准类型里， +// 直接 scan 进 string 依赖驱动猜测格式；显式转型把「取什么格式」写在 SQL 里。 +func (p *PGVectorIndex) Vectors(ctx context.Context, ids []string) ([]Vector, error) { + if len(ids) == 0 { + return []Vector{}, nil + } + found := make(map[string]Vector, len(ids)) + for start := 0; start < len(ids); start += vectorsQueryChunk { + end := min(start+vectorsQueryChunk, len(ids)) + if err := p.lookupVectors(ctx, ids[start:end], found); err != nil { + return nil, err + } + } + return alignVectors(ids, found), nil +} + +// lookupVectors 查一批 id 并把命中的向量写进 found。维度与本索引不符 +// （或解析不了）的行当作未命中 —— 它不能参与本索引的余弦，重算一次才是对的。 +func (p *PGVectorIndex) lookupVectors(ctx context.Context, ids []string, found map[string]Vector) error { + query := `SELECT id, vector::text FROM vec_items WHERE id IN (` + pgPlaceholders(len(ids)) + `)` + rows, err := p.db.QueryContext(ctx, query, idArgs(ids)...) + if err != nil { + return fmt.Errorf("memory/vector: lookup pg vectors: %w", err) + } + defer func() { _ = rows.Close() }() + + for rows.Next() { + var ( + id string + text string + ) + if err := rows.Scan(&id, &text); err != nil { + return fmt.Errorf("memory/vector: scan pg vector: %w", err) + } + v, err := pgParseVector(text) + if err != nil || len(v) != p.dim { + continue + } + found[id] = v + } + if err := rows.Err(); err != nil { + return fmt.Errorf("memory/vector: iterate pg vectors: %w", err) + } + return nil +} + // Search 返回与 q 最相似的最多 topK 个 id，按余弦相似度 // （1 - 余弦距离）降序排列，由 pgvector 在 SQL 中计算。topK 非正 // 返回空结果；长度与索引维度不一致的查询 @@ -211,3 +261,27 @@ func pgVectorLiteral(v Vector) string { b.WriteByte(']') return b.String() } + +// pgParseVector 把 pgvector 的文本形式 "[a,b,c]" 解析回 Vector —— +// pgVectorLiteral 的逆运算。它拒绝空串与任何非浮点的分量，使损坏的行以 +// 错误暴露（调用方据此把它当未命中），而不是被静默读成一串零向量。 +func pgParseVector(text string) (Vector, error) { + trimmed := strings.TrimSpace(text) + if len(trimmed) < 2 || trimmed[0] != '[' || trimmed[len(trimmed)-1] != ']' { + return nil, fmt.Errorf("memory/vector: invalid pgvector literal of length %d", len(trimmed)) + } + body := trimmed[1 : len(trimmed)-1] + if body == "" { + return Vector{}, nil + } + parts := strings.Split(body, ",") + out := make(Vector, len(parts)) + for i, part := range parts { + f, err := strconv.ParseFloat(strings.TrimSpace(part), 32) + if err != nil { + return nil, fmt.Errorf("memory/vector: parse pgvector component %d: %w", i, err) + } + out[i] = float32(f) + } + return out, nil +} diff --git a/internal/runtime/memory/vector/sqlite.go b/internal/runtime/memory/vector/sqlite.go index 4c0ba38..dcae10a 100644 --- a/internal/runtime/memory/vector/sqlite.go +++ b/internal/runtime/memory/vector/sqlite.go @@ -108,6 +108,53 @@ func (s *SQLiteIndex) Add(ctx context.Context, id string, v Vector) error { return nil } +// Vectors 返回与 ids 对齐的已存向量：命中按位置给出，未命中（含维度与 +// 本索引不符的行）为 nil。它是「增量向量化」的读口 —— 调用方据此只嵌 +// 缺的那些，而不是每轮重算整个分区。它只读，不写任何行。 +func (s *SQLiteIndex) Vectors(ctx context.Context, ids []string) ([]Vector, error) { + if len(ids) == 0 { + return []Vector{}, nil + } + found := make(map[string]Vector, len(ids)) + for start := 0; start < len(ids); start += vectorsQueryChunk { + end := min(start+vectorsQueryChunk, len(ids)) + if err := s.lookupVectors(ctx, ids[start:end], found); err != nil { + return nil, err + } + } + return alignVectors(ids, found), nil +} + +// lookupVectors 查一批 id 并把命中的向量写进 found。维度与本索引不符的行 +// 当作未命中（它不能参与本索引的余弦，重算一次才是对的）。 +func (s *SQLiteIndex) lookupVectors(ctx context.Context, ids []string, found map[string]Vector) error { + query := `SELECT id, vector FROM vec_items WHERE id IN (` + sqlitePlaceholders(len(ids)) + `)` + rows, err := s.db.QueryContext(ctx, query, idArgs(ids)...) + if err != nil { + return fmt.Errorf("memory/vector: lookup vectors: %w", err) + } + defer func() { _ = rows.Close() }() + + for rows.Next() { + var ( + id string + blob []byte + ) + if err := rows.Scan(&id, &blob); err != nil { + return fmt.Errorf("memory/vector: scan vector: %w", err) + } + v, err := decodeVector(blob) + if err != nil || len(v) != s.dim { + continue + } + found[id] = v + } + if err := rows.Err(); err != nil { + return fmt.Errorf("memory/vector: iterate vectors: %w", err) + } + return nil +} + // Search 返回与 q 最相似的最多 topK 个 id，按余弦相似度 // 降序排列。它在 Go 侧扫描每个已持久化向量（线性扫描，无 ANN）。topK // 非正返回空结果；长度与 diff --git a/internal/runtime/memory/vector/vectors_test.go b/internal/runtime/memory/vector/vectors_test.go new file mode 100644 index 0000000..c4a2914 --- /dev/null +++ b/internal/runtime/memory/vector/vectors_test.go @@ -0,0 +1,197 @@ +package vector + +import ( + "context" + "testing" +) + +// 本文件验「按 id 取已存向量」这个读口（增量向量化的判据：向量存在性 + 维度）。 +// 它是 M61（2026-10-30）用来把「整理轮每轮全量重嵌」变成「只嵌缺的那些」的那一层， +// 因此这里把它自己的对齐/未命中/维度不符/分块行为逐条钉死。 + +// TestSQLite_Vectors_按入参顺序对齐 覆盖命中、未命中、重复 id 三种位置。 +func TestSQLite_Vectors_按入参顺序对齐(t *testing.T) { + t.Parallel() + ctx := context.Background() + s := newInMemorySQLite(t, 3) + for _, e := range []struct { + id string + v Vector + }{ + {id: "cat", v: Vector{1, 0, 0}}, + {id: "dog", v: Vector{0, 1, 0}}, + } { + if err := s.Add(ctx, e.id, e.v); err != nil { + t.Fatalf("Add(%s): %v", e.id, err) + } + } + + got, err := s.Vectors(ctx, []string{"dog", "missing", "cat", "dog"}) + if err != nil { + t.Fatalf("Vectors: %v", err) + } + if len(got) != 4 { + t.Fatalf("Vectors 返回 %d 条, want 4（必须与入参等长）", len(got)) + } + if !equalVector(got[0], Vector{0, 1, 0}) { + t.Fatalf("got[0] = %v, want dog 的向量", got[0]) + } + if len(got[1]) != 0 { + t.Fatalf("未命中的 id 返回了向量：%v", got[1]) + } + if !equalVector(got[2], Vector{1, 0, 0}) { + t.Fatalf("got[2] = %v, want cat 的向量", got[2]) + } + if !equalVector(got[3], Vector{0, 1, 0}) { + t.Fatalf("重复 id 的第 4 个位置 = %v, want 与第 1 个位置相同", got[3]) + } +} + +// TestSQLite_Vectors_空入参 返回同长度的空切片（调用方可以无条件 range）。 +func TestSQLite_Vectors_空入参(t *testing.T) { + t.Parallel() + s := newInMemorySQLite(t, 2) + got, err := s.Vectors(context.Background(), nil) + if err != nil { + t.Fatalf("Vectors(nil): %v", err) + } + if got == nil || len(got) != 0 { + t.Fatalf("Vectors(nil) = %v, want 空切片", got) + } +} + +// TestSQLite_Vectors_维度不符当未命中：索引维度是硬门，别的维度的行不能参与本 +// 索引的余弦，因此它必须被当成「没有向量」（于是调用方会重算一次）。 +func TestSQLite_Vectors_维度不符当未命中(t *testing.T) { + t.Parallel() + ctx := context.Background() + s := newInMemorySQLite(t, 3) + // 绕开 Add 的维度校验，直接写一条 2 维的行（模拟换维度之后留下的旧行）。 + if _, err := s.db.ExecContext(ctx, + `INSERT INTO vec_items (id, vector, dim) VALUES (?, ?, ?)`, + "legacy", encodeVector(Vector{1, 0}), 2); err != nil { + t.Fatalf("insert legacy row: %v", err) + } + + got, err := s.Vectors(ctx, []string{"legacy"}) + if err != nil { + t.Fatalf("Vectors: %v", err) + } + if len(got) != 1 || len(got[0]) != 0 { + t.Fatalf("got = %v, want 单个未命中（维度不符的行不能当命中）", got) + } + + // 损坏的 blob 同样当未命中，绝不静默读成零向量。 + if _, err := s.db.ExecContext(ctx, + `INSERT INTO vec_items (id, vector, dim) VALUES (?, ?, ?)`, + "broken", []byte{1, 2, 3}, 3); err != nil { + t.Fatalf("insert broken row: %v", err) + } + got, err = s.Vectors(ctx, []string{"broken"}) + if err != nil { + t.Fatalf("Vectors(broken): %v", err) + } + if len(got[0]) != 0 { + t.Fatalf("损坏 blob 被当成了向量：%v", got[0]) + } +} + +// TestSQLite_Vectors_超过单查询分块上限 让 id 数越过 vectorsQueryChunk： +// 结果仍必须与入参逐位对齐（分块只是 SQL 的绑定变量上限，不是语义）。 +func TestSQLite_Vectors_超过单查询分块上限(t *testing.T) { + t.Parallel() + ctx := context.Background() + s := newInMemorySQLite(t, 2) + ids := make([]string, 0, vectorsQueryChunk+8) + want := make(map[string]Vector, vectorsQueryChunk+8) + for i := 0; i < vectorsQueryChunk+8; i++ { + id := "id-" + string(rune('a'+i%26)) + "-" + itoaTest(i) + v := Vector{float32(i % 7), 1} + if err := s.Add(ctx, id, v); err != nil { + t.Fatalf("Add(%s): %v", id, err) + } + ids = append(ids, id) + want[id] = v + } + + got, err := s.Vectors(ctx, ids) + if err != nil { + t.Fatalf("Vectors: %v", err) + } + if len(got) != len(ids) { + t.Fatalf("返回 %d 条, want %d", len(got), len(ids)) + } + for i, id := range ids { + if !equalVector(got[i], want[id]) { + t.Fatalf("第 %d 条（%s）= %v, want %v", i, id, got[i], want[id]) + } + } +} + +// TestPGParseVector 验 pgvector 文本形式的解析（与 pgVectorLiteral 互逆）。 +func TestPGParseVector(t *testing.T) { + t.Parallel() + for _, tt := range []struct { + name string + text string + want Vector + }{ + {name: "single", text: "[1]", want: Vector{1}}, + {name: "three", text: "[1.5,2,-0.25]", want: Vector{1.5, 2, -0.25}}, + {name: "empty", text: "[]", want: Vector{}}, + } { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + got, err := pgParseVector(tt.text) + if err != nil { + t.Fatalf("pgParseVector(%q): %v", tt.text, err) + } + if !equalVector(got, tt.want) { + t.Fatalf("pgParseVector(%q) = %v, want %v", tt.text, got, tt.want) + } + }) + } + + for _, bad := range []string{"", "1,2,3", "[1,2", "[a,b]", "[1,,2]"} { + if _, err := pgParseVector(bad); err == nil { + t.Fatalf("pgParseVector(%q) 没报错（损坏行必须暴露）", bad) + } + } + + // 往返：Vector → 文本 → Vector 必须一致（精度按 float32 取）。 + original := Vector{0.5, -1, 3.25, 1024} + round, err := pgParseVector(pgVectorLiteral(original)) + if err != nil { + t.Fatalf("round-trip parse: %v", err) + } + if !equalVector(round, original) { + t.Fatalf("round-trip = %v, want %v", round, original) + } +} + +// equalVector 比较两个向量（nil 与空视为相等）。 +func equalVector(a, b Vector) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} + +// itoaTest 是测试里的小整数格式化（避免为一处用法引入 strconv 的第二种写法）。 +func itoaTest(i int) string { + if i == 0 { + return "0" + } + out := make([]byte, 0, 4) + for i > 0 { + out = append([]byte{byte('0' + i%10)}, out...) + i /= 10 + } + return string(out) +}