diff --git a/go/internal/api/routes/core.go b/go/internal/api/routes/core.go index cc91d27..6f5d403 100644 --- a/go/internal/api/routes/core.go +++ b/go/internal/api/routes/core.go @@ -270,6 +270,14 @@ func (a *API) Recall(w http.ResponseWriter, r *http.Request) { // Record recall hit rate (has results = hit, empty = miss) selfoptimize.Dash.RecordRecall(len(results) > 0) + // Gap auto-close: recall 命中后自动关闭该 query 对应的 open gap + if len(results) > 0 { + gd := selfoptimize.GetGlobalGapDetector() + if gd != nil { + gd.RecordHit(req.Query) + gd.ClearMisses(req.Query) + } + } // VProp 自动记录:命中 → success,空结果 → failure decisionID := "recall_" + req.Query + "_" + req.Namespace if len(results) > 0 { diff --git a/go/internal/api/server.go b/go/internal/api/server.go index 6852c25..2a6fba6 100644 --- a/go/internal/api/server.go +++ b/go/internal/api/server.go @@ -10,6 +10,7 @@ import ( "log" "net/http" "os" + "runtime" "strconv" "strings" "time" @@ -62,34 +63,11 @@ func NewServer() http.Handler { emb := storage.NewEmbedder(os.Getenv("VLLM_ENDPOINT")) rerank := storage.NewReranker(os.Getenv("RERANK_ENDPOINT"), emb) - // 存储后端选择:LanceDB (Rust IPC) → SQLite(CGO)→ 内存 - var ldb storage.LanceDB - backend := os.Getenv("STORAGE_BACKEND") - switch backend { - case "lancedb": - sockPath := os.Getenv("LANCEDB_SOCKET") - if sockPath == "" { - sockPath = "/tmp/zhiyi-ipc.sock" - } - ldb = storage.NewRustLanceDBClient(sockPath, emb) - log.Printf("[zhiyid] 存储后端: LanceDB (Rust IPC) — %s", sockPath) - case "sqlite": - dbPath := os.Getenv("SQLITE_PATH") - sqliteDB, err := storage.NewSQLiteClient(dbPath) - if err != nil { - log.Printf("[zhiyid] SQLite 初始化失败 (%v),降级为内存存储", err) - ldb = storage.NewMemLanceClient(emb) - } else { - ldb = sqliteDB - log.Printf("[zhiyid] 存储后端: SQLite (CGO) — %s", dbPath) - } - default: - ldb = storage.NewMemLanceClient(emb) - log.Printf("[zhiyid] 存储后端: 内存(零依赖)") - } + // 存储后端 + ldb := initStorageBackend(os.Getenv("STORAGE_BACKEND"), emb) // 启动时数据目录一致性检查(防止路径混乱导致读取废弃数据) - runStartupChecks(backend) + runStartupChecks(os.Getenv("STORAGE_BACKEND")) // G7.2: 注册 LDB getter(供 skill crystallize 使用) routes.RegisterLDBGetter(func() storage.LanceDB { return ldb }) @@ -112,20 +90,8 @@ func NewServer() http.Handler { } }() - // 图谱:SQLite (graph_nodes/graph_edges) — 设计要求,非 FileGraph JSON - graphPath := os.Getenv("GRAPH_PATH") - if graphPath == "" { - graphPath = "/var/lib/memoryweave/graph.db" - } - var graphStore governance.GraphStore - gs, err := governance.NewSQLiteGraphStore(graphPath) - if err != nil { - log.Printf("[zhiyid] WARN: SQLite 图谱初始化失败 (%v),降级为 InMemoryGraph", err) - graphStore = governance.NewInMemoryGraph() - } else { - graphStore = gs - log.Printf("[zhiyid] 图谱后端: SQLiteGraphStore — %s", graphPath) - } + // 图谱 + graphStore := initGraphStore() graphUpdater := governance.NewAutoGraphUpdater(graphStore) graphAPI := routes.NewGraphAPI(graphStore) @@ -141,6 +107,7 @@ func NewServer() http.Handler { // ─── 缺口 gapDetector := selfoptimize.NewGapDetector(emb, ldb) gapDetector.EnableGapRedisPersistence() + selfoptimize.SetGlobalGapDetector(gapDetector) // 注册全局实例供 recall 回调使用 // 因果追踪持久化 routes.CascadeR.Tracker().EnableCausalRedisPersistence() @@ -856,6 +823,25 @@ func NewServer() http.Handler { } return err }) + // 注册 conflict_muchen 处理器:自动裁决低信任差异冲突 + selfoptimize.Flow.Register("conflict_muchen", func(task *selfoptimize.PipelineTask) error { + conflicts := conflictAPI.Detector.ListActive() + for _, c := range conflicts { + if c.Status != "pending" { + continue + } + // 按策略自动裁决:latest_wins / primary_wins / dismiss + resolution := conflictAPI.Detector.AutoResolve(c) + if resolution == "pending" { + // 低信任差异(两记忆 trust 差 < 0.2)→ dismiss + resolution = "dismiss" + } + if err := conflictAPI.Detector.Resolve(c.ID, resolution, ""); err == nil { + selfoptimize.Dash.RecordConflictResolved(true) + } + } + return nil + }) go selfoptimize.Flow.Start() selfoptimize.Executor.Start(selfoptimize.Flow) @@ -1037,9 +1023,15 @@ func NewServer() http.Handler { time.Sleep(15 * time.Second) // 同步 Dashboard → Prometheus gauges metrics.SyncFromDashboard(selfoptimize.Dash.Metrics()) - // 采集 SQLite 数据库文件大小 - if fi, err := os.Stat(graphPath); err == nil { - metrics.SQLiteDBSizeBytes.Set(float64(fi.Size())) + // 采集 SQLite 数据库文件大小(Linux only) + if runtime.GOOS != "windows" { + gPath := os.Getenv("GRAPH_PATH") + if gPath == "" { + gPath = "/var/lib/memoryweave/graph.db" + } + if fi, err := os.Stat(gPath); err == nil { + metrics.SQLiteDBSizeBytes.Set(float64(fi.Size())) + } } // 采集进程 RSS 内存 if data, err := os.ReadFile("/proc/self/status"); err == nil { diff --git a/go/internal/selfoptimize/selfoptimize.go b/go/internal/selfoptimize/selfoptimize.go index 557c12b..a147065 100644 --- a/go/internal/selfoptimize/selfoptimize.go +++ b/go/internal/selfoptimize/selfoptimize.go @@ -395,6 +395,33 @@ func (gd *GapDetector) Close(topic string) { } } +// RecordHit 记录一次召回命中,自动关闭该 topic 的 open gap +func (gd *GapDetector) RecordHit(topic string) { + gd.mu.Lock() + defer gd.mu.Unlock() + if g, ok := gd.gaps[topic]; ok && !g.Closed { + g.Closed = true + gd.persistGaps() + Dash.RecordGapClosed() + } +} + +// ClearMisses 清除某个 topic 的 miss 计数(recall 命中后调用) +func (gd *GapDetector) ClearMisses(topic string) { + gd.mu.Lock() + defer gd.mu.Unlock() + delete(gd.misses, topic) + gd.persistMisses() +} + +// ─── 全局 GapDetector 访问(供 routes 包调用)─────────────── + +var globalGapDetector *GapDetector + +func SetGlobalGapDetector(gd *GapDetector) { globalGapDetector = gd } + +func GetGlobalGapDetector() *GapDetector { return globalGapDetector } + // ─── 因果追踪 ──────────────────────────────────────────── type TraceEntry struct {