diff --git a/go/internal/api/routes/core.go b/go/internal/api/routes/core.go index 6f5d403..36c6daa 100644 --- a/go/internal/api/routes/core.go +++ b/go/internal/api/routes/core.go @@ -23,9 +23,10 @@ type API struct { Reranker *storage.Reranker Pipeline *storage.RecallPipeline ConflictDetector *governance.ConflictDetector + GraphStore governance.GraphStore } -func NewAPI(ldb storage.LanceDB, emb *storage.Embedder, rerank *storage.Reranker, cd *governance.ConflictDetector) *API { +func NewAPI(ldb storage.LanceDB, emb *storage.Embedder, rerank *storage.Reranker, cd *governance.ConflictDetector, gs governance.GraphStore) *API { pipeline := storage.NewRecallPipeline(emb, ldb, rerank) pipeline.SetPrefetchPusher(&WSPrefetchAdapter{}) return &API{ @@ -34,6 +35,7 @@ func NewAPI(ldb storage.LanceDB, emb *storage.Embedder, rerank *storage.Reranker Reranker: rerank, Pipeline: pipeline, ConflictDetector: cd, + GraphStore: gs, } } @@ -238,8 +240,15 @@ func (a *API) Recall(w http.ResponseWriter, r *http.Request) { results, err := a.Pipeline.Recall( req.Query, req.Namespace, req.Limit, req.Diversity) if err != nil { - respondError(w, 500, "recall failed: "+err.Error()) - return + // P0: 降级到 graph.db 关键词搜索 + fallbackResults := a.GraphStore.FallbackTextSearch(req.Query, req.Namespace, req.Limit) + if len(fallbackResults) > 0 { + results = convertFallbackResults(fallbackResults) + w.Header().Set("X-Fallback", "graph") + } else { + respondError(w, 500, "recall failed: "+err.Error()) + return + } } // E4.2: 跨 agent 知识共享 — recall 结果 < 3 时，补充搜索 "shared" namespace @@ -652,3 +661,70 @@ func minInt3(a, b int) int { } return b } + +// convertFallbackResults 将 FallbackTextSearch 结果（map）转换为 RecallResult 格式 +func convertFallbackResults(rows []map[string]interface{}) []models.RecallResult { + results := make([]models.RecallResult, 0, len(rows)) + for _, r := range rows { + score := 0.5 + if pr, ok := r["pagerank"].(float64); ok { + score = pr + } + id := "" + if v, ok := r["id"]; ok { + id = fmt.Sprintf("%v", v) + } + name := "" + if v, ok := r["name"]; ok { + name = fmt.Sprintf("%v", v) + } + relation := "" + if v, ok := r["relation"]; ok { + relation = fmt.Sprintf("%v", v) + } + source := "" + if v, ok := r["source"]; ok { + source = fmt.Sprintf("%v", v) + } + target := "" + if v, ok := r["target"]; ok { + target = fmt.Sprintf("%v", v) + } + // Build readable content from graph edge info + content := fmt.Sprintf("[graph] %s --[%s]--> %s", source, relation, target) + if name != "" { + content = fmt.Sprintf("[graph] %s: %s --[%s]--> %s", name, source, relation, target) + } + results = append(results, models.RecallResult{ + ID: id, + Content: content, + Category: "graph_fallback", + Score: score, + }) + } + return results +} + +// POST /api/v1/graph/edge/feedback — P2: 边反馈 +func (a *API) EdgeFeedback(w http.ResponseWriter, r *http.Request) { + var req struct { + EdgeID string `json:"edge_id"` + Helpful bool `json:"helpful"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + respondError(w, 400, "invalid body") + return + } + if req.EdgeID == "" { + respondError(w, 400, "edge_id required") + return + } + err := a.GraphStore.AddEdgeFeedback(req.EdgeID, req.Helpful) + if err != nil { + respondError(w, 500, "feedback update failed: "+err.Error()) + return + } + // 触发信任评分更新 + _ = a.GraphStore.UpdateEdgeTrustScores() + respond(w, 200, map[string]string{"status": "ok"}) +} diff --git a/go/internal/api/server.go b/go/internal/api/server.go index 62b788f..504d880 100644 --- a/go/internal/api/server.go +++ b/go/internal/api/server.go @@ -59,7 +59,14 @@ func NewServer() http.Handler { // 冲突检测器（E4.1: Commit 时矛盾检测依赖此实例） conflictDetector := governance.NewConflictDetector() - api := routes.NewAPI(ldb, emb, rerank, conflictDetector) + + // 图谱（带 Navigate 结果缓存：TTL 5min，容量 500，热点实体缓存命中加速） + graphStore, cachedGraphStoreRef := storage.NewCachedGraphStore(initGraphStore()) + + api := routes.NewAPI(ldb, emb, rerank, conflictDetector, graphStore) + + // G1: Recall 管线挂图谱扩展 + api.Pipeline.SetGraphExpander(graphStore) // 启动时初始化 Prometheus 指标 go func() { @@ -76,13 +83,10 @@ func NewServer() http.Handler { }() // 图谱（带 Navigate 结果缓存：TTL 5min，容量 500，热点实体缓存命中加速） - graphStore, cachedGraphStoreRef := storage.NewCachedGraphStore(initGraphStore()) + // 已在上面初始化 graphUpdater := governance.NewAutoGraphUpdater(graphStore) graphAPI := routes.NewGraphAPI(graphStore) - // G1: Recall 管线挂图谱扩展 - api.Pipeline.SetGraphExpander(graphStore) - // G4: 自动蒸馏 → 注入图谱更新器 routes.SetGraphUpdater(graphUpdater) @@ -315,6 +319,8 @@ func NewServer() http.Handler { mux.HandleFunc("/api/v1/graph/stats", graphAPI.Stats) mux.HandleFunc("/api/v1/graph/query", graphAPI.Query) mux.HandleFunc("/api/v1/graph/navigate", graphAPI.Navigate) + // P2: 边反馈 + mux.HandleFunc("/api/v1/graph/edge/feedback", api.EdgeFeedback) // 新增：添加关系边（POST JSON body: {"from":"实体A","to":"实体B","relation":"关系类型","namespace":""}） mux.HandleFunc("/api/v1/graph/edge", func(w http.ResponseWriter, r *http.Request) { if r.Method != "POST" { diff --git a/go/internal/governance/graph_file.go b/go/internal/governance/graph_file.go index 8198e16..e0e5700 100644 --- a/go/internal/governance/graph_file.go +++ b/go/internal/governance/graph_file.go @@ -442,6 +442,16 @@ func (fg *FileGraph) CleanupNoiseNodes(dryRun bool) (int, []string, error) { return 0, nil, nil } +// P0: FallbackTextSearch FileGraph stub（已迁移到 SQLite） +func (fg *FileGraph) FallbackTextSearch(query, namespace string, limit int) []map[string]interface{} { + return nil +} + +// P2: 信任评分 stub（FileGraph 不持久化信任数据） +func (fg *FileGraph) AddEdgeFeedback(edgeID string, helpful bool) error { return nil } +func (fg *FileGraph) IncrementEdgeRetrieval(edgeID string) error { return nil } +func (fg *FileGraph) UpdateEdgeTrustScores() error { return nil } + // ─── 图谱扩展 ──────────────────────────────────────────── func (fg *FileGraph) ExpandFromResults(results []models.RecallResult, namespace string, maxHops int) []models.RecallResult { diff --git a/go/internal/governance/graph_file_windows.go b/go/internal/governance/graph_file_windows.go index 51ee160..1083c3e 100644 --- a/go/internal/governance/graph_file_windows.go +++ b/go/internal/governance/graph_file_windows.go @@ -76,4 +76,13 @@ func (fg *FileGraph) lockFile(fd *os.File, exclusive bool) error { // unlockFile 暂不实现（no-op） func (fg *FileGraph) unlockFile(fd *os.File) { -} \ No newline at end of file +} + +// ─── P0/P2 Stubs（Windows FileGraph）──────────────────── + +func (fg *FileGraph) FallbackTextSearch(query, namespace string, limit int) []map[string]interface{} { + return nil +} +func (fg *FileGraph) AddEdgeFeedback(edgeID string, helpful bool) error { return nil } +func (fg *FileGraph) IncrementEdgeRetrieval(edgeID string) error { return nil } +func (fg *FileGraph) UpdateEdgeTrustScores() error { return nil } \ No newline at end of file diff --git a/go/internal/governance/graph_mem.go b/go/internal/governance/graph_mem.go index fdb4083..f6abdae 100644 --- a/go/internal/governance/graph_mem.go +++ b/go/internal/governance/graph_mem.go @@ -501,6 +501,17 @@ func (g *InMemoryGraph) ListNodes(namespace string) []map[string]interface{} { return g.ListNodesByType("", namespace) } +// P0: FallbackTextSearch 内存版 stub +func (g *InMemoryGraph) FallbackTextSearch(query, namespace string, limit int) []map[string]interface{} { + // InMemoryGraph 不支持 SQL LIKE，降级到 SearchNodes + return g.SearchNodes(query, namespace) +} + +// P2: 信任评分 stub（InMemoryGraph 不持久化） +func (g *InMemoryGraph) AddEdgeFeedback(edgeID string, helpful bool) error { return nil } +func (g *InMemoryGraph) IncrementEdgeRetrieval(edgeID string) error { return nil } +func (g *InMemoryGraph) UpdateEdgeTrustScores() error { return nil } + func searchSubstring(s, substr string) bool { for i := 0; i <= len(s)-len(substr); i++ { if s[i:i+len(substr)] == substr { diff --git a/go/internal/governance/graph_sqlite.go b/go/internal/governance/graph_sqlite.go index ca5e148..f667710 100644 --- a/go/internal/governance/graph_sqlite.go +++ b/go/internal/governance/graph_sqlite.go @@ -136,6 +136,11 @@ func (gs *SQLiteGraphStore) migrate() error { // 修复孤儿边：自动补充缺失的节点 gs.repairOrphanEdges() + // P2: Trust scoring columns for graph_edges + gs.migrateAddColumn("graph_edges", "trust_score", "REAL DEFAULT 0.5") + gs.migrateAddColumn("graph_edges", "retrieval_count", "INTEGER DEFAULT 0") + gs.migrateAddColumn("graph_edges", "helpful_count", "INTEGER DEFAULT 0") + return nil } @@ -981,6 +986,50 @@ func (gs *SQLiteGraphStore) GetEntityDegree(entity string) int { return gs.EvidenceCount(entity) } +// P0: FallbackTextSearch — 关键词降级搜索（向量搜索不可用时使用） +func (gs *SQLiteGraphStore) FallbackTextSearch(query, namespace string, limit int) []map[string]interface{} { + gs.mu.RLock() + defer gs.mu.RUnlock() + nsClause := "1=1" + if namespace != "" { + nsClause = fmt.Sprintf("e.namespace = '%s'", escape(namespace)) + } + // 模糊匹配 node name + edge relation，按 pagerank 排序 + sql := fmt.Sprintf( + `SELECT DISTINCT e.id, e.source, e.target, e.relation, e.weight, n.name, n.pagerank + FROM graph_edges e + JOIN graph_nodes n ON e.source = n.id + WHERE (n.name LIKE '%%%s%%' OR e.relation LIKE '%%%s%%') AND %s + ORDER BY n.pagerank DESC + LIMIT %d`, + escape(query), escape(query), nsClause, limit) + return queryRows(gs.db, sql) +} + +// P2: AddEdgeFeedback 记录边反馈（helpful=true 增加 helpful_count，否则增加 retrieval_count） +func (gs *SQLiteGraphStore) AddEdgeFeedback(edgeID string, helpful bool) error { + gs.mu.Lock() + defer gs.mu.Unlock() + if helpful { + return execSQL(gs.db, fmt.Sprintf("UPDATE graph_edges SET helpful_count = helpful_count + 1 WHERE id = '%s'", escape(edgeID))) + } + return execSQL(gs.db, fmt.Sprintf("UPDATE graph_edges SET retrieval_count = retrieval_count + 1 WHERE id = '%s'", escape(edgeID))) +} + +// IncrementEdgeRetrieval 递增边的检索计数 +func (gs *SQLiteGraphStore) IncrementEdgeRetrieval(edgeID string) error { + gs.mu.Lock() + defer gs.mu.Unlock() + return execSQL(gs.db, fmt.Sprintf("UPDATE graph_edges SET retrieval_count = retrieval_count + 1 WHERE id = '%s'", escape(edgeID))) +} + +// UpdateEdgeTrustScores 批量更新边的信任评分（trust_score = helpful_count / retrieval_count） +func (gs *SQLiteGraphStore) UpdateEdgeTrustScores() error { + gs.mu.Lock() + defer gs.mu.Unlock() + return execSQL(gs.db, `UPDATE graph_edges SET trust_score = CASE WHEN retrieval_count > 0 THEN CAST(helpful_count AS REAL) / retrieval_count ELSE 0.5 END`) +} + // ─── CGO 工具 ────────────────────────────────────────── // UpdatePageRanks 批量更新节点的 pagerank 值（§2.5.5） diff --git a/go/internal/governance/graph_store.go b/go/internal/governance/graph_store.go index 3427f46..fd9e8e9 100644 --- a/go/internal/governance/graph_store.go +++ b/go/internal/governance/graph_store.go @@ -44,4 +44,12 @@ type GraphStore interface { // 清理图谱脏数据：删除名称含编码噪音的节点（如 fts=、括号不匹配等） // 返回被删除的节点数和节点 ID 列表 CleanupNoiseNodes(dryRun bool) (int, []string, error) + + // P0: 关键词文本搜索降级（当向量搜索不可用时） + FallbackTextSearch(query, namespace string, limit int) []map[string]interface{} + + // P2: 信任评分 + AddEdgeFeedback(edgeID string, helpful bool) error + IncrementEdgeRetrieval(edgeID string) error + UpdateEdgeTrustScores() error } diff --git a/go/internal/storage/graph_cache.go b/go/internal/storage/graph_cache.go index 0a71879..d8d1e36 100644 --- a/go/internal/storage/graph_cache.go +++ b/go/internal/storage/graph_cache.go @@ -161,6 +161,22 @@ func (c *cachedGraphStore) CleanupNoiseNodes(dryRun bool) (int, []string, error) return c.inner.CleanupNoiseNodes(dryRun) } +// P0: FallbackTextSearch 透传到内层 +func (c *cachedGraphStore) FallbackTextSearch(query, namespace string, limit int) []map[string]interface{} { + return c.inner.FallbackTextSearch(query, namespace, limit) +} + +// P2: 信任评分透传 +func (c *cachedGraphStore) AddEdgeFeedback(edgeID string, helpful bool) error { + return c.inner.AddEdgeFeedback(edgeID, helpful) +} +func (c *cachedGraphStore) IncrementEdgeRetrieval(edgeID string) error { + return c.inner.IncrementEdgeRetrieval(edgeID) +} +func (c *cachedGraphStore) UpdateEdgeTrustScores() error { + return c.inner.UpdateEdgeTrustScores() +} + // ── 缓存内部方法 ── func (c *cachedGraphStore) navigateKey(entity string, maxHops int, relFilter []string) string { diff --git a/go/zhiyid-new b/go/zhiyid-new index 09a091c..1c9d786 100755 Binary files a/go/zhiyid-new and b/go/zhiyid-new differ