Author SHA1 Message Date
AskaEth 6c30ca540a fix(memory): 冷启动放宽改为一次性 + 试算轮区分「没候选」与「向量不可用」
review round1 的 p1/p2 收口(commit 5967272 之后)。

p1:旧实现 `warmed` 只在 2xx 之后才翻 ⇒ 注释里写的「只在第一次成功响应之前」
实际是「服务坏着的全部时间」。reviewer 实测「接受连接但永不回包」时**每次**调用
都等满 30s(fail-fast 变 fail-slow):审查 20s 预算被一次吃光、索引写入拖成
8×30s、开机探针阻塞 30s。

- `warmupLeft` 改为**一次性许可**(CompareAndSwap 取走即失效,成败都算花掉)
  ⇒ 挂死时只有第一个请求慢一次,此后一律 DefaultRequestTimeout。
- `warmupTimeout` 30s → 8s:与常态 5s 同一量级(≤2×),且严格小于审查总预算
  20s(一次冷启动不许吃光整轮);两条界都由 TestDefaultRequestTimeout 守住。
- 放宽被花掉且没成功时留一条 WARN(事后可复核「第一个请求」这个判据)。
- 更正两处自述:api.go 的「不会被 30s 拖住」与 budget.go 的不变式说明(批次形状
  永远按稳态超时推导;冷启动只放宽第一个请求的截止时刻)。
- p2:删掉零调用者的 `WithRequestTimeout` —— 单次超时只允许改常量,批次上限与它
  同源并由编译期/常量断言复核。

p2:试算轮在向量源不可用时 `RecallMode` 与「没有候选可比」同为空串。
- `recallLedger` 增 `unavailable` 计数 + `noteUnavailable()`(strictRecall 失败
  那一笔记一笔);试算汇总行新增 `recall_unavailable_partitions`,**成败都记**。
- 不扩 `recallMode` 取值集合:失败那一路不序列化 result(响应是
  500 tidy_recall_unavailable),新取值只会出现在日志里却要动一次契约,不划算。

测试:三种失败形态 × 两次调用的耗时断言(连接被拒 / 500 / 永不回包),
试算轮的「不可用」与「没候选」两态日志断言;两处变异自证(warmupTimeout 调回
30s ⇒ TestDefaultRequestTimeout 红;去掉 noteUnavailable ⇒ 试算汇总行退化成
recall_mode="" + recall_unavailable_partitions=0)。
2026-10-13 20:51:10 +08:00
AskaEth 59672720bd 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、不动判定规则与任何阈值。
2026-10-13 20:33:26 +08:00
9 changed files with 1036 additions and 44 deletions

No files matched your search

