Merge branch 'feat/embfix-embed-slice-dedup-and-tidy-increm' into dev
This commit is contained in:
AskaEth committed
commit
a8bc2b4470
16 files changed
+1380
-43
No files matched your search
@@ -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
|
||||
}
|
||||
@@ -0,0 +1,285 @@
|
||||
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 的**条数**；HTTP 请求数 ≥ ⌈Sent/批上限⌉ ——
|
||||
// embedding 包还会按字符界（480 字）再切，长正文时请求数更多（见 approxRequests）。
|
||||
Sent int
|
||||
// Dropped 是逐条隔离重试后仍失败、本轮没有向量的条目数（缺口）。
|
||||
Dropped int
|
||||
// Filled 是本轮回填进索引的向量数。
|
||||
Filled int
|
||||
// FillFailed 是回填失败的条数（不影响召回）。
|
||||
FillFailed int
|
||||
}
|
||||
|
||||
// approxRequests 估算本轮发出的 HTTP 请求数：条数 ÷ 批上限（向上取整）。
|
||||
//
|
||||
// 🔴 它是**按条数界算的下限估算**，不是精确值：embedding.Embed 还按字符界
|
||||
// （DefaultBatchChars=480）切片，长正文批会被多切一刀 ⇒ 真实请求数 ≥ 本估算
|
||||
// （短正文、≤480 字/批时两者相等）。它只进日志做观测，不是任何判据。
|
||||
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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -0,0 +1,121 @@
|
||||
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^深度。逐条隔离的代价**恒定且可算**：`embedOnce` 调用 ≤ 1+B
|
||||
// （B = 批上限），定位精度与二分相同（精确到单条），没有任何深度参数。
|
||||
//
|
||||
// ## 三条必须同时成立的约束
|
||||
//
|
||||
// 1. **定位**：只有整批失败才逐条重试，因此正常路径（绝大多数批）**零额外请求**；
|
||||
// 2. **止损**：一条都没拿到 ⇒ 如实返回错误（调用方降级到词项余弦）；有拿到就
|
||||
// 只损失坏的那几条 —— 绝不把「算不出来」说成「没有相似」；
|
||||
// 3. **预算**：父 ctx 已到期时不再逐条重试（预算花完了，重试只会把 worker 拖住）。
|
||||
//
|
||||
// ## HTTP 请求数的上界（必须按**两维**切片算，B = 批上限 = 8）
|
||||
//
|
||||
// `embedding.Embed` 不是「一批一个请求」：它按 budget.go 的**条数界**
|
||||
// （DefaultBatchLimit=8）**与字符界**（DefaultBatchChars=480）同时切片
|
||||
// （chunkTexts），因此一次 `embedOnce(B 条)` 会发出 C 个请求，C 由正文长度决定：
|
||||
// 全批 ≤480 字 ⇒ C=1；均长 75 汉字（budget.go 的实测口径）时 8×75=600 > 480 ⇒ C=2；
|
||||
// 每条自己就超过 480 字 ⇒ 每条独占一片 ⇒ C=B（病理上限）。
|
||||
// 单条重试**永不切分**（它自己就是一片）⇒ 每条恰好 1 个请求。于是：
|
||||
//
|
||||
// 正常批 C（短正文 C=1；长正文 2 起）
|
||||
// 含一条坏正文 首次失败发生在第 1~C 个 chunk 之一，随后逐条 B 次
|
||||
// ⇒ ≤ C + B（短正文 9；现实长正文 10；病理上限 2B=16）
|
||||
// 服务整体挂死 首个 chunk 即失败 + 逐条 B 次 ⇒ **1 + B = 9**（与字符维无关）
|
||||
//
|
||||
// 口径与 budget.go 同源：那里的「建模上界 ≠ 延迟保证」是同一类提醒 —— 这里的
|
||||
// 「≤1+B」是**embedOnce 调用次数**的上界，HTTP 请求数必须再按字符维的 chunk 数算
|
||||
// （正常批 C、含坏正文 ≤C+B）；只有「服务整体挂死 = 1+B」这一条与字符维无关。
|
||||
|
||||
// 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
|
||||
}
|
||||
@@ -0,0 +1,170 @@
|
||||
package memory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// 本文件是「单分片韧性」的断言（M61，2026-10-30；design/23 §15.3 已登记口径）：
|
||||
// 一条慢/坏分片**不丢整条**——同一批里其它条目的向量照常返回，只损失那一条；
|
||||
// 而整体性故障（服务挂死）**不放大**：整批在第一个 chunk 即失败 + 逐条 B 次
|
||||
// ⇒ 1+B 个请求就收手（B = 批上限 8）。
|
||||
//
|
||||
// ⚠️ 请求数口径是**两维**的（见 embed_isolation.go 文件头）：本文件的正文都很短
|
||||
// （8 条合计远小于 DefaultBatchChars=480）⇒ chunkTexts 只按条数界切、C=1，因此
|
||||
// 「含一条坏正文 = 1+B」在这些用例里是精确值；长正文批的真实上界是 C+B。
|
||||
|
||||
// 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 个 chunk（本例正文短 ⇒ C=1）+ 逐条重试 B 次
|
||||
// ⇒ 1+B = 9。长正文批的真实上界是 C+B（C 由 480 字字符界决定，见文件头）。
|
||||
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 封顶（整批在第一个 chunk 就失败 ⇒ 1 个请求 + 逐条 B；
|
||||
// 这一条与字符维的 chunk 数无关，本例正文短、C 本来就是 1）。
|
||||
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("被审条目没有向量却把自己报成未降级：那会把「算不出来」说成「没有重复」")
|
||||
}
|
||||
}
|
||||
@@ -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 的同一理由）。
|
||||
|
||||
@@ -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)
|
||||
// ⇒ 单条审查的预算（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
|
||||
}
|
||||
if len(vecs) != len(texts) {
|
||||
return nil, ErrTidyEmbedCount
|
||||
logEmbedReuse(r.log, "review", stats, tidyEmbedBatch)
|
||||
if len(vecs) == 0 || len(vecs[0]) == 0 {
|
||||
return nil, ErrReviewSelfUnembedded
|
||||
}
|
||||
out = append(out, vecs...)
|
||||
}
|
||||
return out, nil
|
||||
return vecs, nil
|
||||
}
|
||||
@@ -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) {
|
||||
|
||||
@@ -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 自己切片），因此这里即使被改大
|
||||
// 也不会把服务端顶住；这一行只是让调用方别去构造注定被切开的超大切片。
|
||||
|
||||
@@ -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)
|
||||
vecs, stats, err := t.deps.Store.embedIncremental(ctx, t.deps.Embedder, items, tidyEmbedBatch)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(vecs) != len(texts) {
|
||||
return nil, ErrTidyEmbedCount
|
||||
}
|
||||
out = append(out, vecs...)
|
||||
}
|
||||
return out, nil
|
||||
logEmbedReuse(t.log, "tidy", stats, tidyEmbedBatch)
|
||||
return vecs, nil
|
||||
}
|
||||
|
||||
// lexicalSimilarity 是零依赖的相似度回退：两条内容词项集合的余弦。
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -89,4 +89,23 @@ func TestPG_RequiresEnv(t *testing.T) {
|
||||
if len(vs) != 1 || len(vs[0]) != 4 {
|
||||
t.Fatalf("Embed => %d vectors, want one 4-dim", len(vs))
|
||||
}
|
||||
|
||||
// Vectors（增量向量化的读口）：命中按位置给出，未命中为 nil ——
|
||||
// 这条断言顺带验 vector::text 的解析在真实 pgvector 上成立。
|
||||
got, err := idx.Vectors(ctx, []string{"b", "missing", "a"})
|
||||
if err != nil {
|
||||
t.Fatalf("Vectors: %v", err)
|
||||
}
|
||||
if len(got) != 3 {
|
||||
t.Fatalf("Vectors 返回 %d 条, want 3", len(got))
|
||||
}
|
||||
if !equalVector(got[0], Vector{0, 1, 0, 0}) {
|
||||
t.Fatalf("Vectors[0] = %v, want b 的向量", got[0])
|
||||
}
|
||||
if len(got[1]) != 0 {
|
||||
t.Fatalf("未命中的 id 返回了向量：%v", got[1])
|
||||
}
|
||||
if !equalVector(got[2], Vector{1, 0, 0, 0}) {
|
||||
t.Fatalf("Vectors[2] = %v, want a 的向量", got[2])
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
// 非正返回空结果；长度与
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
Reference in new issue
Block a user