fix(memory): embedding 批上限与超时同源 —— 不再「客户端先放弃、服务端继续算」
M9(EMBRECALL)。故障机制(M6 + 本次复测):客户端 5s 单次超时先放弃,服务端仍在 算那一批 ⇒ 本机 llama.cpp 被自己顶住 ⇒ 后续每个请求(含写向量索引的单条请求) 排队到自己的超时 ⇒ 召回全线降级。 - embedding/budget.go(新):把「一次请求的耗时上界 ≤ 单次超时的一半」写成不变式, 批上限(条数 4 / 字符 480)与 5s 超时由**同一个不等式**解出;实测依据与复跑命令 写进注释(空闲态摊销曲线 1→0.19s / 8→1.1s / 12→1.3s / 64→8.5s,约 50ms 固定 + 2.7~3.9ms/字符,n≥32 起有数倍放大 ⇒ 余量取一半)。 - embedding/api.go:Embed 按同一对上限**自己切片**(不变式与调用方无关);冷启动 首次调用单独给 warmupTimeout,拿到第一个成功响应后立刻回常态超时。 - memory/tidy_config.go:tidyEmbedBatch = embedding.DefaultBatchLimit(同源), 并加编译期断言 —— 调回 64 会直接编译失败(constant -60 overflows uint)。 - memory/tidy.go | tidy_recall.go | tidy_dryrun.go:recall_mode 改报**实际发生** 的来源(""=本轮没候选可比 / embedding / lexical / mixed),新增 recall_degraded_partitions;试算轮 RecallMode 同口径(M6 medium finding)。 - 测试:embedding 侧用「实测曲线 + 会睡够时间的假客户端」钉住两态 (批内 < 超时 ⇒ 不降级;旧配对 64 条 ⇒ 必然超时),并断言 abandoned == 0; memory 侧钉住降级/健康/没候选三态与试算轮的如实上报。 不改 preinsert*、不改 .env、不动判定规则与任何阈值。
This commit is contained in:
1 parent
68ba85389b
commit
59672720bd
9 files changed
+836
-45
No files matched your search
@@ -11,6 +11,7 @@ import (
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
|
||||
@@ -38,23 +39,21 @@ const (
|
||||
// DefaultRequestTimeout 是**单次** /embeddings 调用的墙钟上限,独立于 HTTP 客户端
|
||||
// 那一层的 180s 总超时(httpRequestTimeout)。
|
||||
//
|
||||
// 依据(是实测数据,不是拍脑袋的调优):
|
||||
// - 上层「入库前审查」的单条总预算是 20s(memory.defaultReviewTimeout),embedding
|
||||
// 只是它内部的第一步。2026-10-07 实测(本机 llama.cpp + Qwen3-Embedding-0.6B,
|
||||
// 服务空闲):单条文本约 60ms;**64 条一批约 2.1~2.9s**;审查最坏情况
|
||||
// (ScanLimit=200 + 新条目 1)要 4 批 ⇒ 约 8~10s,已经占掉 20s 预算的一半,
|
||||
// 留给后面的 T1 灰区判定(defaultT1Timeout=8s)的余量很薄。一次卡死的 HTTP
|
||||
// 就能把这点余量连同整条审查一起吃掉(实测日志里
|
||||
// `memory/review: context deadline exceeded` 正是这个形态);
|
||||
// - 所以每次调用必须有**自己的**上限:卡住的那一批到点就失败并降级,不能把整条
|
||||
// 审查的 20s 都赌在一次 HTTP 上;
|
||||
// - 5s 的取值:相对实测的 64 条一批 2.9s 留约 1.7 倍余量,能吸收并发争用带来的
|
||||
// 抖动,同时远小于 20s 的审查预算。若日后批次变大或服务变慢,先调这里而不是
|
||||
// 调审查总预算;
|
||||
// - 真慢就**快速失败**:调用方按**既有语义**降级(审查降级到词项余弦并如实留痕
|
||||
// degraded),**绝不静默丢记忆** —— 本常量只砍等待时间,不改任何判定规则;
|
||||
// - 已知代价:llama.cpp 首次加载模型可能超过 5s,那一次调用会失败并降级,模型
|
||||
// 常驻后的下一次调用即恢复正常。降级本就是既有行为,因此可接受。
|
||||
// 它**不是**单独取值:批次上限(DefaultBatchLimit / DefaultBatchChars)与它由
|
||||
// budget.go 里**同一个不等式**推出 —— 单次请求的耗时上界不得超过超时的一半。
|
||||
// 改这一行就等于改了批次上限,反之亦然;两侧永远是配套的。
|
||||
//
|
||||
// 为什么每次调用必须有自己的上限:上层「入库前审查」的单条总预算是 20s
|
||||
// (memory.defaultReviewTimeout),embedding 只是它内部的第一步,后面还有 T1 灰区
|
||||
// 判定(8s)。一次卡死的 HTTP 不能把整条审查的 20s 都赌掉。真慢就**快速失败**,
|
||||
// 调用方按**既有语义**降级(审查降级到词项余弦并如实留痕 degraded),
|
||||
// **绝不静默丢记忆** —— 本常量只砍等待时间,不改任何判定规则。
|
||||
//
|
||||
// 5s 保持不变的依据:本次故障(2026-10-13)的成因不是这个数太小,而是批次太大 ——
|
||||
// 64 条的实测耗时 7.5~8.5s 已经越过它,客户端放弃、服务端还在算,把整台本机
|
||||
// embedding 服务顶住,连带单条请求(写向量索引)也一起超时。所以修法是**把批次
|
||||
// 收进预算内**,而不是把预算放大到迁就批次(后者会让真挂死的请求也拖住整轮)。
|
||||
// 冷启动那一次另有 warmupTimeout(见 budget.go),它只生效到第一次成功响应为止。
|
||||
const DefaultRequestTimeout = 5 * time.Second
|
||||
|
||||
// openAICompatibleRequest is the JSON body sent to an OpenAI-compatible
|
||||
@@ -93,6 +92,12 @@ type APIEmbedder struct {
|
||||
cfg config.EmbeddingConfig
|
||||
client *http.Client
|
||||
log *slog.Logger
|
||||
// timeout 是**单次**请求的墙钟上限,与批次上限同源(见 budget.go)。
|
||||
// 默认 DefaultRequestTimeout;WithRequestTimeout 可换,换了批次上限也跟着换。
|
||||
timeout time.Duration
|
||||
// warmed 报告是否已经拿到过成功响应。未热之前用 warmupTimeout(模型加载 /
|
||||
// GPU 缓冲分配),热之后一律用 timeout。
|
||||
warmed atomic.Bool
|
||||
|
||||
mu sync.RWMutex
|
||||
dim int
|
||||
@@ -130,12 +135,24 @@ func WithLogger(l *slog.Logger) APIOption {
|
||||
}
|
||||
}
|
||||
|
||||
// WithRequestTimeout 换掉单次请求的墙钟上限(非正数忽略,保留默认)。
|
||||
// 它同时决定批次上限 —— 两者是同一条推演(budget.go),因此这里是**唯一**的
|
||||
// 调优点,不允许调用方各自去改批次。
|
||||
func WithRequestTimeout(d time.Duration) APIOption {
|
||||
return func(e *APIEmbedder) {
|
||||
if d > 0 {
|
||||
e.timeout = d
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// NewAPI builds an APIEmbedder from cfg using the default HTTP client.
|
||||
func NewAPI(cfg config.EmbeddingConfig, opts ...APIOption) *APIEmbedder {
|
||||
e := &APIEmbedder{
|
||||
cfg: cfg,
|
||||
client: defaultHTTPClient(),
|
||||
log: slog.New(slog.NewTextHandler(io.Discard, nil)),
|
||||
cfg: cfg,
|
||||
client: defaultHTTPClient(),
|
||||
log: slog.New(slog.NewTextHandler(io.Discard, nil)),
|
||||
timeout: DefaultRequestTimeout,
|
||||
}
|
||||
for _, opt := range opts {
|
||||
opt(e)
|
||||
@@ -155,8 +172,10 @@ func (e *APIEmbedder) Dim() int {
|
||||
// returned vectors aligned by index. It returns ErrNoAPIURL for a blank URL
|
||||
// and a wrapped error for any HTTP or decode failure.
|
||||
//
|
||||
// 单次调用带独立的、比上层审查总预算(20s)短得多的超时(DefaultRequestTimeout),
|
||||
// 卡住即快速失败,把预算留给后续步骤;失败语义与降级口径不变。
|
||||
// texts 长度不受调用方控制:本方法按 budget.go 推出来的上限(条数 / 字符总量)
|
||||
// **自己切片**,分多次请求顺序发出,因此「一次请求的耗时 < 单次超时」这条不变式
|
||||
// 对任何调用方都成立 —— 调用方给多大的切片都不会把服务端顶住。中途任一段失败即
|
||||
// 整体失败(返回已发请求的错误),绝不半途改用混合表示。
|
||||
func (e *APIEmbedder) Embed(ctx context.Context, texts []string) ([]Vector, error) {
|
||||
if len(texts) == 0 {
|
||||
return []Vector{}, nil
|
||||
@@ -168,7 +187,34 @@ func (e *APIEmbedder) Embed(ctx context.Context, texts []string) ([]Vector, erro
|
||||
return nil, ErrNoAPIURL
|
||||
}
|
||||
|
||||
callCtx, cancel := context.WithTimeout(ctx, DefaultRequestTimeout)
|
||||
// 切片用的上限由**稳态**超时推出(e.timeout),与 requestTimeout() 无关:
|
||||
// 冷启动那一次只是**截止时刻**更宽(warmupTimeout),批次形状不变 —— 否则
|
||||
// 首个请求会按 30s 的预算切出一个巨大的批,正好把冷启动的模型顶得更久。
|
||||
items, chars := batchLimitsFor(e.timeout)
|
||||
out := make([]Vector, 0, len(texts))
|
||||
for _, chunk := range chunkTexts(texts, items, chars) {
|
||||
vectors, err := e.embedChunk(ctx, chunk)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, vectors...)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// requestTimeout 返回本次调用该用的单次超时:拿到第一个成功响应之前用
|
||||
// warmupTimeout(冷启动:模型加载 / GPU 缓冲分配),之后一律用配置的 timeout。
|
||||
func (e *APIEmbedder) requestTimeout() time.Duration {
|
||||
if e.warmed.Load() {
|
||||
return e.timeout
|
||||
}
|
||||
return warmupTimeout
|
||||
}
|
||||
|
||||
// embedChunk 发**一次** /embeddings 请求。chunk 必须已经落在 batchLimitsFor 推出来
|
||||
// 的边界内(Embed 负责切),因此它的耗时上界不超过超时的一半。
|
||||
func (e *APIEmbedder) embedChunk(ctx context.Context, texts []string) ([]Vector, error) {
|
||||
callCtx, cancel := context.WithTimeout(ctx, e.requestTimeout())
|
||||
defer cancel()
|
||||
|
||||
req, err := e.newEmbeddingRequest(callCtx, texts)
|
||||
@@ -186,6 +232,8 @@ func (e *APIEmbedder) Embed(ctx context.Context, texts []string) ([]Vector, erro
|
||||
msg := readErrorBody(resp.Body)
|
||||
return nil, fmt.Errorf("embedding: status %d: %s", resp.StatusCode, msg)
|
||||
}
|
||||
// 服务端确实答了 ⇒ 冷启动这一关过了,后续调用回到常态超时。
|
||||
e.warmed.Store(true)
|
||||
|
||||
data, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes))
|
||||
if err != nil {
|
||||
|
||||
@@ -222,6 +222,10 @@ const reviewBudgetRef = 20 * time.Second
|
||||
// 审查总预算:否则一次卡死的 embedding 连接就能吃光整条审查的预算,排在后面的
|
||||
// T1 灰区判定连执行机会都没有(2026-10-07 实测故障形态)。同时验证该超时确实
|
||||
// 打在了发出去的请求上。
|
||||
//
|
||||
// ⚠️ 冷启动那一次走的是 warmupTimeout(见 budget.go),因此这里**以稳态请求**
|
||||
// (第二次及以后)为准;冷启动那一路由 budget_test.go 的
|
||||
// TestAPIEmbedder_WarmupFirstCallBudget 单独钉。
|
||||
func TestDefaultRequestTimeout(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -256,8 +260,12 @@ func TestDefaultRequestTimeout(t *testing.T) {
|
||||
config.EmbeddingConfig{APIURL: "http://127.0.0.1:11434/v1/embeddings"},
|
||||
WithRoundTripper(rt),
|
||||
)
|
||||
if _, err := emb.Embed(context.Background(), []string{"hi"}); err != nil {
|
||||
t.Fatalf("Embed() error: %v", err)
|
||||
// 第一次是冷启动调用(warmupTimeout),结果由它自己的测试断言;这里连发两次,
|
||||
// 只看稳态那一次打出去的 deadline。
|
||||
for i := 0; i < 2; i++ {
|
||||
if _, err := emb.Embed(context.Background(), []string{"hi"}); err != nil {
|
||||
t.Fatalf("第 %d 次 Embed() error: %v", i+1, err)
|
||||
}
|
||||
}
|
||||
if slack <= 0 || slack > DefaultRequestTimeout {
|
||||
t.Fatalf("请求 deadline 余量 = %s, want (0, %s]", slack, DefaultRequestTimeout)
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
package embedding
|
||||
|
||||
import "time"
|
||||
|
||||
// 本文件是**单次 /embeddings 调用的预算推演**:批次上限与单次超时必须由**同一个
|
||||
// 式子**得出。任一侧单独调整都会让「一次请求的实测耗时 < 单次超时」这条不变式
|
||||
// 静默失效 —— 而失效的后果不是报错,是**服务端还在算、客户端已经放弃**:被放弃的
|
||||
// 那一批仍占着本机 llama.cpp 的算力,后面每一个请求(包括写索引时的单条请求)都
|
||||
// 排在它后面等到自己的超时,于是整条记忆写入路径被拖住、召回全线降级。
|
||||
//
|
||||
// ## 不变式
|
||||
//
|
||||
// requestOverhead + 一次请求的字符总量 × perCharBudget ≤ DefaultRequestTimeout / callBudgetShare
|
||||
//
|
||||
// 左半是**推演出来的耗时上界**,右半是该超时下允许用掉的一半(另一半是这台机器的
|
||||
// 抖动余量)。批次上限由解这个不等式得到(见 DefaultBatchLimit / DefaultBatchChars)。
|
||||
//
|
||||
// ## 实测依据(2026-10-13,本机 = AMD RX 6600 XT + llama.cpp b11371 +
|
||||
// Qwen3-Embedding-0.6B Q8_0,端点 127.0.0.1:11434,**服务器空闲**,正文取自活库
|
||||
// data/cyrene.db:短正文约 33 汉字、长正文约 168 汉字;每档 5~6 次取最坏)
|
||||
//
|
||||
// n=1 短 0.066~0.185s 长 0.740s
|
||||
// n=8 短 0.86~1.11s
|
||||
// n=12 短 1.27s 长 5.47s
|
||||
// n=16 短 2.20~3.31s
|
||||
// n=64 短 7.49~8.49s(改前的批次上限)
|
||||
//
|
||||
// 拟合:单次请求耗时 ≈ 固定开销 + 字符总量 × 每字符成本;固定开销实测约 50ms,
|
||||
// 每字符成本实测 2.7~3.9ms。**同一批 n≥32 起出现过数倍放大**(8.97s / 26.6s),
|
||||
// 因此大批次的耗时不可预测 —— 这也是「余量取一半」而不是「卡着边界」的依据。
|
||||
//
|
||||
// 复跑命令(读活库正文、只连本机 embedding 服务、不写任何库):
|
||||
//
|
||||
// python - <<'PY'
|
||||
// import json,sqlite3,time,urllib.request
|
||||
// URL="http://127.0.0.1:11434/v1/embeddings"; MODEL="<model id>"
|
||||
// t=[r[0] for r in sqlite3.connect("data/cyrene.db").execute(
|
||||
// "select content from memory_items order by id limit 64")]
|
||||
// for b in (t[:1],t[:8],t[:12],t[:16],t[:64]):
|
||||
// body=json.dumps({"model":MODEL,"input":b}).encode()
|
||||
// r=urllib.request.Request(URL,data=body,headers={"Content-Type":"application/json"})
|
||||
// s=time.perf_counter(); urllib.request.urlopen(r,timeout=300).read()
|
||||
// print(len(b), round(time.perf_counter()-s,3))
|
||||
// PY
|
||||
const (
|
||||
// requestOverhead 是单次 HTTP 调用的固定开销上界(实测 ~50ms,向上取整)。
|
||||
requestOverhead = 100 * time.Millisecond
|
||||
|
||||
// perCharBudget 是**每字符**的墙钟预算上界(实测 2.7~3.9 ms/字符,取 5ms)。
|
||||
perCharBudget = 5 * time.Millisecond
|
||||
|
||||
// perItemBudget 是**每条文本**的墙钟预算上界:按活库里最长的正文(168 汉字,
|
||||
// 最坏摊销 0.456 s/条)向上取整到 500ms。
|
||||
perItemBudget = 500 * time.Millisecond
|
||||
|
||||
// callBudgetShare 表示一次请求最多用掉超时的 1/2:另一半是抖动余量
|
||||
// (见上面 n≥32 的数倍放大)。调大它 = 余量变小,必须连带复核实测。
|
||||
callBudgetShare = 2
|
||||
|
||||
// DefaultBatchLimit 是单次请求的**条数**上限,由上面三个常量推出来(唯一推演):
|
||||
// (5s/2 − 100ms) / 500ms = 4。
|
||||
DefaultBatchLimit = (int(DefaultRequestTimeout/callBudgetShare) - int(requestOverhead)) / int(perItemBudget)
|
||||
|
||||
// DefaultBatchChars 是单次请求的**字符总量**上限,同一个不等式换成每字符预算:
|
||||
// (5s/2 − 100ms) / 5ms = 480。它防的是「条数很少但每条很长」:
|
||||
// 4 条 × 168 汉字 = 672 字符已越过它,会被再切一刀。
|
||||
DefaultBatchChars = (int(DefaultRequestTimeout/callBudgetShare) - int(requestOverhead)) / int(perCharBudget)
|
||||
|
||||
// warmupTimeout 是**首次**调用(模型加载 / GPU 缓冲分配尚未发生)的墙钟上限。
|
||||
// 它**只在拿到第一个成功响应之前**生效,之后所有调用立刻回到
|
||||
// DefaultRequestTimeout —— 所以它不会让「真挂死的请求」长期拖住整轮审查,
|
||||
// 又不让冷启动那一次注定失败(2026-10-13 部署后第一分钟生产日志里
|
||||
// `vector embed for index failed ... embed 1 texts ... context deadline exceeded`
|
||||
// 就是这一路:模型还没热,5s 不够)。
|
||||
warmupTimeout = 30 * time.Second
|
||||
)
|
||||
|
||||
// callBudget 返回在 timeout 这个单次超时下,一次请求允许用掉的墙钟预算。
|
||||
func callBudget(timeout time.Duration) time.Duration {
|
||||
if timeout <= 0 {
|
||||
timeout = DefaultRequestTimeout
|
||||
}
|
||||
return timeout/callBudgetShare - requestOverhead
|
||||
}
|
||||
|
||||
// batchLimitsFor 由**同一个超时**推出一对批次上限(条数 / 字符总量)。
|
||||
// 它保证:按 (条数 ≤ items 且 字符总量 ≤ chars) 切出来的任何一段,其耗时上界
|
||||
// 都不超过该超时的一半。两者都至少为 1 —— 单条文本本身越界时独占一段,
|
||||
// 正文**绝不截断**(截断会让向量描述的不是这条记忆)。
|
||||
func batchLimitsFor(timeout time.Duration) (items, chars int) {
|
||||
budget := callBudget(timeout)
|
||||
items = int(budget / perItemBudget)
|
||||
if items < 1 {
|
||||
items = 1
|
||||
}
|
||||
chars = int(budget / perCharBudget)
|
||||
if chars < 1 {
|
||||
chars = 1
|
||||
}
|
||||
return items, chars
|
||||
}
|
||||
|
||||
// chunkTexts 按「条数 ≤ maxItems 且 字符总量 ≤ maxChars」把 texts 切成若干段。
|
||||
// 单条文本的 rune 数超过 maxChars 时它独占一段(见 batchLimitsFor 的口径)。
|
||||
// 空输入返回 nil。
|
||||
func chunkTexts(texts []string, maxItems, maxChars int) [][]string {
|
||||
if len(texts) == 0 {
|
||||
return nil
|
||||
}
|
||||
if maxItems < 1 {
|
||||
maxItems = 1
|
||||
}
|
||||
if maxChars < 1 {
|
||||
maxChars = 1
|
||||
}
|
||||
var out [][]string
|
||||
start := 0
|
||||
chars := 0
|
||||
for i, text := range texts {
|
||||
n := len([]rune(text))
|
||||
// 段内已有内容,再加这一条就越界 ⇒ 先收口,本段从这一条重新开始。
|
||||
if i > start && (i-start >= maxItems || chars+n > maxChars) {
|
||||
out = append(out, texts[start:i])
|
||||
start, chars = i, 0
|
||||
}
|
||||
chars += n
|
||||
}
|
||||
return append(out, texts[start:])
|
||||
}
|
||||
@@ -0,0 +1,310 @@
|
||||
package embedding
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
|
||||
)
|
||||
|
||||
// 本文件把 budget.go 的**不变式**钉成可执行的断言,两态各一条:
|
||||
//
|
||||
// - 「批内耗时 < 超时」⇒ 不降级:DerivedBatchCompletesWithinTimeout;
|
||||
// - 「批内耗时 ≥ 超时」⇒ 超时(本次故障形态):LegacyBatchOf64TimesOut;
|
||||
//
|
||||
// 以及**变异自证**:把批上限调回旧值 64(或把推演调松)会让哪条测试红,
|
||||
// 见 TestBatchBudget_InvariantHolds 与 TestAPIEmbedder_LegacyBatchOf64TimesOut。
|
||||
//
|
||||
// 假客户端用的是**实测曲线**(不是随口编的常数):单次请求耗时
|
||||
// ≈ measuredOverhead + 字符总量 × measuredPerChar,取值见下面两个常量
|
||||
// (2026-10-13 本机实测:固定开销 ~50ms、每字符 2.7~3.9ms,此处取上界)。
|
||||
const (
|
||||
measuredOverhead = 50 * time.Millisecond
|
||||
measuredPerChar = 4 * time.Millisecond
|
||||
latencyModelChars = 33 // 活库短正文的平均长度(实测 33 汉字)
|
||||
)
|
||||
|
||||
// latencyTripper 是一个按上面那条曲线**真的睡够时间**的 RoundTripper:
|
||||
// 它模拟「服务端会算完,不管你客户端还在不在等」。这正是本次故障的形态 ——
|
||||
// 客户端 5s 放弃、服务端还在算,后面每个请求都排在它后面。
|
||||
type latencyTripper struct {
|
||||
mu sync.Mutex
|
||||
requests int
|
||||
maxItems int
|
||||
maxChars int
|
||||
// abandoned 计数「客户端先放弃、服务端还在算」的次数 —— 本次故障的形态。
|
||||
// 修好之后它必须是 0(见 TestAPIEmbedder_OversizedBatchIsSplit 的断言)。
|
||||
abandoned int
|
||||
// firstSleep 非零时,第一次请求额外多睡这么久(模拟冷启动的模型加载)。
|
||||
firstSleep time.Duration
|
||||
}
|
||||
|
||||
func (l *latencyTripper) RoundTrip(r *http.Request) (*http.Response, error) {
|
||||
body, err := io.ReadAll(r.Body)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var req openAICompatibleRequest
|
||||
if err := json.Unmarshal(body, &req); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
chars := 0
|
||||
for _, s := range req.Input {
|
||||
chars += len([]rune(s))
|
||||
}
|
||||
|
||||
l.mu.Lock()
|
||||
l.requests++
|
||||
if len(req.Input) > l.maxItems {
|
||||
l.maxItems = len(req.Input)
|
||||
}
|
||||
if chars > l.maxChars {
|
||||
l.maxChars = chars
|
||||
}
|
||||
first := l.requests == 1
|
||||
extra := l.firstSleep
|
||||
l.mu.Unlock()
|
||||
|
||||
wait := measuredOverhead + time.Duration(chars)*measuredPerChar
|
||||
if first && extra > 0 {
|
||||
wait += extra
|
||||
}
|
||||
select {
|
||||
case <-time.After(wait):
|
||||
case <-r.Context().Done():
|
||||
// 客户端放弃了:服务端那一侧不会因为客户端走了就把已排队的算力还回来
|
||||
// (实测的级联堆积正是从这一步开始的),因此这里明确记一笔。
|
||||
l.mu.Lock()
|
||||
l.abandoned++
|
||||
l.mu.Unlock()
|
||||
return nil, r.Context().Err()
|
||||
}
|
||||
|
||||
resp := openAICompatibleResponse{Data: make([]openAICompatibleDatum, 0, len(req.Input))}
|
||||
for i := range req.Input {
|
||||
resp.Data = append(resp.Data, openAICompatibleDatum{Index: i, Embedding: []float32{1, 0}})
|
||||
}
|
||||
raw, err := json.Marshal(resp)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &http.Response{
|
||||
StatusCode: http.StatusOK,
|
||||
Body: io.NopCloser(strings.NewReader(string(raw))),
|
||||
Header: make(http.Header),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// shortTexts 造 n 条与活库短正文等长的文本。
|
||||
func shortTexts(n int) []string {
|
||||
out := make([]string, n)
|
||||
line := strings.Repeat("河", latencyModelChars)
|
||||
for i := range out {
|
||||
out[i] = line
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// TestBatchBudget_InvariantHolds 是**推演本身**的断言(不连任何网络):
|
||||
// 由超时推出来的一对上限,必须使「一次请求的耗时上界 ≤ 超时的一半」成立。
|
||||
//
|
||||
// 变异自证:把 DefaultBatchLimit 的推演改成旧口径(条数写回 64)⇒ 第一条断言红;
|
||||
// 若改成「把超时放大到盖住 64 条」⇒ 最后一条断言红(旧配对必须仍是越界的)。
|
||||
func TestBatchBudget_InvariantHolds(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
if DefaultBatchLimit < 1 || DefaultBatchChars < 1 {
|
||||
t.Fatalf("推出来的上限必须 ≥ 1:条数=%d 字符=%d", DefaultBatchLimit, DefaultBatchChars)
|
||||
}
|
||||
share := DefaultRequestTimeout / callBudgetShare
|
||||
byItems := requestOverhead + time.Duration(DefaultBatchLimit)*perItemBudget
|
||||
if byItems > share {
|
||||
t.Fatalf("条数上限 %d 的耗时上界 %s 超过超时的一半 %s(推演被调松了)",
|
||||
DefaultBatchLimit, byItems, share)
|
||||
}
|
||||
byChars := requestOverhead + time.Duration(DefaultBatchChars)*perCharBudget
|
||||
if byChars > share {
|
||||
t.Fatalf("字符上限 %d 的耗时上界 %s 超过超时的一半 %s(推演被调松了)",
|
||||
DefaultBatchChars, byChars, share)
|
||||
}
|
||||
// 旧配对(批次 64 + 5s 超时)必须仍被判为越界 —— 那正是 2026-10-13 的故障形态:
|
||||
// 实测 64 条 7.5~8.5s > 5s。这里用同一套预算常量复核它仍然出界。
|
||||
legacy := requestOverhead + 64*perItemBudget
|
||||
if legacy <= DefaultRequestTimeout {
|
||||
t.Fatalf("旧配对(64 条)的耗时上界 %s 竟然落在单次超时 %s 之内 —— "+
|
||||
"说明超时被放大到迁就批次了,本次修法不允许", legacy, DefaultRequestTimeout)
|
||||
}
|
||||
}
|
||||
|
||||
// TestChunkTexts_RespectsBothLimits 钉住切片:条数与字符总量两道界都要守,
|
||||
// 单条本身越界的文本独占一段且**不被截断**。
|
||||
func TestChunkTexts_RespectsBothLimits(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
got := chunkTexts(shortTexts(9), 2, 100)
|
||||
if len(got) != 5 {
|
||||
t.Fatalf("2 条/段 切 9 条 = %d 段,want 5", len(got))
|
||||
}
|
||||
for i, seg := range got {
|
||||
if len(seg) > 2 {
|
||||
t.Fatalf("第 %d 段有 %d 条,超过条数上限 2", i, len(seg))
|
||||
}
|
||||
}
|
||||
|
||||
// 字符界绑定:4 条 33 字符 = 132 字符 ⇒ 100 字符/段时必须收口(3 条 = 99 字符满,
|
||||
// 余下 1 条自成一段)。
|
||||
byChars := chunkTexts(shortTexts(4), 64, 100)
|
||||
if len(byChars) != 2 {
|
||||
t.Fatalf("100 字符/段 切 4×33 字符 = %d 段,want 2", len(byChars))
|
||||
}
|
||||
for i, seg := range byChars {
|
||||
total := 0
|
||||
for _, s := range seg {
|
||||
total += len([]rune(s))
|
||||
}
|
||||
if total > 100 {
|
||||
t.Fatalf("第 %d 段 %d 字符,超过字符上限 100", i, total)
|
||||
}
|
||||
}
|
||||
|
||||
// 超长单条:不切碎、不截断,独占一段。
|
||||
huge := strings.Repeat("长", 500)
|
||||
segs := chunkTexts([]string{huge, "短"}, 64, 100)
|
||||
if len(segs) != 2 || segs[0][0] != huge {
|
||||
t.Fatalf("超长单条应独占一段且不改写,got %d 段", len(segs))
|
||||
}
|
||||
|
||||
if chunkTexts(nil, 4, 100) != nil {
|
||||
t.Fatal("空输入应返回 nil")
|
||||
}
|
||||
}
|
||||
|
||||
// TestAPIEmbedder_DerivedBatchCompletesWithinTimeout 钉**不降级态**:
|
||||
// 按推出来的上限发一次(Embed 内部切成 1 段),在实测曲线下必须在单次超时内拿到向量。
|
||||
func TestAPIEmbedder_DerivedBatchCompletesWithinTimeout(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
rt := &latencyTripper{}
|
||||
emb := NewAPI(config.EmbeddingConfig{APIURL: "http://127.0.0.1:11434/v1/embeddings"},
|
||||
WithRoundTripper(rt))
|
||||
|
||||
texts := shortTexts(DefaultBatchLimit)
|
||||
vectors, err := emb.Embed(context.Background(), texts)
|
||||
if err != nil {
|
||||
t.Fatalf("上限内的批不该超时:%v", err)
|
||||
}
|
||||
if len(vectors) != len(texts) {
|
||||
t.Fatalf("got %d vectors, want %d", len(vectors), len(texts))
|
||||
}
|
||||
if rt.requests != 1 {
|
||||
t.Fatalf("上限内的批应发 1 次请求,实际 %d 次", rt.requests)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAPIEmbedder_OversizedBatchIsSplit 钉**不变式与调用方无关**:
|
||||
// 调用方给 3 倍上限的切片时,Embed 自己切成多段、每段都在界内,向量顺序与入参一致。
|
||||
func TestAPIEmbedder_OversizedBatchIsSplit(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
rt := &latencyTripper{}
|
||||
emb := NewAPI(config.EmbeddingConfig{APIURL: "http://127.0.0.1:11434/v1/embeddings"},
|
||||
WithRoundTripper(rt))
|
||||
|
||||
texts := shortTexts(DefaultBatchLimit * 3)
|
||||
vectors, err := emb.Embed(context.Background(), texts)
|
||||
if err != nil {
|
||||
t.Fatalf("超大切片应被切开而不是超时:%v", err)
|
||||
}
|
||||
if len(vectors) != len(texts) {
|
||||
t.Fatalf("got %d vectors, want %d(切开后必须按序拼回)", len(vectors), len(texts))
|
||||
}
|
||||
if rt.requests < 3 {
|
||||
t.Fatalf("3 倍上限应至少切成 3 段,实际 %d 段", rt.requests)
|
||||
}
|
||||
if rt.maxItems > DefaultBatchLimit {
|
||||
t.Fatalf("有请求带了 %d 条,超过推出来的条数上限 %d", rt.maxItems, DefaultBatchLimit)
|
||||
}
|
||||
if rt.maxChars > DefaultBatchChars {
|
||||
t.Fatalf("有请求带了 %d 字符,超过推出来的字符上限 %d", rt.maxChars, DefaultBatchChars)
|
||||
}
|
||||
// 🔴 本次修复的核心不变式,确定性断言(不依赖机器负载):
|
||||
// 没有任何一次请求发生「客户端先放弃、服务端还在算」——那正是级联堆积的起点。
|
||||
if rt.abandoned != 0 {
|
||||
t.Fatalf("有 %d 次请求被客户端先放弃(服务端还在算)⇒ 不变式没守住", rt.abandoned)
|
||||
}
|
||||
// 服务端侧的同一条不变式:任何一次请求的建模耗时(实测曲线)都落在单次超时内。
|
||||
if cost := measuredOverhead + time.Duration(rt.maxChars)*measuredPerChar; cost > DefaultRequestTimeout {
|
||||
t.Fatalf("最大一次请求的实测曲线耗时 %s 越过单次超时 %s", cost, DefaultRequestTimeout)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAPIEmbedder_LegacyBatchOf64TimesOut 是**变异自证**的另一半:同样是实测曲线,
|
||||
// 旧配对(一次请求 64 条 + 5s 超时)必须**超时**。它证明把上限写回 64 不是「保守」,而是
|
||||
// 必然把客户端与服务端撕裂(客户端走了、服务端还在算 ⇒ 后续请求全排队)。
|
||||
func TestAPIEmbedder_LegacyBatchOf64TimesOut(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
rt := &latencyTripper{}
|
||||
emb := NewAPI(config.EmbeddingConfig{APIURL: "http://127.0.0.1:11434/v1/embeddings"},
|
||||
WithRoundTripper(rt))
|
||||
|
||||
// 先热一次(越过 warmupTimeout),否则第一次调用有冷启动预算、不会超时。
|
||||
if _, err := emb.Embed(context.Background(), shortTexts(1)); err != nil {
|
||||
t.Fatalf("预热调用失败:%v", err)
|
||||
}
|
||||
start := time.Now()
|
||||
_, err := emb.embedChunk(context.Background(), shortTexts(64))
|
||||
elapsed := time.Since(start)
|
||||
if err == nil {
|
||||
t.Fatalf("旧配对(64 条 + %s 超时)在实测曲线下不该成功", DefaultRequestTimeout)
|
||||
}
|
||||
if elapsed > DefaultRequestTimeout+2*time.Second {
|
||||
t.Fatalf("超时保护没生效:等了 %s", elapsed)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAPIEmbedder_WarmupFirstCallBudget 钉冷启动那一路:第一次调用(模型加载 /
|
||||
// GPU 缓冲分配)拿的是 warmupTimeout,**只有第一次**;一旦拿到成功响应,后续请求
|
||||
// 立刻回到 DefaultRequestTimeout(真挂死的请求不会被 30s 拖住)。
|
||||
func TestAPIEmbedder_WarmupFirstCallBudget(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
rt := &latencyTripper{firstSleep: DefaultRequestTimeout + time.Second}
|
||||
emb := NewAPI(config.EmbeddingConfig{APIURL: "http://127.0.0.1:11434/v1/embeddings"},
|
||||
WithRoundTripper(rt))
|
||||
|
||||
if _, err := emb.Embed(context.Background(), shortTexts(1)); err != nil {
|
||||
t.Fatalf("冷启动第一次调用应被 warmupTimeout(%s) 兜住:%v", warmupTimeout, err)
|
||||
}
|
||||
if !emb.warmed.Load() {
|
||||
t.Fatal("拿到成功响应后 warmed 应为真")
|
||||
}
|
||||
|
||||
var slack time.Duration
|
||||
steady := roundTripFunc(func(r *http.Request) (*http.Response, error) {
|
||||
dl, ok := r.Context().Deadline()
|
||||
if !ok {
|
||||
t.Error("稳态请求没有 deadline")
|
||||
} else {
|
||||
slack = time.Until(dl)
|
||||
}
|
||||
return &http.Response{
|
||||
StatusCode: http.StatusOK,
|
||||
Body: io.NopCloser(strings.NewReader(`{"data":[{"index":0,"embedding":[1]}]}`)),
|
||||
Header: make(http.Header),
|
||||
}, nil
|
||||
})
|
||||
emb.client = &http.Client{Transport: steady}
|
||||
if _, err := emb.Embed(context.Background(), shortTexts(1)); err != nil {
|
||||
t.Fatalf("稳态调用失败:%v", err)
|
||||
}
|
||||
if slack <= 0 || slack > DefaultRequestTimeout {
|
||||
t.Fatalf("稳态单次超时余量 = %s, want (0, %s]", slack, DefaultRequestTimeout)
|
||||
}
|
||||
}
|
||||
@@ -119,12 +119,17 @@ func (t *Tidier) RunOnce(ctx context.Context) (TidyStats, error) {
|
||||
return stats, err
|
||||
}
|
||||
stats.RunID = runID
|
||||
// ledger 是**本轮实际发生**的召回来源记账口(不是配置里那一种):生产轮可以
|
||||
// 多分区混合(有的走语义、有的降级),因此汇总行只能由实测记录推出。
|
||||
ledger := &recallLedger{}
|
||||
|
||||
failures := t.scanPartitions(ctx, partitions, &stats, tidyRoundPolicy{
|
||||
sink: func(ctx context.Context, cands []TidyCandidate, stats *TidyStats) {
|
||||
t.persistCandidates(ctx, runID, cands, stats)
|
||||
},
|
||||
recall: t.fallbackRecall,
|
||||
recall: func(ctx context.Context, p tidyPartition, items []MemoryItem, stats *TidyStats, budget tidyBudget) ([]tidyPair, error) {
|
||||
return t.fallbackRecall(ctx, p, items, stats, budget, ledger)
|
||||
},
|
||||
})
|
||||
|
||||
// 轮末自动落地(2026-10-07 主人口径):规则层判为高置信的待定候选直接落地,
|
||||
@@ -150,7 +155,7 @@ func (t *Tidier) RunOnce(ctx context.Context) (TidyStats, error) {
|
||||
failures = append(failures, err)
|
||||
}
|
||||
|
||||
t.logRun(stats, run)
|
||||
t.logRun(stats, run, ledger)
|
||||
t.auditAutoApply(auto)
|
||||
|
||||
switch {
|
||||
@@ -165,7 +170,14 @@ func (t *Tidier) RunOnce(ctx context.Context) (TidyStats, error) {
|
||||
|
||||
// logRun 输出本轮的可观测口径:扫描 N / 候选 M / 建议 K / 否决原因分布。
|
||||
// 它带 source(memory/tidy),因此这条汇总永远可回溯到整理子系统。
|
||||
func (t *Tidier) logRun(stats TidyStats, run TidyRun) {
|
||||
//
|
||||
// ledger 给出的是**本轮实际发生**的召回来源(不是配置态)。这一条是 M6 2026-10-13
|
||||
// 的 medium finding 的落点:旧实现按「有没有配 embedder」写 recall_mode,全量降级
|
||||
// 在汇总行里被写成 embedding。现在:
|
||||
// - recall_mode = "":没有任何分区真正算过相似度(没有候选可比);
|
||||
// - recall_mode = lexical ∧ recall_degraded_partitions > 0:语义不可用、降级了;
|
||||
// - recall_mode = mixed:多分区里一部分走了语义、一部分降级 —— 不再被抹平。
|
||||
func (t *Tidier) logRun(stats TidyStats, run TidyRun, ledger *recallLedger) {
|
||||
fields := []any{
|
||||
"run_id", run.ID,
|
||||
"rule_version", run.RuleVersion,
|
||||
@@ -181,7 +193,8 @@ func (t *Tidier) logRun(stats TidyStats, run TidyRun) {
|
||||
"kept", stats.Kept,
|
||||
"errors", run.Errors,
|
||||
"duration_ms", run.FinishedAt.Sub(run.StartedAt).Milliseconds(),
|
||||
"recall_mode", tidyRecallMode(t.deps.Embedder != nil),
|
||||
"recall_mode", ledger.label(),
|
||||
"recall_degraded_partitions", ledger.degradedPartitions(),
|
||||
"auto_apply", t.cfg.autoApplyEnabled(),
|
||||
"auto_applied", stats.AutoApplied,
|
||||
"auto_merge", stats.AutoMerged,
|
||||
@@ -232,11 +245,15 @@ func (t *Tidier) scanPartitions(ctx context.Context, partitions []tidyPartition,
|
||||
// fallbackRecall 是**生产轮**的召回:相似度来源不可用时降级到零依赖的词项
|
||||
// 余弦(离线/无 embedding 配置时可跑,宁可比语义路径更保守),并记一条 WARN。
|
||||
// 两个来源**不可混用阈值**:阈值只有一个(design §13 开放点 2 的 0.80)。
|
||||
func (t *Tidier) fallbackRecall(ctx context.Context, p tidyPartition, items []MemoryItem, stats *TidyStats, budget tidyBudget) ([]tidyPair, error) {
|
||||
pairs, err := t.recallPairs(ctx, p, items, stats, budget)
|
||||
//
|
||||
// 降级这一笔**必须记进 ledger**:汇总行/面板上的召回来源只能由实测记录推出,
|
||||
// 否则「全量降级」会被报成「向量正常」(M6 2026-10-13 的 medium finding)。
|
||||
func (t *Tidier) fallbackRecall(ctx context.Context, p tidyPartition, items []MemoryItem, stats *TidyStats, budget tidyBudget, ledger *recallLedger) ([]tidyPair, error) {
|
||||
pairs, err := t.recallPairs(ctx, p, items, stats, budget, ledger)
|
||||
if err == nil {
|
||||
return pairs, nil
|
||||
}
|
||||
ledger.noteLexical(true)
|
||||
t.log.Warn("memory tidy embedding recall unavailable, falling back to lexical",
|
||||
"partition", p.label(), "error", err)
|
||||
return t.collectPairs(p, items, t.cfg.RecallThreshold,
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding"
|
||||
)
|
||||
|
||||
// 本文件收拢记忆整理(design/23-记忆整理与去重)全部的标识、默认值与旋钮,
|
||||
@@ -74,7 +75,17 @@ const (
|
||||
|
||||
// tidyEmbedBatch 是单次 Embed 调用的文本条数上限,避免把整个分区的文本
|
||||
// 一次性交给远端 embedding API。
|
||||
tidyEmbedBatch = 64
|
||||
//
|
||||
// 🔴 它**不是**拍出来的数字,而是 internal/runtime/embedding 里**同一条推演**
|
||||
// 的结果(budget.go:一次请求的耗时上界 ≤ 单次超时的一半 ⇒ 批上限 = 超时的一半
|
||||
// ÷ 单条预算上界)。2026-10-13:旧值 64 的实测耗时 7.5~8.5s 已越过客户端 5s
|
||||
// 单次超时 ⇒ 客户端放弃、服务端继续算 ⇒ 本机 embedding 服务被顶住,单条请求
|
||||
// (写向量索引)也跟着超时。把批上限与超时绑到同一条推演上,就不会再出现
|
||||
// 「只调一侧、另一侧静默失效」。
|
||||
//
|
||||
// 批上限在 embedder 内部**还有一道强制**(Embed 自己切片),因此这里即使被改大
|
||||
// 也不会把服务端顶住;这一行只是让调用方别去构造注定被切开的超大切片。
|
||||
tidyEmbedBatch = embedding.DefaultBatchLimit
|
||||
|
||||
// minTidyContentRunes 是参与比对的最小内容长度(按 rune)。过短的内容
|
||||
// (如单字)没有可判定的事实要素,进比对只会产出噪声,直接跳过。
|
||||
@@ -126,6 +137,11 @@ const (
|
||||
maxTidyRejectBatchLimit = 10000
|
||||
)
|
||||
|
||||
// 编译期不变式(2026-10-13):批上限一旦被调回超过 embedding 推出来的值
|
||||
// (例:写回旧值 64),下面这行常量表达式会变成负数、uint 溢出 ⇒ **编译失败**。
|
||||
// 这条断言把「批上限与超时同源」从注释里的承诺变成编译器守的门。
|
||||
const _ = uint(embedding.DefaultBatchLimit - tidyEmbedBatch)
|
||||
|
||||
// TidyConfig 携带记忆整理的调优旋钮。零值经 normalize 选取安全默认值,
|
||||
// 因此半填的 Config 仍可用。
|
||||
type TidyConfig struct {
|
||||
|
||||
@@ -52,11 +52,15 @@ type TidyDryRun struct {
|
||||
// 「完整的建议」展示)。
|
||||
func (t *Tidier) DryRun(ctx context.Context) (TidyDryRun, error) {
|
||||
started := t.now()
|
||||
out := TidyDryRun{RecallMode: tidyRecallMode(t.deps.Embedder != nil)}
|
||||
out := TidyDryRun{}
|
||||
stats, partitions, err := t.beginRound(ctx)
|
||||
if err != nil {
|
||||
return out, err
|
||||
}
|
||||
// ledger 是**本轮实际发生**的召回来源:试算会写进 RecallMode 返回给面板。
|
||||
// 旧实现按「有没有配 embedder」写(tidyRecallMode(deps.Embedder != nil)),
|
||||
// 那会让面板把「向量不可用」读成「向量正常」(M6 2026-10-13 的 medium finding)。
|
||||
ledger := &recallLedger{}
|
||||
|
||||
// 收集出口:闭包只在 scanPartitions 的**串行**循环里被调用,因此无需加锁。
|
||||
collected := make([]TidyCandidate, 0)
|
||||
@@ -64,12 +68,17 @@ func (t *Tidier) DryRun(ctx context.Context) (TidyDryRun, error) {
|
||||
sink: func(_ context.Context, cands []TidyCandidate, _ *TidyStats) {
|
||||
collected = append(collected, cands...)
|
||||
},
|
||||
recall: t.strictRecall,
|
||||
recall: func(ctx context.Context, p tidyPartition, items []MemoryItem, stats *TidyStats, budget tidyBudget) ([]tidyPair, error) {
|
||||
return t.strictRecall(ctx, p, items, stats, budget, ledger)
|
||||
},
|
||||
}
|
||||
failures := t.scanPartitions(ctx, partitions, &stats, policy)
|
||||
|
||||
out.Stats = stats
|
||||
out.Candidates = collected
|
||||
// 如实上报**实际发生**的来源:空串 = 没有任何分区真正算过相似度(「没候选」,
|
||||
// 与「向量不可用」不是一回事)。
|
||||
out.RecallMode = ledger.label()
|
||||
out.DurationMS = t.now().Sub(started).Milliseconds()
|
||||
if len(failures) > 0 {
|
||||
return out, fmt.Errorf("memory/tidy: dry-run finished with errors: %w", errors.Join(failures...))
|
||||
@@ -80,8 +89,8 @@ func (t *Tidier) DryRun(ctx context.Context) (TidyDryRun, error) {
|
||||
|
||||
// strictRecall 是**试算轮**的召回:语义召回不可用时**不**降级,如实上报。
|
||||
// 日志只记分区与错误,绝不带记忆正文(候选预览里已含正文)。
|
||||
func (t *Tidier) strictRecall(ctx context.Context, p tidyPartition, items []MemoryItem, stats *TidyStats, budget tidyBudget) ([]tidyPair, error) {
|
||||
pairs, err := t.recallPairs(ctx, p, items, stats, budget)
|
||||
func (t *Tidier) strictRecall(ctx context.Context, p tidyPartition, items []MemoryItem, stats *TidyStats, budget tidyBudget, ledger *recallLedger) ([]tidyPair, error) {
|
||||
pairs, err := t.recallPairs(ctx, p, items, stats, budget, ledger)
|
||||
if err != nil {
|
||||
t.log.Warn("memory tidy dry-run recall unavailable",
|
||||
"partition", p.label(), "items", len(items), "error", err)
|
||||
|
||||
@@ -25,8 +25,75 @@ const (
|
||||
tidyRecallEmbedding = "embedding"
|
||||
// tidyRecallLexical 是本轮相似度来自零依赖的词项余弦。
|
||||
tidyRecallLexical = "lexical"
|
||||
// tidyRecallMixed 是多分区的一轮里**部分**分区走了语义、部分降级到词项余弦。
|
||||
// 它取代「挑一个代表值报上去」的做法:降级一旦发生就必须看得见。
|
||||
tidyRecallMixed = "mixed"
|
||||
)
|
||||
|
||||
// recallLedger 记录**一轮里实际发生**的召回来源(不是配置里那一种)。
|
||||
//
|
||||
// 为什么需要它(M6 2026-10-13 的 medium finding):整理轮与试算轮的 `recall_mode`
|
||||
// 曾按「有没有配 embedder」写出去(旧实现 tidyRecallMode(hasEmbedder bool)),于是
|
||||
// **全量降级**在汇总行/面板里被写成 `embedding` —— 汇总与事实相反,运维据此判读会
|
||||
// 把「向量不可用」读成「向量正常工作」。本类型把口径翻过来:只在**真的用了**那条
|
||||
// 路的地方记一笔,label() 完全由这些实测记录推出。
|
||||
type recallLedger struct {
|
||||
// embedding / lexical 是**真正用该来源算过相似度**的分区数。
|
||||
embedding int
|
||||
lexical int
|
||||
// degraded 是**因相似度来源不可用**而降级到词项余弦的分区数(不是「没配
|
||||
// embedder」那种配置形态)。它把「没候选」(label 为空串)与「向量不可用」
|
||||
// (label 为 lexical 且 degraded > 0)分开 —— 这两件事在运维上不是一回事。
|
||||
degraded int
|
||||
}
|
||||
|
||||
// noteEmbedding 记一笔「这个分区真的用了语义 embedding」。
|
||||
func (l *recallLedger) noteEmbedding() {
|
||||
if l != nil {
|
||||
l.embedding++
|
||||
}
|
||||
}
|
||||
|
||||
// noteLexical 记一笔「这个分区用了词项余弦」;degraded 表示它是**降级**来的
|
||||
// (embedder 报错 / 向量条数错位),而不是「本来就没配 embedder」。
|
||||
func (l *recallLedger) noteLexical(degraded bool) {
|
||||
if l == nil {
|
||||
return
|
||||
}
|
||||
l.lexical++
|
||||
if degraded {
|
||||
l.degraded++
|
||||
}
|
||||
}
|
||||
|
||||
// label 返回本轮**实际**的召回来源:
|
||||
// - "":没有任何分区真正算过相似度(没有候选可比)—— 「没候选」不是「降级」;
|
||||
// - embedding / lexical:所有算过的分区都走同一条路;
|
||||
// - mixed:两种都有(多分区里有一部分降级)。
|
||||
func (l *recallLedger) label() string {
|
||||
if l == nil {
|
||||
return ""
|
||||
}
|
||||
switch {
|
||||
case l.embedding == 0 && l.lexical == 0:
|
||||
return ""
|
||||
case l.embedding == 0:
|
||||
return tidyRecallLexical
|
||||
case l.lexical == 0:
|
||||
return tidyRecallEmbedding
|
||||
default:
|
||||
return tidyRecallMixed
|
||||
}
|
||||
}
|
||||
|
||||
// degradedPartitions 返回本轮因向量不可用而降级的分区数。
|
||||
func (l *recallLedger) degradedPartitions() int {
|
||||
if l == nil {
|
||||
return 0
|
||||
}
|
||||
return l.degraded
|
||||
}
|
||||
|
||||
// tidyPair 是一条被召回的候选对。
|
||||
type tidyPair struct {
|
||||
a, b MemoryItem
|
||||
@@ -35,18 +102,23 @@ type tidyPair struct {
|
||||
mode string
|
||||
}
|
||||
|
||||
// recallPairs 在分区内召回候选对。条目少于 2 条时为空操作(无候选、不报错)。
|
||||
// recallPairs 在分区内召回候选对。条目少于 2 条时为空操作(无候选、不报错),
|
||||
// 且**不往 ledger 记任何一笔** —— 「没有候选可比」不是「走哪条路召回」。
|
||||
//
|
||||
// 相似度来源不可用时(embedder 报错、向量条数错位)返回错误 —— 回退决策不在
|
||||
// 这里:生产轮把它降级成词项余弦,试算轮如实上报(见 tidyRecall 策略)。
|
||||
//
|
||||
// ledger 是**实际发生**的来源记账口(见 recallLedger):它只在真正选定相似度
|
||||
// 函数的那一行记账,因此谁也无法把它报成「配置里那一种」。
|
||||
//
|
||||
// 返回的对按「分数降序 + id」确定性排序,使每分区候选上限的截断可复现
|
||||
// (同一批数据两次运行产出同一批候选 ⇒ 幂等,design §4.1 A)。
|
||||
func (t *Tidier) recallPairs(ctx context.Context, p tidyPartition, items []MemoryItem, stats *TidyStats, budget tidyBudget) ([]tidyPair, error) {
|
||||
func (t *Tidier) recallPairs(ctx context.Context, p tidyPartition, items []MemoryItem, stats *TidyStats, budget tidyBudget, ledger *recallLedger) ([]tidyPair, error) {
|
||||
if len(items) < 2 {
|
||||
return nil, nil
|
||||
}
|
||||
if t.deps.Embedder == nil {
|
||||
ledger.noteLexical(false)
|
||||
return t.collectPairs(p, items, t.cfg.RecallThreshold,
|
||||
lexicalSimFunc(items), tidyRecallLexical, stats, budget), nil
|
||||
}
|
||||
@@ -57,6 +129,7 @@ func (t *Tidier) recallPairs(ctx context.Context, p tidyPartition, items []Memor
|
||||
if len(vecs) != len(items) {
|
||||
return nil, ErrTidyEmbedCount
|
||||
}
|
||||
ledger.noteEmbedding()
|
||||
return t.collectPairs(p, items, t.cfg.RecallThreshold,
|
||||
vectorSimFunc(vecs), tidyRecallEmbedding, stats, budget), nil
|
||||
}
|
||||
@@ -205,10 +278,6 @@ func clampUnit(v float64) float64 {
|
||||
return v
|
||||
}
|
||||
|
||||
// tidyRecallMode 报告本轮相似度的来源,便于运维判断阈值是否在预期口径上。
|
||||
func tidyRecallMode(hasEmbedder bool) string {
|
||||
if hasEmbedder {
|
||||
return tidyRecallEmbedding
|
||||
}
|
||||
return tidyRecallLexical
|
||||
}
|
||||
// 旧的 tidyRecallMode(hasEmbedder bool) 已删除(M6 2026-10-13 的 medium finding):
|
||||
// 它按**配置态**报「本轮相似度来源」,把全量降级掩盖成 embedding。现在的口径是
|
||||
// recallLedger —— 只报**实际发生**的那一种,见本文件顶部的 recallLedger。
|
||||
@@ -0,0 +1,185 @@
|
||||
package memory
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding"
|
||||
)
|
||||
|
||||
// 本文件钉 M6(2026-10-13)的 medium finding:整理轮/试算轮的 `recall_mode`
|
||||
// 必须是**实际发生**的那一种,不是配置态。
|
||||
//
|
||||
// 旧实现(tidyRecallMode(hasEmbedder bool))按「有没有配 embedder」写,于是
|
||||
// 「配了但是向量不可用 ⇒ 全量降级」在汇总行里被写成 embedding —— 汇总与事实相反。
|
||||
//
|
||||
// 「配了但不可用」的 embedder 复用 tidy_dryrun_test.go 的 failingEmbedder
|
||||
// (正是线上那 2568 次 `embedding recall unavailable` 的形态)。
|
||||
|
||||
// newLogCaptureTidier 构建一个把汇总行写进 buffer 的整理器(便于断言日志字段)。
|
||||
func newLogCaptureTidier(t *testing.T, store *Store, cfg TidyConfig, emb embedding.Embedder) (*Tidier, *bytes.Buffer) {
|
||||
t.Helper()
|
||||
buf := &bytes.Buffer{}
|
||||
tr, err := NewTidier(cfg, TidyDeps{
|
||||
Store: store,
|
||||
Embedder: emb,
|
||||
Log: slog.New(slog.NewTextHandler(buf, nil)),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("NewTidier: %v", err)
|
||||
}
|
||||
return tr, buf
|
||||
}
|
||||
|
||||
// TestRecallLedger_LabelComesFromActualUse 钉记账口本身:label 完全由「真的用了
|
||||
// 哪条路」推出,混合与降级都不许被抹平。
|
||||
func TestRecallLedger_LabelComesFromActualUse(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
notes func(l *recallLedger)
|
||||
want string
|
||||
degraded int
|
||||
}{
|
||||
{name: "没算过(没候选)", notes: func(*recallLedger) {}, want: ""},
|
||||
{name: "全部语义", notes: func(l *recallLedger) { l.noteEmbedding() }, want: tidyRecallEmbedding},
|
||||
{name: "全部词项", notes: func(l *recallLedger) { l.noteLexical(false) }, want: tidyRecallLexical},
|
||||
{
|
||||
name: "降级到词项",
|
||||
notes: func(l *recallLedger) { l.noteLexical(true) },
|
||||
want: tidyRecallLexical,
|
||||
degraded: 1,
|
||||
},
|
||||
{
|
||||
name: "多分区混合(部分降级)",
|
||||
notes: func(l *recallLedger) { l.noteEmbedding(); l.noteLexical(true) },
|
||||
want: tidyRecallMixed,
|
||||
degraded: 1,
|
||||
},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
tc := tc
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
var l recallLedger
|
||||
tc.notes(&l)
|
||||
if got := l.label(); got != tc.want {
|
||||
t.Fatalf("label = %q, want %q", got, tc.want)
|
||||
}
|
||||
if got := l.degradedPartitions(); got != tc.degraded {
|
||||
t.Fatalf("degraded = %d, want %d", got, tc.degraded)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestTidyRunLog_RecallModeReportsDegradation 是 finding 的正面回归:配了 embedder
|
||||
// 但它报错 ⇒ 汇总行必须写 `recall_mode=lexical` 且 `recall_degraded_partitions=1`,
|
||||
// **绝不能**写成 embedding。
|
||||
func TestTidyRunLog_RecallModeReportsDegradation(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
store := newTestStore(t)
|
||||
seedDryRunFixture(t, store)
|
||||
|
||||
emb := failingEmbedder{err: errors.New("embedding: request: context deadline exceeded")}
|
||||
tr, buf := newLogCaptureTidier(t, store, tidyTestCfg(0.80), emb)
|
||||
if _, err := tr.RunOnce(ctx); err != nil {
|
||||
t.Fatalf("RunOnce 应当降级跑完而不是失败:%v", err)
|
||||
}
|
||||
|
||||
line := lastLogLine(t, buf, "memory tidy run finished")
|
||||
if !strings.Contains(line, "recall_mode=lexical") {
|
||||
t.Fatalf("降级轮的汇总行必须写 lexical,实际:%s", line)
|
||||
}
|
||||
if strings.Contains(line, "recall_mode=embedding") {
|
||||
t.Fatalf("降级轮被报成了 embedding(正是本 finding 的形态):%s", line)
|
||||
}
|
||||
if !strings.Contains(line, "recall_degraded_partitions=1") {
|
||||
t.Fatalf("降级分区数没写出来:%s", line)
|
||||
}
|
||||
}
|
||||
|
||||
// TestTidyRunLog_RecallModeReportsEmbeddingWhenHealthy 是反面:embedder 正常时
|
||||
// 汇总行必须如实写 embedding、降级数为 0(不能因为「怕误报」而一律写降级)。
|
||||
func TestTidyRunLog_RecallModeReportsEmbeddingWhenHealthy(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
store := newTestStore(t)
|
||||
seedDryRunFixture(t, store)
|
||||
|
||||
tr, buf := newLogCaptureTidier(t, store, tidyTestCfg(0.80), dryRunFixtureEmbedder())
|
||||
if _, err := tr.RunOnce(ctx); err != nil {
|
||||
t.Fatalf("RunOnce: %v", err)
|
||||
}
|
||||
|
||||
line := lastLogLine(t, buf, "memory tidy run finished")
|
||||
if !strings.Contains(line, "recall_mode=embedding") {
|
||||
t.Fatalf("健康轮必须写 embedding,实际:%s", line)
|
||||
}
|
||||
if !strings.Contains(line, "recall_degraded_partitions=0") {
|
||||
t.Fatalf("健康轮的降级数必须为 0:%s", line)
|
||||
}
|
||||
}
|
||||
|
||||
// TestTidyRunLog_RecallModeEmptyWhenNoPairs 钉「没候选」与「向量不可用」可区分:
|
||||
// 分区里只有一条(压根没有对可比)时,来源是**空串**,而不是任何一条召回路径。
|
||||
func TestTidyRunLog_RecallModeEmptyWhenNoPairs(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
store := newTestStore(t)
|
||||
base := tidyTestTime(t)
|
||||
seedTidyItem(t, store, habitItem(fixtureHabitShort, base, 0.9))
|
||||
|
||||
emb := failingEmbedder{err: errors.New("embedding: request: context deadline exceeded")}
|
||||
tr, buf := newLogCaptureTidier(t, store, tidyTestCfg(0.80), emb)
|
||||
if _, err := tr.RunOnce(ctx); err != nil {
|
||||
t.Fatalf("RunOnce: %v", err)
|
||||
}
|
||||
|
||||
line := lastLogLine(t, buf, "memory tidy run finished")
|
||||
if !strings.Contains(line, `recall_mode=""`) {
|
||||
t.Fatalf("没有候选可比时来源应为空串(区别于降级),实际:%s", line)
|
||||
}
|
||||
if !strings.Contains(line, "recall_degraded_partitions=0") {
|
||||
t.Fatalf("没有候选可比不算降级:%s", line)
|
||||
}
|
||||
}
|
||||
|
||||
// TestTidyDryRun_RecallModeIsActualNotConfigured 是**试算轮**(面板那一条)的同口径
|
||||
// 回归:配了 embedder 但向量不可用 ⇒ 试算如实报错(严格模式),且返回的 RecallMode
|
||||
// **不**写成 embedding。
|
||||
func TestTidyDryRun_RecallModeIsActualNotConfigured(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
store := newTestStore(t)
|
||||
seedDryRunFixture(t, store)
|
||||
|
||||
emb := failingEmbedder{err: errors.New("embedding: request: context deadline exceeded")}
|
||||
tr, _ := newLogCaptureTidier(t, store, tidyTestCfg(0.80), emb)
|
||||
out, err := tr.DryRun(ctx)
|
||||
if !errors.Is(err, ErrTidyRecallUnavailable) {
|
||||
t.Fatalf("试算轮必须如实上报不可用,err = %v", err)
|
||||
}
|
||||
if out.RecallMode == tidyRecallEmbedding {
|
||||
t.Fatalf("向量不可用却把 RecallMode 报成 embedding(配置态口径):%q", out.RecallMode)
|
||||
}
|
||||
}
|
||||
|
||||
// lastLogLine 返回 buffer 里最后一条含 marker 的日志行。
|
||||
func lastLogLine(t *testing.T, buf *bytes.Buffer, marker string) string {
|
||||
t.Helper()
|
||||
lines := strings.Split(strings.TrimSpace(buf.String()), "\n")
|
||||
for i := len(lines) - 1; i >= 0; i-- {
|
||||
if strings.Contains(lines[i], marker) {
|
||||
return lines[i]
|
||||
}
|
||||
}
|
||||
t.Fatalf("日志里没有找到 %q:\n%s", marker, buf.String())
|
||||
return ""
|
||||
}
|
||||
Reference in new issue
Block a user