+76 -20
View File
@@ -11,6 +11,7 @@ import (
"net/http"
"strings"
"sync"
"sync/atomic"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
@@ -38,23 +39,24 @@ 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 +95,11 @@ type APIEmbedder struct {
cfg config.EmbeddingConfig
client *http.Client
log *slog.Logger
// warmupLeft 是**一次性**冷启动放宽的开关:为真时下一个请求用 warmupTimeout,
// 取走即置假(CompareAndSwap ⇒ 并发下也只有一个请求能拿到)。
// 它**不是**「未成功就一直放宽」—— 失败也照样花掉这一次,因此服务挂死时
// 只会慢一次,之后每个请求都按 DefaultRequestTimeout 快速失败(见 budget.go)。
warmupLeft atomic.Bool
mu sync.RWMutex
dim int
@@ -131,12 +138,16 @@ func WithLogger(l *slog.Logger) APIOption {
}
// NewAPI builds an APIEmbedder from cfg using the default HTTP client.
// 单次超时不可由选项更换:它是 budget.go 那条推演的输入,批次上限与它同源,
// 因此**只允许改常量**(改了之后批次上限自动跟着改,测试的编译期/常量断言立刻复核)。
func NewAPI(cfg config.EmbeddingConfig, opts ...APIOption) *APIEmbedder {
e := &APIEmbedder{
cfg: cfg,
client: defaultHTTPClient(),
log: slog.New(slog.NewTextHandler(io.Discard, nil)),
}
// 冷启动放宽是一个**一次性**许可:见 APIEmbedder.warmupLeft。
e.warmupLeft.Store(true)
for _, opt := range opts {
opt(e)
}
@@ -155,8 +166,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 +181,40 @@ func (e *APIEmbedder) Embed(ctx context.Context, texts []string) ([]Vector, erro
return nil, ErrNoAPIURL
}
callCtx, cancel := context.WithTimeout(ctx, DefaultRequestTimeout)
// 切片用的上限由**稳态**超时推出(budget.go 的 DefaultRequestTimeout):
// 冷启动那一次只放宽**截止时刻**,批形状不变 —— 否则首个请求会按更宽的预算
// 切出一个巨大的批,正好把冷启动的模型顶得更久。
items, chars := batchLimitsFor(DefaultRequestTimeout)
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
}
// callTimeout 返回本次请求该用的单次超时,并**当场花掉**一次性冷启动放宽许可
// (成功、失败都算花掉)。因此:
// - 本进程的第一个请求最多等 warmupTimeout(冷启动),
// - 从第二个请求起一律 DefaultRequestTimeout —— 服务挂死时不会每次都慢。
//
// 返回值 relaxed 表示这一次是不是拿到了放宽,供失败时留痕。
func (e *APIEmbedder) callTimeout() (timeout time.Duration, relaxed bool) {
if e.warmupLeft.CompareAndSwap(true, false) {
return warmupTimeout, true
}
return DefaultRequestTimeout, false
}
// embedChunk 发**一次** /embeddings 请求。chunk 必须已经落在 batchLimitsFor 推出来
// 的边界内(Embed 负责切),因此它的耗时上界不超过单次超时的一半。
func (e *APIEmbedder) embedChunk(ctx context.Context, texts []string) ([]Vector, error) {
// 先取超时(取走即花掉那次放宽),再建 ctx:放宽许可与这次请求一一对应。
timeout, relaxed := e.callTimeout()
callCtx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
req, err := e.newEmbeddingRequest(callCtx, texts)
@@ -178,6 +224,16 @@ func (e *APIEmbedder) Embed(ctx context.Context, texts []string) ([]Vector, erro
resp, err := e.client.Do(req)
if err != nil {
// 冷启动放宽被花掉且没成功时留一条 WARN:它是「本进程第一次请求」这个
// 判断唯一能被事后复核的地方(此后一切失败都按常态超时快速失败)。
// 只记两个预算与错误(错误文本里已带端点,不另记 URL,避免重复落一条
// 可能含 userinfo 的地址)。
if relaxed {
e.log.Warn("embedding: one-shot cold-start allowance spent without a successful response",
"warmup_timeout", warmupTimeout.String(),
"steady_timeout", DefaultRequestTimeout.String(),
"error", err)
}
return nil, fmt.Errorf("embedding: request: %w", err)
}
defer func() { _ = resp.Body.Close() }()
+24 -2
View File
@@ -222,6 +222,11 @@ const reviewBudgetRef = 20 * time.Second
// 审查总预算:否则一次卡死的 embedding 连接就能吃光整条审查的预算,排在后面的
// T1 灰区判定连执行机会都没有(2026-10-07 实测故障形态)。同时验证该超时确实
// 打在了发出去的请求上。
//
// 🔴 冷启动放宽(warmupTimeout)必须一并守住同一条界:它只发生一次,但它那一次的
// 等待**也**不能吃光整轮审查 —— 否则「一次挂死 = 整轮审查没有预算给 T1」这个形态
// 会从常态超时挪到放宽预算上原样复现(reviewer round1 的 p1)。因此这里对
// warmupTimeout 做与 DefaultRequestTimeout **同一条**断言。
func TestDefaultRequestTimeout(t *testing.T) {
t.Parallel()
@@ -236,6 +241,19 @@ func TestDefaultRequestTimeout(t *testing.T) {
t.Fatalf("DefaultRequestTimeout = %s, want 5s(值变了要连带复核审查预算口径)",
DefaultRequestTimeout)
}
// 冷启动放宽的两条界:仍在常态超时的**同一量级**(不得成倍放大),
// 且严格小于审查总预算(一次冷启动不许吃光整轮)。
if warmupTimeout <= 0 {
t.Fatalf("warmupTimeout = %s, 必须为正", warmupTimeout)
}
if warmupTimeout > 2*DefaultRequestTimeout {
t.Fatalf("冷启动放宽 %s 超过常态超时 %s 的两倍 —— 它必须是同一量级的一次性放宽",
warmupTimeout, DefaultRequestTimeout)
}
if warmupTimeout >= reviewBudgetRef {
t.Fatalf("冷启动放宽 %s 必须严格小于审查总预算 %s(否则一次冷启动就吃光整轮审查)",
warmupTimeout, reviewBudgetRef)
}
var slack time.Duration
rt := roundTripFunc(func(r *http.Request) (*http.Response, error) {
@@ -256,8 +274,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)
+145
View File
@@ -0,0 +1,145 @@
package embedding
import "time"
// 本文件是**单次 /embeddings 调用的预算推演**:批次上限与单次超时必须由**同一个
// 式子**得出。任一侧单独调整都会让「一次请求的实测耗时 < 单次超时」这条不变式
// 静默失效 —— 而失效的后果不是报错,是**服务端还在算、客户端已经放弃**:被放弃的
// 那一批仍占着本机 llama.cpp 的算力,后面每一个请求(包括写索引时的单条请求)都
// 排在它后面等到自己的超时,于是整条记忆写入路径被拖住、召回全线降级。
//
// ## 不变式
//
// requestOverhead + 一次请求的字符总量 × perCharBudget ≤ DefaultRequestTimeout / callBudgetShare
//
// 左半是**推演出来的耗时上界**,右半是该超时下允许用掉的一半(另一半是这台机器的
// 抖动余量)。批次上限由解这个不等式得到(见 DefaultBatchLimit / DefaultBatchChars)。
//
// ⚠️ 这条不等式**只按稳态超时(DefaultRequestTimeout)推导,且只约束批次形状**;
// 冷启动那次放宽(warmupTimeout)**不改批形状**,只放宽进程第一个请求的**截止
// 时刻**,而且只发生一次(见 warmupTimeout 的说明)。批形状与超时仍然同源:
// 无论第一个请求的截止多宽,它带的条数/字符数永远按稳态超时算出来。
//
// ## 实测依据(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 是**一次性的**冷启动放宽:本 embedder 生命周期里的**第一个**
// 请求可以拿到这个截止(模型加载 / GPU 缓冲分配还没发生过),此后**每一个**
// 请求(含刚失败的那个之后的下一个)一律回到 DefaultRequestTimeout。
//
// 🔴 它**不是**「只要还没成功过就一直放宽」—— 那个口径会在服务挂死时把
// fail-fast 变成 fail-slow(每次都等 warmupTimeout),并把审查的 20s 预算、
// 索引写入的 5s 预算整段吃掉。一次性 + 8s 两条一起才是有界的:
// - 一次性:放宽最多花掉**一个**请求的时长,最坏情形之后立刻回到 5s;
// - 8s:与常态 5s **同一量级**(1.6×),且严格小于审查单条总预算 20s
// (memory.defaultReviewTimeout)⇒ 一次冷启动不可能吃光整轮审查,
// 后面的 T1 灰区判定(8s)仍有预算。这条不等式由测试守住
// (api_url_test.go 的 TestDefaultRequestTimeout)。
//
// 依据:2026-10-13 部署后第一分钟的生产日志里有一条
// `vector embed for index failed ... embed 1 texts ... context deadline exceeded`
// —— 模型还没热时 5s 不够;而稳态实测单条只要 0.03~0.19s(见上面的曲线),
// 所以放宽只需要覆盖「进程的第一个请求」这一刻。
warmupTimeout = 8 * 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:])
}
+393
View File
@@ -0,0 +1,393 @@
package embedding
import (
"context"
"encoding/json"
"errors"
"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_WarmupIsOneShot 钉 p1 的核心:冷启动放宽**一次性**,三种失败形态
// 下都必须是「第一次最多等 warmupTimeout,第二次起一律 DefaultRequestTimeout」。
//
// 这是 reviewer 在 round1 打回的那条:旧实现 `warmed` 只在 2xx 之后才翻,于是
// 「服务接受连接但永不回包」时**每一次**调用都等完整放宽预算(30s),把 fail-fast
// 变成 fail-slow —— 一次挂死就能吃光审查的 20s 预算、把索引写入拖成 8×30s。
func TestAPIEmbedder_WarmupIsOneShot(t *testing.T) {
t.Parallel()
cases := []struct {
name string
rt func(t *testing.T) http.RoundTripper
}{
{
name: "连接被拒",
rt: func(t *testing.T) http.RoundTripper {
return roundTripFunc(func(*http.Request) (*http.Response, error) {
return nil, errors.New("dial tcp 127.0.0.1:11434: connectex: 目标计算机积极拒绝")
})
},
},
{
name: "HTTP 500",
rt: func(t *testing.T) http.RoundTripper {
return roundTripFunc(func(*http.Request) (*http.Response, error) {
return &http.Response{
StatusCode: http.StatusInternalServerError,
Body: io.NopCloser(strings.NewReader("server exploded")),
Header: make(http.Header),
}, nil
})
},
},
{
name: "接受连接但永不回包",
rt: func(t *testing.T) http.RoundTripper {
return roundTripFunc(func(r *http.Request) (*http.Response, error) {
<-r.Context().Done() // 只等服务端超时,永不回包
return nil, r.Context().Err()
})
},
},
}
for _, tc := range cases {
tc := tc
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
emb := NewAPI(config.EmbeddingConfig{APIURL: "http://127.0.0.1:11434/v1/embeddings"},
WithRoundTripper(tc.rt(t)))
// 第一次:放宽预算。断言上界 = warmupTimeout(不得再宽),
// 下界只对会真正等到截止的那一种(挂死)有意义。
start := time.Now()
_, err := emb.Embed(context.Background(), shortTexts(1))
first := time.Since(start)
if err == nil {
t.Fatal("这一组失败形态不该成功")
}
t.Logf("首次耗时 = %s", first.Round(time.Millisecond))
if first > warmupTimeout+time.Second {
t.Fatalf("首次耗时 %s 超过冷启动放宽 %s", first, warmupTimeout)
}
// 第二次:**必须**回到常态超时 —— 一次挂死绝不许变成每次都慢。
start = time.Now()
if _, err := emb.Embed(context.Background(), shortTexts(1)); err == nil {
t.Fatal("这一组失败形态不该成功")
}
second := time.Since(start)
t.Logf("第二次耗时 = %s(常态超时 %s)", second.Round(time.Millisecond), DefaultRequestTimeout)
if second > DefaultRequestTimeout+time.Second {
t.Fatalf("第二次耗时 %s 超过常态超时 %s ⇒ 放宽成了长期的", second, DefaultRequestTimeout)
}
// 放宽许可必须已经被花掉(它是一次性的,任何路径都不许给它续期)。
if emb.warmupLeft.Load() {
t.Fatal("第二次请求之后放宽许可仍为真 —— 一次性的约束被绕过了")
}
})
}
}
// TestAPIEmbedder_WarmupCoversColdStartOnce 钉放宽的**正面**作用:第一次请求
// (模型加载 / GPU 缓冲分配)可以比常态超时多等一点,只要它落在 warmupTimeout
// 之内;而这一刻过去之后,请求立刻回到常态超时(deadline 断言)。
func TestAPIEmbedder_WarmupCoversColdStartOnce(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.warmupLeft.Load() {
t.Fatal("放宽许可应在第一次请求时就被花掉")
}
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)
}
}
+23 -6
View File
@@ -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,
+17 -1
View File
@@ -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 {
+34 -6
View File
@@ -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,25 +68,38 @@ 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
// 如实上报**实际发生**的来源:空串 = 没有任何分区真正算过相似度(「没候选」)。
// 「尝试过但向量源不可用」是另一件事,由 recall_unavailable_partitions 单独给出
// (见 logDryRun 的说明与 recallLedger.unavailable)。
out.RecallMode = ledger.label()
out.DurationMS = t.now().Sub(started).Milliseconds()
// 汇总行**成败都记**(与 RunOnce 同一口径):失败那一路的
// `recall_unavailable_partitions` 正是「向量不可用」与「没候选」的唯一区分点,
// 只在成功时记就等于把这条口径丢掉。
t.logDryRun(out, ledger)
if len(failures) > 0 {
return out, fmt.Errorf("memory/tidy: dry-run finished with errors: %w", errors.Join(failures...))
}
t.logDryRun(out)
return out, nil
}
// 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)
//
// 失败时要往 ledger 记一笔 unavailable:否则「尝试过但向量源不可用」与「没有候选
// 可比」在观测面上同值(reviewer round1 的 p2)。
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 {
ledger.noteUnavailable()
t.log.Warn("memory tidy dry-run recall unavailable",
"partition", p.label(), "items", len(items), "error", err)
return nil, fmt.Errorf("%w: %w", ErrTidyRecallUnavailable, err)
@@ -92,7 +109,17 @@ func (t *Tidier) strictRecall(ctx context.Context, p tidyPartition, items []Memo
// logDryRun 输出试算的观测口径。它带 source(memory/tidy),**不带任何记忆
// 正文** —— 只记计数、分区数与截断标签。
func (t *Tidier) logDryRun(out TidyDryRun) {
//
// 🔴 `recall_mode` + `recall_unavailable_partitions` 一起才把三件事分开
// (reviewer round1 的 p2:它们曾同值):
// - mode=""、unavailable=0:本轮**没有候选可比**(压根没算过相似度);
// - mode=""、unavailable>0:**尝试过召回但向量源不可用**(严格口径不降级,
// 调用方拿到的是 `ErrTidyRecallUnavailable` ⇒ HTTP 500 `tidy_recall_unavailable`);
// - mode=embedding/lexical:真的用该来源算过。
//
// 没有为此扩 `recallMode` 的取值集合:试算失败时**不序列化** result(响应是错误
// 体),一个新取值只会出现在这行日志里,却要动一次 API 契约与文档 —— 不划算。
func (t *Tidier) logDryRun(out TidyDryRun, ledger *recallLedger) {
t.log.Info("memory tidy dry-run finished",
"rule_version", TidyRuleVersion,
"partitions", out.Stats.Partitions,
@@ -104,6 +131,7 @@ func (t *Tidier) logDryRun(out TidyDryRun) {
"veto_reasons", tidyReasonsLabel(out.Stats.VetoReasons),
"errors", out.Stats.Errors,
"recall_mode", out.RecallMode,
"recall_unavailable_partitions", ledger.unavailablePartitions(),
"duration_ms", out.DurationMS)
}
+108 -9
View File
@@ -25,8 +25,105 @@ 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
// unavailable 是**尝试过召回但向量源不可用、且本轮不降级**(试算轮的严格口径)
// 的分区数。它与 degraded 的区别是「有没有改用词项余弦」:生产轮会降级
// (degraded),试算轮只报错。
//
// 🔴 它存在的理由(reviewer round1 的 p2):试算轮在向量源不可用时
// `RecallMode` 与「没有候选可比」**同为空串**,两者在那个字段上分不开。
// 现在 provenance 由 unavailablePartitions() 单独给出,日志行能区分;
// 调用方(面板)那条路本来就有明确错误码(500 tidy_recall_unavailable),
// 因此**没有**为此扩 recallMode 的取值集合(见 logDryRun 的说明)。
unavailable 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++
}
}
// noteUnavailable 记一笔「这个分区尝试召回但向量源不可用,且本轮没有降级」
// (试算轮的严格口径:宁可报错,也不把「没算出来」说成「算过了,没有」)。
func (l *recallLedger) noteUnavailable() {
if l != nil {
l.unavailable++
}
}
// label 返回本轮**实际**的召回来源:
// - "":没有任何分区真正算过相似度(没有候选可比)—— 「没候选」不是「降级」;
// - embedding / lexical:所有算过的分区都走同一条路;
// - mixed:两种都有(多分区里有一部分降级)。
//
// ⚠️ 「尝试过但不可用」**不改变** label:它没有产出任何相似度,报 embedding /
// lexical 都是假的;它的存在由 unavailablePartitions() 单独给出(见 recallLedger
// 的 unavailable 字段说明)。
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
}
// unavailablePartitions 返回本轮「尝试召回但向量源不可用、且没有降级」的分区数。
func (l *recallLedger) unavailablePartitions() int {
if l == nil {
return 0
}
return l.unavailable
}
// tidyPair 是一条被召回的候选对。
type tidyPair struct {
a, b MemoryItem
@@ -35,18 +132,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 +159,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 +308,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,216 @@
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_UnavailableIsDistinctFromNoCandidates 钉 reviewer round1 p2:
// 试算轮「尝试过召回但向量源不可用」与「没有候选可比」**必须能区分**。
//
// 两者的 `RecallMode` 都是空串(都不曾产出相似度,报 embedding/lexical 都是假的),
// 区分点在 `recall_unavailable_partitions`:不可用时为 1,没候选时为 0。调用方
// (面板)另有一条更硬的区分 —— 失败那一路不序列化 result,直接是
// `500 tidy_recall_unavailable`。
func TestTidyDryRun_UnavailableIsDistinctFromNoCandidates(t *testing.T) {
t.Parallel()
t.Run("向量源不可用", func(t *testing.T) {
t.Parallel()
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)
out, err := tr.DryRun(context.Background())
if !errors.Is(err, ErrTidyRecallUnavailable) {
t.Fatalf("试算轮必须如实上报不可用,err = %v", err)
}
if out.RecallMode == tidyRecallEmbedding {
t.Fatalf("向量不可用却把 RecallMode 报成 embedding(配置态口径):%q", out.RecallMode)
}
line := lastLogLine(t, buf, "memory tidy dry-run finished")
if !strings.Contains(line, "recall_unavailable_partitions=1") {
t.Fatalf("汇总行必须写出「尝试过但不可用」的计数,实际:%s", line)
}
})
t.Run("没有候选可比", func(t *testing.T) {
t.Parallel()
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.DryRun(context.Background()); err != nil {
t.Fatalf("没有对可比时试算不该报错:%v", err)
}
line := lastLogLine(t, buf, "memory tidy dry-run finished")
if !strings.Contains(line, "recall_unavailable_partitions=0") {
t.Fatalf("没候选不是「不可用」,计数必须为 0,实际:%s", line)
}
if !strings.Contains(line, `recall_mode=""`) {
t.Fatalf("没候选时来源为空串,实际:%s", line)
}
})
}
// 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 ""
}