From a162e2769ef32bccbca2851786b4c6e68e9425bc Mon Sep 17 00:00:00 2001 From: xiaoyi Date: Mon, 12 Oct 2026 09:23:09 +0800 Subject: [PATCH] =?UTF-8?q?perf(distill):=20=E6=89=93=E5=88=86=E7=BB=93?= =?UTF-8?q?=E6=9E=9C=E6=8C=89=E5=86=85=E5=AE=B9=E5=93=88=E5=B8=8C=E8=AE=B0?= =?UTF-8?q?=E5=BF=86=E5=8C=96=20=E2=80=94=E2=80=94=20=E7=9C=81=E6=8E=89?= =?UTF-8?q?=E7=BA=A6=2060%=20=E7=9A=84=202B=20=E8=B0=83=E7=94=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 问题（实测）： - episodes 表重复率 78.5%（近 2 天 58.4%，今天 64.3%）；同一条内容一天被写 9 次 - 干净窗口 392 条评分里**只有 146 条唯一** ⇒ 约 60% 的 2B 打分调用打在重复内容上 为什么不是批内去重：观测到的重复是「同内容一天 ×9」，**分散在不同批次** （每批 1~15 条），批内去重抓不到。对症方案是按内容哈希记忆化 —— 同内容只调一次 LLM，之后复用；批内/跨批/跨重启一并覆盖。 改动： 1) 新增 distill/score_cache.go：sha256(content) → LLMResponse 的 FIFO 有界缓存 （cap 2 万条），**落盘**（原子写：tmp+rename）——服务今天已重启 6 次， 纯内存缓存覆盖率永远上不去。损坏文件只从空开始，不阻断蒸馏。 2) engine.go：原 callLLM5D 改名 callLLM5DUncached（**函数体零改动**）， 外层新增 callLLM5D 做「查缓存 → 未命中再调 → 回填」。 禁用开关 ZHIYI_SCORE_CACHE=0。 3) flush 收尾打印累计命中率并落盘（每轮一次，避免频繁 IO）。 零风险论证：只复用「同一输入的上一次输出」，不改变任何判分逻辑； 最终写入 memories 时本来就去重（实测 memories 重复率 0.3%）， 所以跳过重复项的打分**不改变最终写入结果**。 测试：新增 8 个用例（命中/不同内容不污染/淘汰边界/落盘往返/损坏兜底/ 哈希稳定/nil 安全/端到端不打第二次 HTTP），全部通过。 决定性判据 = 端到端：同内容第二次调用后 HTTP 请求数**不增加** （证据：首次后=2 同内容后=2 不同内容后=3）。 ⚠️ 过程中发现两件必须记下的事： ① 测试守护：本包 TestMain 默认**关闭**缓存 —— 有 4 个测试用同一内容反复调 callLLM5D 验证重试/解析行为，缓存会把第二次短路，使它们**静默失去覆盖** （表面全绿）。首版没关，4 个测试当场挂掉才暴露。 ② 判据写错：首版断言绝对计数 '==1'，被 NewEngine 的并发探活 （RefreshConcurrency 打一次端点）干扰成 2 而误报失败 —— 是判据错，非缓存失效。 改为相对计数后通过。 --- go/internal/distill/engine.go | 28 +++ go/internal/distill/main_test.go | 19 ++ go/internal/distill/score_cache.go | 183 ++++++++++++++++++++ go/internal/distill/score_cache_e2e_test.go | 75 ++++++++ go/internal/distill/score_cache_test.go | 135 +++++++++++++++ 5 files changed, 440 insertions(+) create mode 100644 go/internal/distill/main_test.go create mode 100644 go/internal/distill/score_cache.go create mode 100644 go/internal/distill/score_cache_e2e_test.go create mode 100644 go/internal/distill/score_cache_test.go diff --git a/go/internal/distill/engine.go b/go/internal/distill/engine.go index 7a018fd..cdc60f2 100644 --- a/go/internal/distill/engine.go +++ b/go/internal/distill/engine.go @@ -491,6 +491,17 @@ func (e *Engine) flush() { e.mu.Unlock() log.Printf("[distill] flush DONE: processed %d items (concurrency=%d)", len(batch), effConcurrency) + + // 打分缓存汇总（2026-10-12）：命中率直接等于「本轮省下的 2B 调用占比」。 + // 落盘放这里（每轮一次）而不是每条一次，避免频繁 IO。 + if c := getScoreCache(); c != nil { + h, m := c.stats() + if h+m > 0 { + log.Printf("[scorecache] 累计 命中=%d 未命中=%d 命中率=%.1f%%（命中=省下的 2B 调用）", + h, m, float64(h)/float64(h+m)*100) + } + c.save() + } } // distillOne 蒸馏单条 @@ -612,7 +623,24 @@ func (e *Engine) distillOne(input DistillInput) DistillResult { } // callLLM5D 调用 LLM 进行 5维评估 + 实体/事实提取 +// callLLM5D 打分入口 —— 先查内容哈希缓存，命中直接复用（不调 2B）。 +// +// 2026-10-12：episodes 重复率 78.5%、约 60% 的打分调用打在重复内容上（详见 score_cache.go）。 +// 本层只做记忆化，**不改变任何判分逻辑**；禁用见 ZHIYI_SCORE_CACHE=0。 func (e *Engine) callLLM5D(content string) (LLMResponse, error) { + c := getScoreCache() + if r, ok := c.get(content); ok { + return r, nil + } + r, err := e.callLLM5DUncached(content) + if err == nil { + c.put(content, r) + } + return r, err +} + +// callLLM5DUncached 原始打分实现（函数体未改动，仅改名，便于外层套缓存）。 +func (e *Engine) callLLM5DUncached(content string) (LLMResponse, error) { // 2026-09-14 限制 distill prompt ≤8k token，防止打爆本地 2B 单槽 content, sentTokens, origTokens := limitDistillContent(content) if origTokens > maxDistillTokens { diff --git a/go/internal/distill/main_test.go b/go/internal/distill/main_test.go new file mode 100644 index 0000000..8e266fb --- /dev/null +++ b/go/internal/distill/main_test.go @@ -0,0 +1,19 @@ +package distill + +import ( + "os" + "testing" +) + +// TestMain 默认关闭打分缓存。 +// +// 为什么必须关：TestCallLLM5D_LineProtocolParse / BackoffOn429 / BackoffOn5xx / +// FlushPeakRespectsSlotMatchedConcurrency 都用**同一内容**反复调 callLLM5D， +// 靠第二次调用验证「重试」「解析」行为。缓存一旦生效，第二次直接被短路返回， +// 这些测试就**静默失去了覆盖**（表面全绿，实则没测到东西）。 +// +// 缓存本身的正确性由 score_cache_test.go 直接测结构体（不依赖全局实例）。 +func TestMain(m *testing.M) { + scoreCacheDisabled = true + os.Exit(m.Run()) +} diff --git a/go/internal/distill/score_cache.go b/go/internal/distill/score_cache.go new file mode 100644 index 0000000..39edd8a --- /dev/null +++ b/go/internal/distill/score_cache.go @@ -0,0 +1,183 @@ +// 织忆 MemoryWeave — 打分结果记忆化缓存（2026-10-12） +// +// 要治的问题（实测）：episodes 表**重复率 78.5%**（近 2 天 58.4%，今天 64.3%）， +// 同一条内容一天被写成 9 次；干净窗口 392 条评分里**只有 146 条唯一** ⇒ +// **约 60% 的 2B 打分调用打在重复内容上，纯烧算力**。 +// +// 为什么不是「批内去重」：观测到的重复是「同一条内容一天 ×9」，**分散在不同批次** +// （每批约 1~15 条），批内去重抓不到。真正对症的是**按内容哈希记忆化** —— +// 同一内容永远只调一次 LLM，之后复用上次结果；批内/跨批/跨重启一并覆盖。 +// +// 为什么零风险：只是复用「同一输入的上一次输出」。 +// ① 最终写入 memories 时本来就会去重（实测 memories 重复率 0.3%）， +// +// 所以跳过重复项的打分**不改变最终写入结果**； +// +// ② 关掉即恢复原行为：ZHIYI_SCORE_CACHE=0。 +// +// 为什么必须落盘：服务重启频繁（今天已重启 5 次），纯内存缓存一重启就清零， +// 覆盖率永远上不去。 +package distill + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "log" + "os" + "path/filepath" + "sync" +) + +const ( + // scoreCacheCap 上限条数。一条 LLMResponse 序列化约 0.3~1KB ⇒ 2 万条约 6~20MB，可接受。 + scoreCacheCap = 20000 + // scoreCachePath 落盘位置（与 LanceDB 同目录，服务有写权限）。 + scoreCachePath = "/var/lib/memoryweave/distill-score-cache.json" +) + +type scoreCache struct { + mu sync.Mutex + m map[string]LLMResponse + order []string // 插入顺序，用于 FIFO 淘汰 + path string + cap int + hits int64 + miss int64 + dirty bool +} + +var ( + scoreCacheOnce sync.Once + globalScoreCache *scoreCache + // scoreCacheDisabled 关闭开关。生产由环境变量控制；**单元测试默认置 true** + // （见 score_cache_test.go 的 TestMain）——因为多个测试用同一内容反复调 + // callLLM5D 验证重试/解析行为，缓存会把第二次调用短路，使它们失去覆盖。 + scoreCacheDisabled = os.Getenv("ZHIYI_SCORE_CACHE") == "0" + // scoreCacheFilePath 落盘路径，测试可覆盖为临时目录（测试不得读写生产缓存文件）。 + scoreCacheFilePath = scoreCachePath +) + +func getScoreCache() *scoreCache { + scoreCacheOnce.Do(func() { + if scoreCacheDisabled { + log.Printf("[scorecache] 已禁用（环境变量 ZHIYI_SCORE_CACHE=0 或测试环境）") + return + } + c := &scoreCache{m: map[string]LLMResponse{}, path: scoreCacheFilePath, cap: scoreCacheCap} + c.load() + globalScoreCache = c + }) + return globalScoreCache +} + +func scoreKey(content string) string { + sum := sha256.Sum256([]byte(content)) + return hex.EncodeToString(sum[:]) +} + +// load 从磁盘恢复；文件不存在/损坏都只是「从空开始」，不阻断蒸馏。 +func (c *scoreCache) load() { + b, err := os.ReadFile(c.path) + if err != nil { + log.Printf("[scorecache] 无历史缓存（%v），从空开始", err) + return + } + var payload struct { + Order []string `json:"order"` + Items map[string]LLMResponse `json:"items"` + } + if err := json.Unmarshal(b, &payload); err != nil { + log.Printf("[scorecache] ⚠️ 缓存文件解析失败，从空开始（不阻断）: %v", err) + return + } + c.m = payload.Items + if c.m == nil { + c.m = map[string]LLMResponse{} + } + c.order = payload.Order + log.Printf("[scorecache] 已载入 %d 条历史打分（%s）", len(c.m), c.path) +} + +// save 落盘（原子写：先写临时文件再 rename，避免写一半被重启切成坏文件）。 +// 判空：ZHIYI_SCORE_CACHE=0 时全局实例为 nil，任何方法都不得 panic。 +func (c *scoreCache) save() { + if c == nil { + return + } + c.mu.Lock() + if !c.dirty { + c.mu.Unlock() + return + } + payload := struct { + Order []string `json:"order"` + Items map[string]LLMResponse `json:"items"` + }{c.order, c.m} + c.dirty = false + c.mu.Unlock() + + b, err := json.Marshal(payload) + if err != nil { + log.Printf("[scorecache] 序列化失败: %v", err) + return + } + if err := os.MkdirAll(filepath.Dir(c.path), 0o755); err != nil { + log.Printf("[scorecache] 建目录失败: %v", err) + return + } + tmp := c.path + ".tmp" + if err := os.WriteFile(tmp, b, 0o644); err != nil { + log.Printf("[scorecache] 写临时文件失败: %v", err) + return + } + if err := os.Rename(tmp, c.path); err != nil { + log.Printf("[scorecache] rename 失败: %v", err) + } +} + +func (c *scoreCache) get(content string) (LLMResponse, bool) { + if c == nil { + return LLMResponse{}, false + } + k := scoreKey(content) + c.mu.Lock() + defer c.mu.Unlock() + if r, ok := c.m[k]; ok { + c.hits++ + return r, true + } + c.miss++ + return LLMResponse{}, false +} + +func (c *scoreCache) put(content string, r LLMResponse) { + if c == nil { + return + } + k := scoreKey(content) + c.mu.Lock() + defer c.mu.Unlock() + if _, exists := c.m[k]; exists { + return + } + c.m[k] = r + c.order = append(c.order, k) + c.dirty = true + // FIFO 淘汰：超上限时丢最老的，直到回到上限。 + for len(c.order) > c.cap { + old := c.order[0] + c.order = c.order[1:] + delete(c.m, old) + } +} + +// stats 返回 (命中, 未命中) 快照，供每轮 flush 汇总打印。 +func (c *scoreCache) stats() (int64, int64) { + if c == nil { + return 0, 0 + } + c.mu.Lock() + defer c.mu.Unlock() + return c.hits, c.miss +} diff --git a/go/internal/distill/score_cache_e2e_test.go b/go/internal/distill/score_cache_e2e_test.go new file mode 100644 index 0000000..8723b50 --- /dev/null +++ b/go/internal/distill/score_cache_e2e_test.go @@ -0,0 +1,75 @@ +package distill + +import ( + "net/http" + "net/http/httptest" + "path/filepath" + "sync" + "sync/atomic" + "testing" +) + +// TestScoreCacheSkipsSecondLLMCall —— 决定性验证：**同内容第二次调用不得增加网络请求**。 +// +// 「省下约 60% 的 2B 调用」这句话，等价于「第二次遇到同内容时 HTTP 请求数不增加」。 +// 只测缓存结构体（不碰网络）证明不了这一点，必须打真 HTTP 桩数请求数。 +// +// ⚠️ 断言必须用**相对计数**：NewEngine 会走 RefreshConcurrency 探一次端点 +// （日志 total_slots=…），那次探活也计入桩的请求数。首版用绝对值 "==1" 判， +// 被探活干扰成 2 而误报失败 —— 是**判据写错**，不是缓存失效。 +func TestScoreCacheSkipsSecondLLMCall(t *testing.T) { + // 本包 TestMain 默认关缓存（保护重试/解析类测试），这里显式打开。 + oldDisabled, oldOnce, oldInst, oldPath := scoreCacheDisabled, scoreCacheOnce, globalScoreCache, scoreCacheFilePath + scoreCacheDisabled = false + scoreCacheOnce = sync.Once{} + globalScoreCache = nil + scoreCacheFilePath = filepath.Join(t.TempDir(), "c.json") + t.Cleanup(func() { + scoreCacheDisabled, scoreCacheOnce, globalScoreCache = oldDisabled, oldOnce, oldInst + scoreCacheFilePath = oldPath + }) + + var calls int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&calls, 1) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(testLLMBody)) + })) + defer srv.Close() + + e := NewEngine(srv.URL, "test-model", "test-key") + const dup = "牵挂: 待处理: 这条内容今天已经被写过 9 次" + + // ① 首次：真调（计数基线含探活） + if _, err := e.callLLM5D(dup); err != nil { + t.Fatalf("首次调用失败: %v", err) + } + after1 := atomic.LoadInt32(&calls) + if after1 == 0 { + t.Fatal("首次调用竟未发出任何 HTTP 请求") + } + + // ② 同内容再来一次：**计数不得增加**（这就是省下的那一次 2B 调用） + if _, err := e.callLLM5D(dup); err != nil { + t.Fatalf("第二次调用失败: %v", err) + } + after2 := atomic.LoadInt32(&calls) + if after2 != after1 { + t.Fatalf("❌ 同内容第二次仍打了 LLM：%d → %d（应保持不变）", after1, after2) + } + + // ③ 不同内容：必须真调（不能把不同内容也当命中 —— 那会给出错误分数） + if _, err := e.callLLM5D("另一条完全不同的内容"); err != nil { + t.Fatalf("第三次调用失败: %v", err) + } + after3 := atomic.LoadInt32(&calls) + if after3 <= after2 { + t.Fatalf("❌ 不同内容应真调：%d → %d（应增加）", after2, after3) + } + + // ④ 命中计数与三次调用的语义一致：3 次调用里恰好 1 次命中 + if h, _ := getScoreCache().stats(); h != 1 { + t.Fatalf("应恰好 1 次命中，实际 %d", h) + } + t.Logf("证据：请求数 首次后=%d 同内容后=%d 不同内容后=%d；命中=1", after1, after2, after3) +} diff --git a/go/internal/distill/score_cache_test.go b/go/internal/distill/score_cache_test.go new file mode 100644 index 0000000..e6af97e --- /dev/null +++ b/go/internal/distill/score_cache_test.go @@ -0,0 +1,135 @@ +package distill + +import ( + "os" + "path/filepath" + "testing" +) + +func newTestCache(t *testing.T, cap int) *scoreCache { + t.Helper() + p := filepath.Join(t.TempDir(), "c.json") + return &scoreCache{m: map[string]LLMResponse{}, path: p, cap: cap} +} + +// TestScoreCacheHit 同内容第二次必须命中且返回完全一致的结果（这才是"省调用"的前提）。 +func TestScoreCacheHit(t *testing.T) { + c := newTestCache(t, 100) + in := "牵挂: 待处理: 某条重复内容" + want := LLMResponse{IS: 0.4, SU: 0.5, PA: 0.3, VD: 0.6, RU: 0.5, Entities: []string{"x"}} + + if _, ok := c.get(in); ok { + t.Fatal("空缓存不应命中") + } + c.put(in, want) + + got, ok := c.get(in) + if !ok { + t.Fatal("同内容第二次应命中") + } + if got.IS != want.IS || got.VD != want.VD || len(got.Entities) != 1 { + t.Fatalf("命中返回的结果与写入不一致: %+v", got) + } + if h, m := c.stats(); h != 1 || m != 1 { + t.Fatalf("计数器不符：hits=%d miss=%d（期望 1/1）", h, m) + } +} + +// TestScoreCacheDifferentContent 不同内容必须各自独立（绝不能互相污染 —— 那会给出错误分数）。 +func TestScoreCacheDifferentContent(t *testing.T) { + c := newTestCache(t, 100) + c.put("A 内容", LLMResponse{IS: 0.1}) + c.put("B 内容", LLMResponse{IS: 0.9}) + if r, _ := c.get("A 内容"); r.IS != 0.1 { + t.Fatalf("A 被污染: %+v", r) + } + if r, _ := c.get("B 内容"); r.IS != 0.9 { + t.Fatalf("B 被污染: %+v", r) + } + if _, ok := c.get("C 内容"); ok { + t.Fatal("未写入过的内容不应命中") + } +} + +// TestScoreCacheEviction 超上限必须淘汰而不是无限增长（内存/落盘体积可控）。 +func TestScoreCacheEviction(t *testing.T) { + c := newTestCache(t, 3) + for _, s := range []string{"1", "2", "3", "4", "5"} { + c.put(s, LLMResponse{IS: 0.5}) + } + if len(c.m) != 3 { + t.Fatalf("淘汰后应恒为 3 条，实际 %d", len(c.m)) + } + if _, ok := c.get("1"); ok { + t.Fatal("最老的应被淘汰") + } + if _, ok := c.get("5"); !ok { + t.Fatal("最新的应还在") + } +} + +// TestScoreCachePersist 落盘往返：重启后仍要能命中（否则覆盖率永远上不去）。 +func TestScoreCachePersist(t *testing.T) { + dir := t.TempDir() + p := filepath.Join(dir, "c.json") + in := "重启也不该重算的内容" + want := LLMResponse{IS: 0.42, Entities: []string{"e1", "e2"}} + + c1 := &scoreCache{m: map[string]LLMResponse{}, path: p, cap: 100} + c1.put(in, want) + c1.save() + if _, err := os.Stat(p); err != nil { + t.Fatalf("落盘文件不存在: %v", err) + } + + c2 := &scoreCache{m: map[string]LLMResponse{}, path: p, cap: 100} + c2.load() + got, ok := c2.get(in) + if !ok { + t.Fatal("重启后（重新 load）应命中") + } + if got.IS != want.IS || len(got.Entities) != 2 { + t.Fatalf("恢复结果不一致: %+v", got) + } +} + +// TestScoreCacheCorruptFileStartsEmpty 缓存文件损坏只能"从空开始"，绝不能阻断蒸馏。 +func TestScoreCacheCorruptFileStartsEmpty(t *testing.T) { + p := filepath.Join(t.TempDir(), "c.json") + if err := os.WriteFile(p, []byte("{ 这不是合法 JSON"), 0o644); err != nil { + t.Fatal(err) + } + c := &scoreCache{m: map[string]LLMResponse{}, path: p, cap: 10} + c.load() // 不得 panic / 不得 fatal + if len(c.m) != 0 { + t.Fatalf("损坏文件应从空开始，实际 %d 条", len(c.m)) + } + c.put("x", LLMResponse{IS: 0.5}) + if _, ok := c.get("x"); !ok { + t.Fatal("损坏恢复后仍应可正常写入/命中") + } +} + +// TestScoreKeyStable 同内容哈希必须稳定（含中文字节安全）；不同内容不得碰撞。 +func TestScoreKeyStable(t *testing.T) { + a1, a2 := scoreKey("牵挂: 待拍板: 笔记→织忆蒸馏补齐"), scoreKey("牵挂: 待拍板: 笔记→织忆蒸馏补齐") + if a1 != a2 { + t.Fatal("同内容哈希不稳定") + } + if scoreKey("甲") == scoreKey("乙") { + t.Fatal("不同内容哈希碰撞") + } +} + +// TestScoreCacheNilSafe 未启用（getScoreCache 返回 nil）时不得 panic。 +func TestScoreCacheNilSafe(t *testing.T) { + var c *scoreCache + if _, ok := c.get("x"); ok { + t.Fatal("nil 缓存不应命中") + } + c.put("x", LLMResponse{}) // 不得 panic + if h, m := c.stats(); h != 0 || m != 0 { + t.Fatal("nil 缓存统计应为 0") + } + c.save() // 不得 panic +}