diff --git a/internal/runtime/memory/retrieve.go b/internal/runtime/memory/retrieve.go index 7250d68..bda05ce 100644 --- a/internal/runtime/memory/retrieve.go +++ b/internal/runtime/memory/retrieve.go @@ -73,12 +73,23 @@ type Retriever struct { recencyHalfLife time.Duration log *slog.Logger - // seedMu 保护一次性索引回填状态（seeded 与 fallbackWarned）： - // 回填每进程最多一次，而语义回退的告警也只报一次，避免每个 - // Retrieve 都写日志。 + // seedMu 保护一次性索引回填状态与回退路径的一次性告警位 + // （seeded / fallbackWarned / budgetWarned）：回填每进程最多一次， + // 语义回退与其预算超限的告警也只报一次，避免每个 Retrieve + // 都写日志。 seedMu sync.Mutex seeded bool fallbackWarned bool + budgetWarned bool + + // fallbackBudget 是内存余弦回退把候选正文向量化时允许花掉的 + // 时间上限（见 WithFallbackBudget）。非正时取 defaultFallbackBudget。 + fallbackBudget time.Duration + + // fallbackMu 保护回退路径的向量缓存指针，使它可按需初始化 + // （缓存本体自带锁，见 vectorCache）。 + fallbackMu sync.Mutex + fallbackCache *vectorCache } // RetrieverOption 定制由 NewRetriever 构建的 Retriever。 @@ -163,6 +174,22 @@ func WithRecencyHalfLife(days float64) RetrieverOption { } } +// WithFallbackBudget 设置内存余弦回退为「本次候选」向量化时允许花掉的 +// 时间上限。非正值被忽略，保留默认值（defaultFallbackBudget）。 +// +// 它是性能护栏而不是功能开关：持久索引正常作答时这条路径根本不执行， +// 一旦执行（索引确实缺候选向量），它把「整库候选依次向量化」的代价 +// 封在预算内 —— 超预算就用手上已有的向量打分，绝不为了补齐语义分 +// 让一个回合继续等下去（真机事故：302 条候选 = 39.3 秒，见 +// retrieve_fallback.go 的文件注释）。测试用短预算来验证这条护栏。 +func WithFallbackBudget(d time.Duration) RetrieverOption { + return func(r *Retriever) { + if d > 0 { + r.fallbackBudget = d + } + } +} + // NewRetriever 在 store 之上构建 Retriever。nil store 非法，会在 // 调用时 panic；调用方应传入非 nil 的 store。 // @@ -302,40 +329,25 @@ func (r *Retriever) visibleCandidates(ctx context.Context, personaID string) ([] return items, nil } -// keywordScores 用内存内 TF-IDF 倒排索引（embedding.Index）为每条条目 -// 与 query 打分，复用 embedding 包的分词器。得分 -// 以最大值为基准归一化到 [0,1]，以便与语义 -// 得分可比。 -func (r *Retriever) keywordScores(items []MemoryItem, query string) map[string]float64 { - scores := make(map[string]float64, len(items)) - index := embedding.NewIndex[string]() - for _, it := range items { - index.Add(keywordText(it), it.ID) - } - hits := index.Search(query) - if len(hits) == 0 { - return scores - } - max := hits[0].Score - if max <= 0 { - return scores - } - for _, h := range hits { - scores[h.ID] = h.Score / max - } - return scores -} - // semanticScores 计算每个条目对 query 的语义相关性。接入了向量 // 索引（WithVectorIndex）时，它经 idx.Search 从持久化索引 // 作答 —— 真正的向量路径（design §3）。索引没有给出任何候选命中时 -// （索引空、索引报错或 query 为空），它**回退**到内存余弦：配置了 +// （索引空或索引报错），它**回退**到内存余弦：配置了 // 增强却仍拿到一片 0 分，等于增强既无效又遮住了可用的路径 // （审计 P0-3；design/03 §3「真正启用向量检索」、design/04 §7）。 // 两条路径都没有时返回空 map，因此语义相关性 // 不贡献任何分数。 +// +// query 为空（无任何可向量化的文本）时整条语义路径**直接跳过**： +// 没有 query 就没有可比的语义方向，拿空串去向量化既问不出东西， +// 又要把整库候选逐条 embed 一遍 —— 真机实测这条弯路是 39.3 秒的来源 +// （见 retrieve_fallback.go）。此时排名由关键词与新近度决定， +// 与 Retrieve 的既有口径一致。 func (r *Retriever) semanticScores(ctx context.Context, items []MemoryItem, query string) map[string]float64 { scores := make(map[string]float64, len(items)) + if !hasSemanticQuery(query) { + return scores + } if r.vindex != nil { // 持久索引为空时先做一次性回填（每进程最多一次），使既有 // 条目也进入语义路径，而不是只有新写的条目可被检索。 @@ -350,32 +362,14 @@ func (r *Retriever) semanticScores(ctx context.Context, items []MemoryItem, quer if r.embedder == nil { return scores } + return r.inMemorySemanticScores(ctx, items, query) +} - texts := make([]string, len(items)) - for i, it := range items { - texts[i] = it.Content - } - vectors, err := r.embedder.Embed(ctx, texts) - if err != nil { - r.log.Warn("memory: semantic embed of candidates failed, semantic scores skipped", "error", err) - return scores - } - queryVectors, err := r.embedder.Embed(ctx, []string{query}) - if err != nil || len(queryVectors) == 0 { - r.log.Warn("memory: semantic embed of query failed, semantic scores skipped", "error", err) - return scores - } - qvec := queryVectors[0] - for i, vec := range vectors { - if i >= len(items) { - break - } - sim := cosineSim(qvec, vec) - if sim > 0 { - scores[items[i].ID] = sim - } - } - return scores +// hasSemanticQuery 报告 query 是否含有可供向量化的语义内容： +// 非空且非纯空白。附件类回合（主人只发文件、正文为空）正是空 query， +// 它不该触发任何向量化工作。 +func hasSemanticQuery(query string) bool { + return strings.TrimSpace(query) != "" } // vectorIndexScores 从持久化向量索引作答语义相关性： @@ -385,8 +379,10 @@ func (r *Retriever) semanticScores(ctx context.Context, items []MemoryItem, quer // 索引是**全局**的（承载所有角色/作用域的向量），而本次检索的候选 // 只是其中一个子集：不属于候选的命中不参与排名，也不得充当 // 归一化基准（也不得因占满 topK 名额而把候选挤出命中）。因此 -// 这里只保留候选集内的命中。query 为空、索引报错或候选内 -// 无命中时退化为空 map —— 调用方据此回退内存余弦。 +// 这里只保留候选集内的命中。索引报错或候选内无命中时退化为 +// 空 map —— 调用方据此回退内存余弦。query 为空的早退只是防御： +// 语义路径的调用方（semanticScores）已经在更外层拦掉了空 query， +// 不会再把「没问过索引」误当成「索引答不出」。 func (r *Retriever) vectorIndexScores(ctx context.Context, items []MemoryItem, query string) map[string]float64 { scores := make(map[string]float64, len(items)) if strings.TrimSpace(query) == "" { @@ -447,14 +443,6 @@ func (r *Retriever) projectFor(ctx context.Context, it MemoryItem) *projectindex return p } -// keywordText 是用于关键词匹配的索引文本：content 加 tags。 -func keywordText(it MemoryItem) string { - if len(it.Tags) == 0 { - return it.Content - } - return it.Content + " " + strings.Join(it.Tags, " ") -} - // Inject 为 personaID 检索最多 topK 条相关记忆，并把它们格式化为 // 紧凑的上下文块，可供注入提示词。没有检索到任何内容时 // 返回空字符串，因此没有相关记忆时调用方可以忽略 diff --git a/internal/runtime/memory/retrieve_fallback.go b/internal/runtime/memory/retrieve_fallback.go new file mode 100644 index 0000000..fd00a69 --- /dev/null +++ b/internal/runtime/memory/retrieve_fallback.go @@ -0,0 +1,199 @@ +package memory + +// 本文件收拢语义路径的**内存余弦回退**：持久向量索引没给出候选命中时， +// 用内存余弦补齐语义分（索引正常作答时这里一行都不执行）。 +// +// 它为什么必须是有界的（2026-10-05 真机事故）：这条路径原先一次性把 +// **整库候选正文**塞进一个 embedding 请求，再接一次 query。主人真机 +// 只发了一个不带正文的附件（空 query），索引因此「没答上」，回退于是 +// 把 302 条候选正文全部向量化 —— 单个回合的 memory.recall 花了 39.3 秒， +// 而模型本身只用 2.6 秒。因此这里给每一处代价都封了顶： +// +// - 单请求文本条数上限（fallbackEmbedBatch）：既不把一个巨大的 +// 请求丢给远端接口，也让「回退一共嵌入了多少条」可数； +// - 每次回退的时间预算（defaultFallbackBudget / WithFallbackBudget）： +// 超预算就用手上已有的向量打分，其余条目本轮没有语义分 —— +// 回退是兜底，绝不允许它把回合拖住； +// - 正文→向量的进程内缓存（fallbackVectorCacheMax）：同一批正文 +// 不再每回合重复向量化（embedder 对相同输入是确定性的）。 +// +// 空 query 在更外层就被拦掉（semanticScores 的 hasSemanticQuery 门）， +// 因此真机那次「整库向量化」不会再发生：没有 query 就没有语义方向， +// 排名交给关键词与新近度。 + +import ( + "context" + "sync" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding" +) + +const ( + // fallbackEmbedBatch 是回退路径单次 Embed 携带的候选条数上限。 + // 取 32：常见的 OpenAI 兼容 /embeddings 接口容许的批量都远大于它， + // 而它小到不会让单次请求本身成为长尾。 + fallbackEmbedBatch = 32 + + // defaultFallbackBudget 是单次回退把候选正文向量化时允许花掉的 + // 时间上限。取 1 秒：主人对「一个回合的记忆召回」的要求是亚秒级， + // 回退作为兜底绝不能把这笔代价变成无界的等待 —— 宁可本轮少补一些 + // 语义分（并留下 Warn），也不让回合变慢。索引正常作答时用不到它。 + defaultFallbackBudget = 1 * time.Second + + // fallbackVectorCacheMax 是回退向量缓存的条目上限。1024 条的 + // 最坏内存占用约 1024 × 1024 维 × 4 字节 ≈ 4 MiB（维度由 + // embedder 决定），对常驻进程可接受；超出按 FIFO 淘汰。 + fallbackVectorCacheMax = 1024 +) + +// inMemorySemanticScores 是索引答不出候选命中时的语义兜底：先取 query +// 向量（代价与索引路径相同，都是一次 embed），再用候选正文与它的余弦 +// 相似度打分。候选向量的取得是**有界**的（见 candidateVectors）。 +// +// query 向量embed失败时返回空分值：没有 query 向量，余弦无从谈起 —— +// 但候选正文一个字节都没有发出去。 +func (r *Retriever) inMemorySemanticScores(ctx context.Context, items []MemoryItem, query string) map[string]float64 { + scores := make(map[string]float64, len(items)) + queryVectors, err := r.embedder.Embed(ctx, []string{query}) + if err != nil || len(queryVectors) == 0 { + r.log.Warn("memory: semantic embed of query failed, semantic scores skipped", "error", err) + return scores + } + qvec := queryVectors[0] + + vectors := r.candidateVectors(ctx, items) + for _, it := range items { + vec, ok := vectors[it.Content] + if !ok { + continue + } + if sim := cosineSim(qvec, vec); sim > 0 { + scores[it.ID] = sim + } + } + return scores +} + +// candidateVectors 返回候选条目的「正文 → 向量」映射：正文命中进程内 +// 缓存的条目直接复用，只有未命中的才在预算内分批向量化。因此同一批 +// 候选在第二回合起几乎不需要再调 embedder（缓存未满时），真机事故里 +// 那种「每回合把整库重算一遍」不再发生。 +func (r *Retriever) candidateVectors(ctx context.Context, items []MemoryItem) map[string]embedding.Vector { + cache := r.vectorCache() + out := make(map[string]embedding.Vector, len(items)) + missing := make([]MemoryItem, 0, len(items)) + for _, it := range items { + if v, ok := cache.get(it.Content); ok { + out[it.Content] = v + continue + } + missing = append(missing, it) + } + if len(missing) > 0 { + r.embedCandidates(ctx, missing, out, cache) + } + return out +} + +// embedCandidates 分批向量化 missing 里的正文，把结果写进 out 与缓存。 +// +// 预算用尽（或调用方取消）时它**就地收手**：out 里已有的向量照旧参与 +// 本轮打分，剩下的条目只是没有语义分 —— 绝不为了补齐它们继续阻塞检索。 +// 单批失败按整段失败处理（把半批结果当完整结果用，会让同一份语义分 +// 随批次边界漂移），并记一条带来源的日志。 +func (r *Retriever) embedCandidates(ctx context.Context, missing []MemoryItem, out map[string]embedding.Vector, cache *vectorCache) { + bctx, cancel := context.WithTimeout(ctx, r.embedBudget()) + defer cancel() + + embedded := 0 + for start := 0; start < len(missing); start += fallbackEmbedBatch { + end := min(start+fallbackEmbedBatch, len(missing)) + batch := missing[start:end] + texts := make([]string, len(batch)) + for i, it := range batch { + texts[i] = it.Content + } + vecs, err := r.embedder.Embed(bctx, texts) + if err != nil { + if ctx.Err() != nil { + return + } + if bctx.Err() != nil { + r.warnFallbackBudget(len(missing), embedded) + return + } + r.log.Warn("memory: semantic embed of candidates failed, semantic scores skipped", + "error", err, "candidates", len(missing)) + return + } + for i, v := range vecs { + if i >= len(batch) { + break + } + out[batch[i].Content] = v + cache.put(batch[i].Content, v) + embedded++ + } + } +} + +// vectorCache 返回回退路径的向量缓存，按需初始化（零值 Retriever 也能用）。 +func (r *Retriever) vectorCache() *vectorCache { + r.fallbackMu.Lock() + defer r.fallbackMu.Unlock() + if r.fallbackCache == nil { + r.fallbackCache = newVectorCache(fallbackVectorCacheMax) + } + return r.fallbackCache +} + +// embedBudget 返回回退路径的时间预算，未配置时取默认值。 +func (r *Retriever) embedBudget() time.Duration { + if r.fallbackBudget > 0 { + return r.fallbackBudget + } + return defaultFallbackBudget +} + +// vectorCache 是「正文 → 向量」的有界 FIFO 缓存。键就是将要发给 +// embedder 的那段文本本身，因此内容变了自然换键，不会命中旧向量 +// （embedder 对相同输入确定性，缓存因此与实时计算等价）。 +type vectorCache struct { + mu sync.Mutex + vecs map[string]embedding.Vector + order []string + limit int +} + +// newVectorCache 构建容量为 limit 的缓存；非正 limit 取默认上限。 +func newVectorCache(limit int) *vectorCache { + if limit <= 0 { + limit = fallbackVectorCacheMax + } + return &vectorCache{vecs: make(map[string]embedding.Vector, limit), limit: limit} +} + +// get 返回 key 的向量；未命中返回 false。 +func (c *vectorCache) get(key string) (embedding.Vector, bool) { + c.mu.Lock() + defer c.mu.Unlock() + v, ok := c.vecs[key] + return v, ok +} + +// put 写入 key 的向量，已存在则保留先写入的那份（同一正文的向量相同， +// 重写只会白费一次淘汰）。超过上限时按写入顺序淘汰最旧的一条。 +func (c *vectorCache) put(key string, v embedding.Vector) { + c.mu.Lock() + defer c.mu.Unlock() + if _, ok := c.vecs[key]; ok { + return + } + if len(c.order) >= c.limit { + delete(c.vecs, c.order[0]) + c.order = c.order[1:] + } + c.vecs[key] = v + c.order = append(c.order, key) +} diff --git a/internal/runtime/memory/retrieve_fallback_test.go b/internal/runtime/memory/retrieve_fallback_test.go new file mode 100644 index 0000000..398c2a3 --- /dev/null +++ b/internal/runtime/memory/retrieve_fallback_test.go @@ -0,0 +1,219 @@ +package memory + +import ( + "bytes" + "context" + "log/slog" + "strconv" + "strings" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding" +) + +// 本文件是 2026-10-05 真机性能事故的回归测试：主人只发了一个**不带正文**的 +// 附件（空 query），memory.recall 一次花了 39.3 秒 —— 索引因为空 query +// 没被问过，回退路径于是把整库 302 条候选正文逐条向量化。三条断言分别 +// 钉住「空 query 不做任何语义工作」「回退不再把整库塞进一次请求且不重复 +// 嵌入」「回退被预算拦住、不阻塞回合」。 + +// recordingEmbedder 记录每次 Embed 收到的文本条数，并返回确定性向量： +// 它是「回退路径到底嵌入了多少条正文」的观测点。 +type recordingEmbedder struct { + dim int + + mu sync.Mutex + batches []int + texts int +} + +func (e *recordingEmbedder) Embed(_ context.Context, texts []string) ([]embedding.Vector, error) { + e.mu.Lock() + e.batches = append(e.batches, len(texts)) + e.texts += len(texts) + e.mu.Unlock() + + out := make([]embedding.Vector, len(texts)) + for i := range texts { + v := make(embedding.Vector, e.Dim()) + v[0] = 1 + out[i] = v + } + return out, nil +} + +func (e *recordingEmbedder) Dim() int { + if e.dim > 0 { + return e.dim + } + return 4 +} + +// batchSizes 返回到目前为止每次 Embed 的文本条数（副本）。 +func (e *recordingEmbedder) batchSizes() []int { + e.mu.Lock() + defer e.mu.Unlock() + return append([]int(nil), e.batches...) +} + +// blockingBatchEmbedder 模拟「远端 embedding 接口对一批文本迟迟不回」： +// 单条文本立即返回，一次多条则一直等到 ctx 结束 —— 用来验证回退路径 +// 真的会收手，而不是陪着接口一起把回合拖住。 +type blockingBatchEmbedder struct{ dim int } + +func (e blockingBatchEmbedder) Embed(ctx context.Context, texts []string) ([]embedding.Vector, error) { + if len(texts) > 1 { + <-ctx.Done() + return nil, ctx.Err() + } + out := make([]embedding.Vector, len(texts)) + for i := range texts { + v := make(embedding.Vector, e.dim) + v[0] = 1 + out[i] = v + } + return out, nil +} + +func (e blockingBatchEmbedder) Dim() int { return e.dim } + +// TestRetriever_EmptyQuery_NoSemanticWork_空查询不做任何语义工作 是事故的 +// 主回归：空 query（附件类回合没有正文）既不该去问向量索引，也不该把 +// 候选正文送去向量化。旧行为下这里会把每个候选正文都 embed 一遍、 +// 真机 302 条即 39.3 秒。 +func TestRetriever_EmptyQuery_NoSemanticWork_空查询不做任何语义工作(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + + // 候选条数跨过回退的单批上限：一旦这条路径被触发，嵌入条数会明显 + // 大于「一次 query 向量」，断言因此不会因边界而假绿。 + candidates := fallbackEmbedBatch + 8 + for i := 0; i < candidates; i++ { + addItem(t, store, MemoryItem{ + PersonaID: "persona.a", + Content: "记忆条目 " + strconv.Itoa(i), + Scope: ScopePersona, + Kind: KindWorldEvent, + }) + } + + emb := &recordingEmbedder{dim: 8} + // 索引恒「无候选命中」：这正是事故里那条回退路径的入口。 + idx := &countingIndex{Index: &mockVectorIndex{dim: 8}} + r := NewRetriever(store, WithEmbedder(emb), WithVectorIndex(idx)) + + for _, query := range []string{"", " ", "\n\t"} { + results, err := r.Retrieve(ctx, "persona.a", query, 5) + if err != nil { + t.Fatalf("Retrieve(query=%q): %v", query, err) + } + if len(results) == 0 { + t.Fatalf("query=%q：空查询仍应按新近度返回候选，got 0 results", query) + } + } + if emb.texts != 0 { + t.Fatalf("空 query 向量化了 %d 条文本，want 0（真机事故：302 条候选正文 = 39.3 秒）", emb.texts) + } + if idx.searches != 0 { + t.Fatalf("空 query 查询了向量索引 %d 次，want 0（空 query 必然「无候选命中」，只会把正常检索误报成索引失效）", + idx.searches) + } +} + +// TestRetriever_Fallback_BoundedAndCached_回退分批且不重复嵌候选 钉住回退 +// 路径的两条护栏：候选正文分批送出（不再一个请求塞整库），且同一批 +// 正文在第二回合不再重复向量化（embedder 对相同输入确定性）。 +func TestRetriever_Fallback_BoundedAndCached_回退分批且不重复嵌候选(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + + candidates := fallbackEmbedBatch*2 + 5 + for i := 0; i < candidates; i++ { + content := "coffee note " + strconv.Itoa(i) + addItem(t, store, MemoryItem{ + PersonaID: "persona.a", Content: content, + Scope: ScopePersona, Kind: KindWorldEvent, + }) + } + + emb := &recordingEmbedder{dim: 8} + idx := &countingIndex{Index: &mockVectorIndex{dim: 8}} + r := NewRetriever(store, WithEmbedder(emb), WithVectorIndex(idx)) + + results, err := r.Retrieve(ctx, "persona.a", "coffee", 5) + if err != nil { + t.Fatalf("Retrieve: %v", err) + } + if len(results) == 0 || results[0].Score <= 0 { + t.Fatalf("回退没有给出任何语义分：results=%d score=%v", len(results), results[0].Score) + } + + batches := emb.batchSizes() + if len(batches) < 2 { + t.Fatalf("候选正文没有分批：一次 Embed 带了 %v 条，want 多次且每次 ≤ %d", batches, fallbackEmbedBatch) + } + for _, n := range batches { + if n > fallbackEmbedBatch { + t.Fatalf("单次 Embed 携带 %d 条文本，超过上限 %d（回退不得把整库塞进一个请求）", + n, fallbackEmbedBatch) + } + } + if got, want := emb.texts, candidates+1; got != want { + t.Fatalf("首次回退向量化了 %d 条文本，want %d（%d 条候选分批 + 1 条 query）", got, want, candidates) + } + + // 第二次检索：候选正文全部命中缓存，因此只多一次 query 向量化。 + if _, err := r.Retrieve(ctx, "persona.a", "coffee", 5); err != nil { + t.Fatalf("second Retrieve: %v", err) + } + if got, want := emb.texts, candidates+2; got != want { + t.Fatalf("第二次回退向量化了 %d 条文本（累计），want %d —— 候选正文被重复嵌入，缓存没生效", + got, want) + } +} + +// TestRetriever_Fallback_Budget_超预算不阻塞回合 钉住预算护栏：embedding +// 接口对一批文本迟迟不回时，回退用手上已有的向量收手，回合不被拖住， +// 并且留下一条带来源的 Warn（语义分只补了一部分必须可见）。 +func TestRetriever_Fallback_Budget_超预算不阻塞回合(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + + for i := 0; i < fallbackEmbedBatch+1; i++ { + addItem(t, store, MemoryItem{ + PersonaID: "persona.a", Content: "记忆条目 " + strconv.Itoa(i), + Scope: ScopePersona, Kind: KindWorldEvent, + }) + } + + var buf bytes.Buffer + logger := slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug})) + idx := &countingIndex{Index: &mockVectorIndex{dim: 8}} + r := NewRetriever(store, + WithEmbedder(blockingBatchEmbedder{dim: 8}), + WithVectorIndex(idx), + WithRetrieverLogger(logger), + WithFallbackBudget(50*time.Millisecond), + ) + + start := time.Now() + results, err := r.Retrieve(ctx, "persona.a", "咖啡", 5) + elapsed := time.Since(start) + if err != nil { + t.Fatalf("Retrieve: %v", err) + } + if len(results) == 0 { + t.Fatal("回退超预算不应丢掉检索结果（关键词与新近度仍在）") + } + if elapsed > 5*time.Second { + t.Fatalf("回退没有被预算拦住：检索耗时 %v", elapsed) + } + if !strings.Contains(buf.String(), "exceeded its embed budget") { + t.Fatalf("回退超预算没有告警，语义分只补了一部分却无人可见；日志=%q", buf.String()) + } +} diff --git a/internal/runtime/memory/retrieve_keyword.go b/internal/runtime/memory/retrieve_keyword.go new file mode 100644 index 0000000..4c1146f --- /dev/null +++ b/internal/runtime/memory/retrieve_keyword.go @@ -0,0 +1,43 @@ +package memory + +import ( + "strings" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding" +) + +// 本文件收拢排名里的**关键词侧**：TF-IDF 打分与用于匹配的索引文本。 +// 它们与语义侧（retrieve_semantic.go / retrieve_fallback.go）对称， +// 拆开的目的是让 retrieve.go 保持可读（文件 ≤500 行，AGENTS §10）。 + +// keywordScores 用内存内 TF-IDF 倒排索引（embedding.Index）为每条条目 +// 与 query 打分，复用 embedding 包的分词器。得分 +// 以最大值为基准归一化到 [0,1]，以便与语义 +// 得分可比。 +func (r *Retriever) keywordScores(items []MemoryItem, query string) map[string]float64 { + scores := make(map[string]float64, len(items)) + index := embedding.NewIndex[string]() + for _, it := range items { + index.Add(keywordText(it), it.ID) + } + hits := index.Search(query) + if len(hits) == 0 { + return scores + } + max := hits[0].Score + if max <= 0 { + return scores + } + for _, h := range hits { + scores[h.ID] = h.Score / max + } + return scores +} + +// keywordText 是用于关键词匹配的索引文本：content 加 tags。 +func keywordText(it MemoryItem) string { + if len(it.Tags) == 0 { + return it.Content + } + return it.Content + " " + strings.Join(it.Tags, " ") +} diff --git a/internal/runtime/memory/retrieve_semantic.go b/internal/runtime/memory/retrieve_semantic.go index 4da51ab..d29e69e 100644 --- a/internal/runtime/memory/retrieve_semantic.go +++ b/internal/runtime/memory/retrieve_semantic.go @@ -6,9 +6,9 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding" ) -// 本文件收拢语义打分的辅助件：内存余弦相似度与「索引没答上、改走 -// 内存回退」的一次性告警。它们的用量小但分散，独立成文件是为了让 -// retrieve.go 保持可读（文件 ≤500 行，AGENTS §10）。 +// 本文件收拢语义打分的辅助件：内存余弦相似度与语义路径的 +// 一次性告警（回退发生、回退超预算）。它们的用量小但分散， +// 独立成文件是为了让 retrieve.go 保持可读（文件 ≤500 行，AGENTS §10）。 // cosineSim 计算两个向量的余弦相似度，由调用方夹到 [0,1]。 // 长度为零或不一致的向量得 0。 @@ -42,3 +42,20 @@ func (r *Retriever) warnSemanticFallback() { r.fallbackWarned = true r.log.Debug("memory: vector index had no candidate hit, using in-memory cosine fallback") } + +// warnFallbackBudget 报告一次「内存余弦回退没能在预算内把候选正文 +// 向量化完，本轮语义分只有一部分」。它是 Warn 而不是 Debug：语义分 +// 少了一部分会静默改变排名，运维必须看得见（AGENTS §10 日志带来源）。 +// 同样每个检索器只报一次 —— 该状态在一次运行里基本不变（索引缺向量 +// 或 embedder 慢），逐次告警只会把日志刷满；条数由日志字段给出， +// 需要细看时看 candidates/embedded。 +func (r *Retriever) warnFallbackBudget(candidates, embedded int) { + r.seedMu.Lock() + defer r.seedMu.Unlock() + if r.budgetWarned { + return + } + r.budgetWarned = true + r.log.Warn("memory: in-memory cosine fallback exceeded its embed budget, semantic scores partial", + "candidates", candidates, "embedded", embedded, "budget", r.embedBudget()) +}