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 个请求后收手
164 lines
5.4 KiB
Go
164 lines
5.4 KiB
Go
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
|
||
}
|