From 3d054e75900238a378030b976c7a2696703415c2 Mon Sep 17 00:00:00 2001 From: xiaowei Date: Wed, 3 Jun 2026 12:44:31 +0800 Subject: [PATCH] =?UTF-8?q?=E8=87=AA=E9=AA=8C=E8=AF=81=E6=9C=BA=E5=88=B6?= =?UTF-8?q?=EF=BC=9Aconsolidate=20=E8=81=9A=E7=B1=BB=E4=B8=8B=E9=99=90=20+?= =?UTF-8?q?=20timestamp=20=E5=89=8D=E7=BD=AE=E6=A3=80=E6=9F=A5=20+=20?= =?UTF-8?q?=E5=90=AF=E5=8A=A8=E6=95=B0=E6=8D=AE=E7=9B=AE=E5=BD=95=E6=A3=80?= =?UTF-8?q?=E6=9F=A5=20+=20=E4=BF=AE=E5=A4=8D=E5=9B=BE=E8=B0=B1=E6=89=A9?= =?UTF-8?q?=E5=B1=95=E5=AD=97=E6=AE=B5=E5=90=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. consolidation_pipe.go: - 前置：抽样检查记忆时间戳（< 2024-01-01 视为可疑 epoch-0） - 后置：clusters <= 1 时记录 ERROR + patterns 标记异常 2. server.go: runStartupChecks() 启动时检测废弃路径 + socket 可达性 3. graph_expander.go: Navigate 返回字段从 "target"/"source" 修正为 "to"/"from" --- go/internal/api/routes/consolidation_pipe.go | 66 +++++++++++++++++++- go/internal/api/server.go | 40 ++++++++++++ go/internal/governance/graph_expander.go | 6 +- 3 files changed, 108 insertions(+), 4 deletions(-) diff --git a/go/internal/api/routes/consolidation_pipe.go b/go/internal/api/routes/consolidation_pipe.go index 4efecd5..4d6ecc1 100644 --- a/go/internal/api/routes/consolidation_pipe.go +++ b/go/internal/api/routes/consolidation_pipe.go @@ -50,7 +50,14 @@ func (cp *ConsolidationPipeline) Run() (*ConsolidationReport, error) { // mode: "cluster_only" | "full" | "prune_only" // full 模式触发 LLM 质量回溯 + 衰减校准（仅低频调度使用） func (cp *ConsolidationPipeline) RunWithMode(mode string) (*ConsolidationReport, error) { - // ─── 尝试 Rust sidecar ────────────────────────────── + // ─── 前置检查：timestamp 合理性（防止 epoch-0 数据污染聚类结果）──────── + if mode == "full" || mode == "cluster_only" { + if suspicious, total := cp.checkTimestampSanity(); suspicious > 0 { + log.Printf("[WARN] timestamp 检查: %d/%d 条记忆时间戳可疑（可能是 epoch-0），结果仅供参考", suspicious, total) + } + } + + // ─── 尝试 Rust sidecar ─────────────────────────────────────── var report *ConsolidationReport if cp.dataDir != "" && cp.sqlitePath != "" { // 用指定模式调用 Rust sidecar @@ -64,6 +71,13 @@ func (cp *ConsolidationPipeline) RunWithMode(mode string) (*ConsolidationReport, Patterns: []string{fmt.Sprintf("decay_rates=%v", rustReport.DecayRates)}, GraphPruned: 0, } + + // ─── 后置检查：聚类数量下限 ───────────────────────────── + if rustReport.Clusters <= 1 { + log.Printf("[ERROR] consolidate 生成 clusters=%d（疑似 epoch-0 数据或聚类失效），结果可能无效", rustReport.Clusters) + report.Patterns = append(report.Patterns, fmt.Sprintf("⚠️ clusters=%d 可能异常", rustReport.Clusters)) + } + if rustReport.Quality != nil { report.Patterns = append(report.Patterns, fmt.Sprintf("quality_score=%.2f low_info=%d hallucinations=%d", @@ -270,6 +284,56 @@ func (cp *ConsolidationPipeline) updateGraph() (int, error) { return before - after, nil } +// checkTimestampSanity 抽样检查记忆时间戳合理性（5% sampling，不阻塞） +// 返回 (可疑数量, 抽样总数)。epoch-0（< 2024-01-01）视为可疑。 +func (cp *ConsolidationPipeline) checkTimestampSanity() (suspicious, total int) { + const epochThreshold int64 = 1704067200 // 2024-01-01 00:00:00 UTC + + memories, err := cp.ldb.GetCandidatesForForgetting() + if err != nil || len(memories) == 0 { + return 0, 0 + } + + // 全量不超过 100 条 + sampleSize := len(memories) + if sampleSize > 100 { + sampleSize = 100 + } + step := len(memories) / sampleSize + if step < 1 { + step = 1 + } + + for i := 0; i < len(memories); i += step { + total++ + ts := memoryInt64Val(memories[i], "created_at") + if ts > 0 && ts < epochThreshold { + suspicious++ + } + } + return suspicious, total +} + +// memoryInt64Val 安全取 int64（处理 string/int/float） +func memoryInt64Val(m map[string]interface{}, key string) int64 { + v, ok := m[key] + if !ok { + return 0 + } + switch val := v.(type) { + case int64: + return val + case float64: + return int64(val) + case int: + return int64(val) + case string: + t, _ := time.Parse(time.RFC3339, val) + return t.Unix() + } + return 0 +} + // ─── 报告 ───────────────────────────────────────────── type ConsolidationReport struct { diff --git a/go/internal/api/server.go b/go/internal/api/server.go index a436d5e..6852c25 100644 --- a/go/internal/api/server.go +++ b/go/internal/api/server.go @@ -87,6 +87,10 @@ func NewServer() http.Handler { ldb = storage.NewMemLanceClient(emb) log.Printf("[zhiyid] 存储后端: 内存（零依赖）") } + + // 启动时数据目录一致性检查（防止路径混乱导致读取废弃数据） + runStartupChecks(backend) + // G7.2: 注册 LDB getter（供 skill crystallize 使用） routes.RegisterLDBGetter(func() storage.LanceDB { return ldb }) @@ -1109,3 +1113,39 @@ func AgentTypeDecayOrDefault(agentType string) float64 { } return 0.015 } + +// runStartupChecks 启动时数据目录一致性检查 +// 检测废弃路径（如 /home/muc/data）并警告，防止数据源混乱 +func runStartupChecks(backend string) { + canonicalDataDir := "/var/lib/memoryweave" + deprecatedPaths := []string{ + "/home/muc/data", + "/home/muc/.local/share/memoryweave", + } + + for _, dep := range deprecatedPaths { + if _, err := os.Stat(dep); err == nil { + log.Printf("[WARN] 检测到废弃数据目录 %s，仍有数据残留（当前 canonical: %s）", dep, canonicalDataDir) + } + } + + // LanceDB 后端：检查 Rust sidecar socket 是否可达 + if backend == "lancedb" { + sockPath := os.Getenv("LANCEDB_SOCKET") + if sockPath == "" { + sockPath = "/tmp/zhiyi-ipc.sock" + } + if _, err := os.Stat(sockPath); os.IsNotExist(err) { + log.Printf("[WARN] LanceDB socket 不存在 (%s)，Rust sidecar 可能未运行", sockPath) + } else { + log.Printf("[startup] LanceDB socket 就绪: %s", sockPath) + } + } + + // canonical 目录存在性检查 + if _, err := os.Stat(canonicalDataDir); os.IsNotExist(err) { + log.Printf("[WARN] canonical 数据目录不存在: %s（首次部署？）", canonicalDataDir) + } else { + log.Printf("[startup] 数据目录正常: %s", canonicalDataDir) + } +} diff --git a/go/internal/governance/graph_expander.go b/go/internal/governance/graph_expander.go index 14c2a0d..fc74b5e 100644 --- a/go/internal/governance/graph_expander.go +++ b/go/internal/governance/graph_expander.go @@ -26,10 +26,10 @@ func (g *InMemoryGraph) ExpandFromResults(results []models.RecallResult, namespa continue } for _, p := range paths { - target, _ := p["target"].(string) - source, _ := p["source"].(string) + to, _ := p["to"].(string) + from, _ := p["from"].(string) - for _, id := range []string{target, source} { + for _, id := range []string{to, from} { if id != "" && !seen[id] { seen[id] = true expanded = append(expanded, models.RecallResult{