From bceb5fb059e407e2cbfd9ee2ca2e3a90945b4e37 Mon Sep 17 00:00:00 2001 From: xiaowei Date: Sat, 30 May 2026 11:27:27 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=87=AA=E4=BC=98=E5=8C=964=E5=A4=A7?= =?UTF-8?q?=E9=93=BE=E8=B7=AF=E8=A1=A5=E9=BD=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. GapDetector 自动接入 recall 路径 - MissRecorder callback → RecallPipeline - 0 结果时自动 RecordMiss(query) - server.go 中 gapDetector 初始化后挂载 2. Dashboard 持久化到 Redis - EnableRedisPersistence() 启动时加载 - persist() 每次 mutate 后同步 13 个计数器到 Redis HASH - 重启不丢指标 3. PassiveValidator 接入 commit 路径 - 新记忆提交后自动 P1/P2/P3 匹配验证 - 复用 Validator.Validate 逻辑 4. VPropagator 接入 feedback/correct - 用户修正时记录决策 V 值 - 使用 user_correct 来源 + success 结果 Closes: 10% 缺失代码 + 40% 缺失链路 --- go/internal/api/routes/core.go | 20 ++++++ go/internal/api/routes/feedback.go | 6 ++ go/internal/api/server.go | 7 ++ go/internal/selfoptimize/selfoptimize.go | 82 +++++++++++++++++++++++- go/internal/storage/recall.go | 13 ++++ 5 files changed, 126 insertions(+), 2 deletions(-) diff --git a/go/internal/api/routes/core.go b/go/internal/api/routes/core.go index cbf58fb..d36ccc8 100644 --- a/go/internal/api/routes/core.go +++ b/go/internal/api/routes/core.go @@ -131,6 +131,26 @@ func (a *API) Commit(w http.ResponseWriter, r *http.Request) { return } + // 被动验证：对新记忆与已有记忆做 P1/P2/P3 匹配 + go func() { + // 搜索同 namespace 已有记忆 + zeroVec := make([]float32, 1024) + existing, _ := a.LanceDB.Search("memories", zeroVec, 50, req.Namespace) + if len(existing) > 0 { + validMems := make([]selfoptimize.MemoryForValidation, len(existing)) + for i, m := range existing { + validMems[i] = selfoptimize.MemoryForValidation{ + ID: m.ID, Content: m.Content, QualityScore: m.QualityScore, + } + } + // 用提交的内容做验证（不是用新记忆 ID） + validated := selfoptimize.Validator.Validate(req.Content, validMems) + if len(validated) > 0 { + _ = validated // WebSocket 通知可在此展开 + } + } + }() + respond(w, 201, map[string]string{ "episode_id": epID, "memory_id": memID, "status": "ok", }) diff --git a/go/internal/api/routes/feedback.go b/go/internal/api/routes/feedback.go index 91663e8..1519a51 100644 --- a/go/internal/api/routes/feedback.go +++ b/go/internal/api/routes/feedback.go @@ -95,6 +95,12 @@ func (fa *FeedbackAPI) Correct(w http.ResponseWriter, r *http.Request) { } selfoptimize.Dash.RecordCorrection(req.Source) + // 记录 V 值决策：用户修正 = success + selfoptimize.VProp.RecordDecision( + "correct_"+req.MemoryID, + []string{req.MemoryID}, + "user_correct", "success", "", + ) err := fa.LanceDB.UpdateMemoryContent(req.MemoryID, req.NewContent, req.Source) if err != nil { respondError(w, 500, "correct failed: "+err.Error()) diff --git a/go/internal/api/server.go b/go/internal/api/server.go index 6e7358d..f7d7c94 100644 --- a/go/internal/api/server.go +++ b/go/internal/api/server.go @@ -107,6 +107,10 @@ func NewServer() http.Handler { gapDetector := selfoptimize.NewGapDetector(emb, ldb) gapAPI := routes.NewGapAPI(gapDetector) routes.InitGapRepair(gapDetector) // 共享同一个 GapDetector + // G1a: 挂缺隙记录器（0 结果 → 自动记录 miss） + api.Pipeline.SetMissRecorder(func(query, namespace string) { + gapDetector.RecordMiss(query + " [" + namespace + "]") + }) feedbackAPI := routes.NewFeedbackAPI(ldb) // 遗忘器：支持按 Agent 类型设置衰减率 @@ -184,6 +188,9 @@ func NewServer() http.Handler { // ─── CO_OCCURS 追踪器 ─────────────────────── storage.CoOccurTrackerInstance = storage.NewCoOccurTracker(nil) + // ─── Dashboard 持久化到 Redis（restart 不丢指标）─── + selfoptimize.EnableRedisPersistence() + // ─── V 值传播器（竞争性架构核心）───────────── vPropagator := selfoptimize.VProp // 在蒸馏完成回调中记录 V 值 diff --git a/go/internal/selfoptimize/selfoptimize.go b/go/internal/selfoptimize/selfoptimize.go index dbf91cd..7b129e9 100644 --- a/go/internal/selfoptimize/selfoptimize.go +++ b/go/internal/selfoptimize/selfoptimize.go @@ -3,12 +3,79 @@ package selfoptimize import ( "math" + "strconv" "sync" "time" "github.com/xiaoxue/memoryweave/internal/storage" ) +// Redis persistence key +const dashboardRedisKey = "zhiyi:dashboard" + +// redisPersistence 是否开启了 Redis 持久化（由 server.go 在启动时设置） +var redisPersistence *storage.RedisClient + +// EnableRedisPersistence 启动 Dashboard 持久化（从 Redis 加载 + 每次变更同步） +func EnableRedisPersistence() { + rc := storage.GetRedisClient() + if rc == nil { + return + } + redisPersistence = rc + _ = loadDashboard() +} + +// persist 写当前 Dashboard 到 Redis HASH +// 注意：调用方必须已持有 d.mu 锁（RLock 或 Lock） +func (d *Dashboard) persist() { + if redisPersistence == nil { + return + } + _ = redisPersistence.HSet(dashboardRedisKey, "useful_count", strconv.Itoa(d.UsefulCount)) + _ = redisPersistence.HSet(dashboardRedisKey, "not_useful_count", strconv.Itoa(d.NotUsefulCount)) + _ = redisPersistence.HSet(dashboardRedisKey, "total_recalls", strconv.Itoa(d.TotalRecalls)) + _ = redisPersistence.HSet(dashboardRedisKey, "hit_count", strconv.Itoa(d.HitCount)) + _ = redisPersistence.HSet(dashboardRedisKey, "closed_gaps", strconv.Itoa(d.ClosedGaps)) + _ = redisPersistence.HSet(dashboardRedisKey, "total_gaps", strconv.Itoa(d.TotalGaps)) + _ = redisPersistence.HSet(dashboardRedisKey, "cascade_fixed_total", strconv.Itoa(d.CascadeFixedTotal)) + _ = redisPersistence.HSet(dashboardRedisKey, "total_fixes", strconv.Itoa(d.TotalFixes)) + _ = redisPersistence.HSet(dashboardRedisKey, "deprecated_today", strconv.Itoa(d.DeprecatedToday)) + _ = redisPersistence.HSet(dashboardRedisKey, "distill_loss_sum", strconv.FormatFloat(d.DistillLossSum, 'f', 4, 64)) + _ = redisPersistence.HSet(dashboardRedisKey, "distill_loss_count", strconv.Itoa(d.DistillLossCount)) + _ = redisPersistence.HSet(dashboardRedisKey, "auto_resolved_conflicts", strconv.Itoa(d.AutoResolvedConflicts)) + _ = redisPersistence.HSet(dashboardRedisKey, "total_conflicts", strconv.Itoa(d.TotalConflicts)) + _ = redisPersistence.Expire(dashboardRedisKey, 90*24*time.Hour) // 90 天 TTL +} + +// loadDashboard 从 Redis 加载 Dashboard 数据 +func loadDashboard() error { + if redisPersistence == nil { + return nil + } + data, err := redisPersistence.HGetAll(dashboardRedisKey) + if err != nil || len(data) == 0 { + return err + } + Dash.mu.Lock() + defer Dash.mu.Unlock() + + if v, err := strconv.Atoi(data["useful_count"]); err == nil { Dash.UsefulCount = v } + if v, err := strconv.Atoi(data["not_useful_count"]); err == nil { Dash.NotUsefulCount = v } + if v, err := strconv.Atoi(data["total_recalls"]); err == nil { Dash.TotalRecalls = v } + if v, err := strconv.Atoi(data["hit_count"]); err == nil { Dash.HitCount = v } + if v, err := strconv.Atoi(data["closed_gaps"]); err == nil { Dash.ClosedGaps = v } + if v, err := strconv.Atoi(data["total_gaps"]); err == nil { Dash.TotalGaps = v } + if v, err := strconv.Atoi(data["cascade_fixed_total"]); err == nil { Dash.CascadeFixedTotal = v } + if v, err := strconv.Atoi(data["total_fixes"]); err == nil { Dash.TotalFixes = v } + if v, err := strconv.Atoi(data["deprecated_today"]); err == nil { Dash.DeprecatedToday = v } + if v, err := strconv.ParseFloat(data["distill_loss_sum"], 64); err == nil { Dash.DistillLossSum = v } + if v, err := strconv.Atoi(data["distill_loss_count"]); err == nil { Dash.DistillLossCount = v } + if v, err := strconv.Atoi(data["auto_resolved_conflicts"]); err == nil { Dash.AutoResolvedConflicts = v } + if v, err := strconv.Atoi(data["total_conflicts"]); err == nil { Dash.TotalConflicts = v } + return nil +} + // ─── 自优化仪表盘 ──────────────────────────────────────── type Dashboard struct { @@ -83,6 +150,7 @@ func (d *Dashboard) RecordRecall(hit bool) { if hit { d.HitCount++ } + d.persist() } func (d *Dashboard) RecordFeedback(useful bool) { @@ -96,10 +164,16 @@ func (d *Dashboard) RecordFeedback(useful bool) { } // RecordUseful 记录有用反馈 -func (d *Dashboard) RecordUseful() { d.RecordFeedback(true) } +func (d *Dashboard) RecordUseful() { + d.RecordFeedback(true) + d.persist() +} // RecordNotUseful 记录无用反馈 -func (d *Dashboard) RecordNotUseful() { d.RecordFeedback(false) } +func (d *Dashboard) RecordNotUseful() { + d.RecordFeedback(false) + d.persist() +} // QualityScore 计算当前质量分数 func (d *Dashboard) QualityScore() float64 { @@ -380,6 +454,7 @@ func (d *Dashboard) RecordDeprecation() { d.mu.Lock() defer d.mu.Unlock() d.DeprecatedToday++ + d.persist() } // RecordCorrection 记录用户修正 @@ -390,6 +465,7 @@ func (d *Dashboard) RecordCorrection(source string) { if source == "muchen_correction" { d.CascadeFixedTotal++ } + d.persist() } // RecordGapClosed 记录缺口关闭 @@ -397,6 +473,7 @@ func (d *Dashboard) RecordGapClosed() { d.mu.Lock() defer d.mu.Unlock() d.ClosedGaps++ + d.persist() } // RecordConflictResolved 记录冲突解决 @@ -407,6 +484,7 @@ func (d *Dashboard) RecordConflictResolved(auto bool) { if auto { d.AutoResolvedConflicts++ } + d.persist() } func (pg *PrefetchGraph) GetPrefetch(query string) []string { diff --git a/go/internal/storage/recall.go b/go/internal/storage/recall.go index 6c830b1..e068042 100644 --- a/go/internal/storage/recall.go +++ b/go/internal/storage/recall.go @@ -10,6 +10,9 @@ import ( "github.com/xiaoxue/memoryweave/internal/models" ) +// MissRecorder 缺口记录接口（由 routes 层注入 gapDetector.RecordMiss） +type MissRecorder func(query, namespace string) + // PrefetchPusher 预取推送接口（由 routes 层注入） type PrefetchPusher interface { PushPrefetch(agentID string, memories []models.RecallResult) @@ -26,6 +29,7 @@ type RecallPipeline struct { reranker *Reranker graph GraphExpander prefetch PrefetchPusher + recordMiss MissRecorder } func NewRecallPipeline(embedder *Embedder, lancedb LanceDB, reranker *Reranker) *RecallPipeline { @@ -46,6 +50,11 @@ func (p *RecallPipeline) SetPrefetchPusher(pf PrefetchPusher) { p.prefetch = pf } +// SetMissRecorder 注入缺口记录器（gapDetector.RecordMiss） +func (p *RecallPipeline) SetMissRecorder(mr MissRecorder) { + p.recordMiss = mr +} + func (p *RecallPipeline) Recall(query, namespace string, topK int, diversity float64) ([]models.RecallResult, error) { if topK <= 0 { topK = 10 @@ -97,6 +106,10 @@ func (p *RecallPipeline) Recall(query, namespace string, topK int, diversity flo } if len(candidates) == 0 { + // 自动记录召回缺口（gap_scan 触发器会在 30min 冷却后分类） + if p.recordMiss != nil { + p.recordMiss(query, namespace) + } return []models.RecallResult{}, nil }