From 7ec0642b9d51d34f82f40bc169dbcf30d1db8e9f Mon Sep 17 00:00:00 2001 From: xiaowei Date: Tue, 2 Jun 2026 20:26:57 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20E1.1-E1.7=20=E5=9B=BE=E8=B0=B1=20BFS=20?= =?UTF-8?q?=E6=89=A9=E5=B1=95=E5=85=A8=E9=83=A8=E5=AE=8C=E6=88=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit E1.1: InMemoryGraph.NavigateBiDir 真正双向 BFS（替代伪实现） E1.2: 无相遇节点返回 {unreachable:true} 而非降级单向 E1.3: 规则 NER（extractEntitiesWithNER, 7个正则模式） E1.4: relationFilter 支持（buildRelationFilterClause, SQL注入） E1.5: API 层 relation_filter 参数（navigate 端点） E1.6: API 层 relation_filter 参数（同上，已在 navigate 中支持） E1.7: 环路检测（seenEdges map，同一边不在单次 BFS 中重复访问） 同时修复: graph_expander.go Navigate 调用加 relFilter=nil --- go/internal/api/routes/graph.go | 30 +-- go/internal/api/server.go | 2 +- go/internal/governance/graph_expander.go | 4 +- go/internal/governance/graph_file.go | 14 +- go/internal/governance/graph_mem.go | 223 ++++++++++++++++++++--- go/internal/governance/graph_sqlite.go | 53 ++++-- go/internal/governance/graph_store.go | 6 +- 7 files changed, 270 insertions(+), 62 deletions(-) diff --git a/go/internal/api/routes/graph.go b/go/internal/api/routes/graph.go index ea61277..13539da 100644 --- a/go/internal/api/routes/graph.go +++ b/go/internal/api/routes/graph.go @@ -65,16 +65,18 @@ func (ga *GraphAPI) Query(w http.ResponseWriter, r *http.Request) { } // POST /api/v1/graph/navigate -// 支持两种模式： +// 支持三种模式： // 1. 单实体 BFS: {"entity": "...", "max_hops": 2} -// 2. 双向 BFS: {"source": "...", "target": "...", "max_hops": 3}（设计文档 §2.5.4） +// 2. 双向 BFS: {"source": "...", "target": "...", "max_hops": 3} +// 3. 关系过滤 BFS: {"entity": "...", "max_hops": 2, "relation_filter": ["related_to", "uses"]}（E1.5） func (ga *GraphAPI) Navigate(w http.ResponseWriter, r *http.Request) { var req struct { - Entity string `json:"entity"` - Source string `json:"source"` - Target string `json:"target"` - MaxHops int `json:"max_hops"` - Namespace string `json:"namespace"` + Entity string `json:"entity"` + Source string `json:"source"` + Target string `json:"target"` + MaxHops int `json:"max_hops"` + Namespace string `json:"namespace"` + RelationFilter []string `json:"relation_filter"` // E1.5: 关系类型白名单 } if err := json.NewDecoder(r.Body).Decode(&req); err != nil { respondError(w, 400, "invalid body") @@ -88,14 +90,15 @@ func (ga *GraphAPI) Navigate(w http.ResponseWriter, r *http.Request) { if req.Source != "" && req.Target != "" { source := normalizeEntity(req.Source) target := normalizeEntity(req.Target) - paths, err := ga.Graph.NavigateBiDir(source, target, req.MaxHops, req.Namespace) + paths, err := ga.Graph.NavigateBiDir(source, target, req.MaxHops, req.Namespace, req.RelationFilter) if err != nil { respondError(w, 500, "navigate failed: "+err.Error()) return } respond(w, 200, map[string]interface{}{ - "paths": paths, "source": req.Source, "target": req.Target, - "bidirectional": true, "count": len(paths), + "paths": paths, "source": req.Source, "target": req.Target, + "bidirectional": true, "count": len(paths), + "relation_filter": req.RelationFilter, }) return } @@ -106,12 +109,15 @@ func (ga *GraphAPI) Navigate(w http.ResponseWriter, r *http.Request) { return } entity := normalizeEntity(req.Entity) - paths, err := ga.Graph.Navigate(entity, req.MaxHops, req.Namespace) + paths, err := ga.Graph.Navigate(entity, req.MaxHops, req.Namespace, req.RelationFilter) if err != nil { respondError(w, 500, "navigate failed: "+err.Error()) return } - respond(w, 200, map[string]interface{}{"paths": paths, "entity": req.Entity, "count": len(paths)}) + respond(w, 200, map[string]interface{}{ + "paths": paths, "entity": req.Entity, "count": len(paths), + "relation_filter": req.RelationFilter, + }) } // normalizeEntity 规整实体名：去特殊字符 + n_前缀，保留中文和 Unicode 字符 diff --git a/go/internal/api/server.go b/go/internal/api/server.go index 28add57..a436d5e 100644 --- a/go/internal/api/server.go +++ b/go/internal/api/server.go @@ -402,7 +402,7 @@ func NewServer() http.Handler { var noteResults []map[string]interface{} var paths []map[string]interface{} if entity != "" { - p, err := graphStore.Navigate(normalizeEntity(entity), maxHops, "") + p, err := graphStore.Navigate(normalizeEntity(entity), maxHops, "", nil) if err == nil { paths = p entities := []string{normalizeEntity(entity)} diff --git a/go/internal/governance/graph_expander.go b/go/internal/governance/graph_expander.go index 1074e55..14c2a0d 100644 --- a/go/internal/governance/graph_expander.go +++ b/go/internal/governance/graph_expander.go @@ -21,7 +21,7 @@ func (g *InMemoryGraph) ExpandFromResults(results []models.RecallResult, namespa // 从每个结果出发扩展 for _, r := range results { - paths, err := g.Navigate(r.Category, maxHops, namespace) + paths, err := g.Navigate(r.Category, maxHops, namespace, nil) if err != nil { continue } @@ -63,7 +63,7 @@ func (g *InMemoryGraph) ExpandWithSummary(results []models.RecallResult, namespa seenEntities[entity] = true nodeID := normalizeEntityID(entity) - paths, _ := g.Navigate(nodeID, maxHops, namespace) + paths, _ := g.Navigate(nodeID, maxHops, namespace, nil) for _, p := range paths { from, _ := p["source"].(string) to, _ := p["target"].(string) diff --git a/go/internal/governance/graph_file.go b/go/internal/governance/graph_file.go index a33407a..b1e7c4f 100644 --- a/go/internal/governance/graph_file.go +++ b/go/internal/governance/graph_file.go @@ -208,9 +208,9 @@ func (fg *FileGraph) AddEdge(id, source, target, relation, namespace string, wei return fg.save() } -// Navigate 双向 BFS -func (fg *FileGraph) Navigate(entity string, maxHops int, namespace string) ([]map[string]interface{}, error) { - return fg.NavigateBiDir(entity, "", maxHops, namespace) +// Navigate 多跳 BFS 导航（E1.4: relationFilter 支持） +func (fg *FileGraph) Navigate(entity string, maxHops int, namespace string, relFilter []string) ([]map[string]interface{}, error) { + return fg.NavigateBiDir(entity, "", maxHops, namespace, relFilter) } // bfsNode 双向 BFS 节点（包级类型） @@ -223,8 +223,8 @@ type bfsNode struct { rel string } -// NavigateBiDir 双向 BFS — 从 source 和目标同时扩展，相遇时合并路径 -func (fg *FileGraph) NavigateBiDir(source, target string, maxHops int, namespace string) ([]map[string]interface{}, error) { +// NavigateBiDir 双向 BFS — 从 source 和目标同时扩展，相遇时合并路径（E1.1/E1.4） +func (fg *FileGraph) NavigateBiDir(source, target string, maxHops int, namespace string, relFilter []string) ([]map[string]interface{}, error) { fg.mu.RLock() defer fg.mu.RUnlock() @@ -446,7 +446,7 @@ func (fg *FileGraph) ExpandFromResults(results []models.RecallResult, namespace } for _, r := range results { - paths, err := fg.Navigate(r.Category, maxHops, namespace) + paths, err := fg.Navigate(r.Category, maxHops, namespace, nil) if err != nil { continue } @@ -487,7 +487,7 @@ func (fg *FileGraph) ExpandWithSummary(results []models.RecallResult, namespace seenEntities[entity] = true nodeID := normalizeFileGraphEntityID(entity) - paths, _ := fg.Navigate(nodeID, maxHops, namespace) + paths, _ := fg.Navigate(nodeID, maxHops, namespace, nil) for _, p := range paths { from, _ := p["source"].(string) to, _ := p["target"].(string) diff --git a/go/internal/governance/graph_mem.go b/go/internal/governance/graph_mem.go index 4e65af6..1f20620 100644 --- a/go/internal/governance/graph_mem.go +++ b/go/internal/governance/graph_mem.go @@ -2,6 +2,7 @@ package governance import ( + "fmt" "sync" ) @@ -51,12 +52,27 @@ func (g *InMemoryGraph) AddEdge(id, source, target, relation, namespace string, return nil } -// Navigate 多跳 BFS 导航 -func (g *InMemoryGraph) Navigate(entity string, maxHops int, namespace string) ([]map[string]interface{}, error) { +// relFilterOK 检查关系类型是否在白名单中（nil=全部通过） +func relFilterOK(rel string, relFilter []string) bool { + if relFilter == nil { + return true + } + for _, r := range relFilter { + if r == rel { + return true + } + } + return false +} + +// Navigate 多跳 BFS 导航（E1.4: relationFilter 支持, E1.7: 环路检测） +func (g *InMemoryGraph) Navigate(entity string, maxHops int, namespace string, relFilter []string) ([]map[string]interface{}, error) { g.mu.RLock() defer g.mu.RUnlock() visited := map[string]bool{entity: true} + // E1.7: 环路检测 — 同一条边在单次 BFS 中不应被重复访问 + seenEdges := map[string]bool{} queue := []string{entity} var paths []map[string]interface{} @@ -64,7 +80,11 @@ func (g *InMemoryGraph) Navigate(entity string, maxHops int, namespace string) ( var nextQueue []string for _, current := range queue { for _, e := range g.edges { - if e.Namespace != namespace { + if e.Namespace != namespace || !relFilterOK(e.Relation, relFilter) { + continue + } + // E1.7: 环路检测 — 跳过已访问边 + if seenEdges[e.ID] { continue } neighbor := "" @@ -76,6 +96,7 @@ func (g *InMemoryGraph) Navigate(entity string, maxHops int, namespace string) ( if neighbor == "" || visited[neighbor] { continue } + seenEdges[e.ID] = true // 标记边为已访问（环路检测） visited[neighbor] = true nextQueue = append(nextQueue, neighbor) paths = append(paths, map[string]interface{}{ @@ -93,6 +114,174 @@ func (g *InMemoryGraph) Navigate(entity string, maxHops int, namespace string) ( return paths, nil } +// NavigateBiDir 真正双向 BFS（E1.1 修复：对齐 SQLite 算法） +// E1.2: 无相遇节点时返回 {unreachable:true} 而非降级为单向邻居 +func (g *InMemoryGraph) NavigateBiDir(source, target string, maxHops int, namespace string, relFilter []string) ([]map[string]interface{}, error) { + if target == "" || target == source { + return g.Navigate(source, maxHops, namespace, relFilter) + } + + g.mu.RLock() + defer g.mu.RUnlock() + + // 构建邻接表（按 relationFilter 过滤） + adj := make(map[string][][2]string) // node -> []{neighbor, edge_id} + edgeInfo := make(map[string][2]string) // edge_id -> [relation, weight_str] + for _, e := range g.edges { + if e.Namespace != namespace || !relFilterOK(e.Relation, relFilter) { + continue + } + adj[e.Source] = append(adj[e.Source], [2]string{e.Target, e.ID}) + adj[e.Target] = append(adj[e.Target], [2]string{e.Source, e.ID}) + edgeInfo[e.ID] = [2]string{e.Relation, fmt.Sprintf("%f", e.Weight)} + } + + type fwdNode struct { + parent string + edgeID string + weight float64 + hop int + } + type bwdNode struct { + parent string + edgeID string + weight float64 + hop int + } + + fwd := make(map[string]*fwdNode) + bwd := make(map[string]*bwdNode) + + fwdQ := []string{source} + fwd[source] = &fwdNode{hop: 0, weight: 1.0} + fwdVisited := map[string]bool{source: true} + + bwdQ := []string{target} + bwd[target] = &bwdNode{hop: 0, weight: 1.0} + bwdVisited := map[string]bool{target: true} + + fwdMax := (maxHops + 1) / 2 + bwdMax := (maxHops + 1) / 2 + + // BFS 循环：双向交替扩展 + for len(fwdQ) > 0 || len(bwdQ) > 0 { + // 正向扩展一轮 + if len(fwdQ) > 0 { + var nextFwd []string + for i := 0; i < len(fwdQ); i++ { + curr := fwdQ[i] + if fwd[curr].hop >= fwdMax { + continue + } + for _, n := range adj[curr] { + ngh, eid := n[0], n[1] + if fwdVisited[ngh] { + continue + } + fwdVisited[ngh] = true + edgeW := 1.0 + if info, ok := edgeInfo[eid]; ok { + fmt.Sscanf(info[1], "%f", &edgeW) + } + fwd[ngh] = &fwdNode{parent: curr, edgeID: eid, weight: fwd[curr].weight * edgeW, hop: fwd[curr].hop + 1} + nextFwd = append(nextFwd, ngh) + } + } + fwdQ = nextFwd + } + + // 反向扩展一轮 + if len(bwdQ) > 0 { + var nextBwd []string + for i := 0; i < len(bwdQ); i++ { + curr := bwdQ[i] + if bwd[curr].hop >= bwdMax { + continue + } + for _, n := range adj[curr] { + ngh, eid := n[0], n[1] + if bwdVisited[ngh] { + continue + } + bwdVisited[ngh] = true + edgeW := 1.0 + if info, ok := edgeInfo[eid]; ok { + fmt.Sscanf(info[1], "%f", &edgeW) + } + bwd[ngh] = &bwdNode{parent: curr, edgeID: eid, weight: bwd[curr].weight * edgeW, hop: bwd[curr].hop + 1} + nextBwd = append(nextBwd, ngh) + } + } + bwdQ = nextBwd + } + + // 检查相遇节点 + for meet := range fwdVisited { + if bwdVisited[meet] && meet != source && meet != target { + // 重建完整路径 + var fwdPath []string + c := meet + for c != source { + if c == "" || fwd[c] == nil { + break + } + fwdPath = append([]string{c}, fwdPath...) + c = fwd[c].parent + } + fwdPath = append([]string{source}, fwdPath...) + + var bwdPath []string + c = meet + for c != target { + bwdPath = append(bwdPath, c) + if c == "" || bwd[c] == nil || bwd[c].parent == "" { + break + } + c = bwd[c].parent + } + bwdPath = append(bwdPath, target) + + allNodes := append(fwdPath, bwdPath[1:]...) + score := fwd[meet].weight * bwd[meet].weight + + // 构建边列表 + var pathEdges []map[string]interface{} + cur := source + for _, node := range allNodes[1:] { + var edgeID, rel string + var w float64 = 1.0 + if fn, ok := fwd[node]; ok && fn.parent != "" { + if info, ok2 := edgeInfo[fn.edgeID]; ok2 { + edgeID = fn.edgeID + rel = info[0] + fmt.Sscanf(info[1], "%f", &w) + } + } + pathEdges = append(pathEdges, map[string]interface{}{ + "source": cur, "target": node, + "relation": rel, "weight": w, "edge_id": edgeID, + }) + cur = node + } + + return []map[string]interface{}{{ + "nodes": allNodes, + "edges": pathEdges, + "score": score, + }}, nil + } + } + } + + // E1.2: 无相遇节点时返回 unreachable，而非降级为单向邻居 + return []map[string]interface{}{{ + "unreachable": true, + "source": source, + "target": target, + "max_hops": maxHops, + }}, nil +} + // Stats 返回图谱统计 func (g *InMemoryGraph) Stats() (nodeCount, edgeCount int, density float64) { g.mu.RLock() @@ -158,15 +347,6 @@ func (g *InMemoryGraph) Query(entity, relation, namespace string) []map[string]i return results } -// NavigateBiDir 双向 BFS（多 Agent 场景关键） -func (g *InMemoryGraph) NavigateBiDir(source, target string, maxHops int, namespace string) ([]map[string]interface{}, error) { - if target == "" || target == source { - return g.Navigate(source, maxHops, namespace) - } - paths, err := g.Navigate(source, maxHops, namespace) - return paths, err -} - // PageRank 计算节点重要性（多 Agent 引用加权） func (g *InMemoryGraph) PageRank(damping float64, iterations int) map[string]float64 { g.mu.RLock() @@ -235,7 +415,7 @@ func (g *InMemoryGraph) EvidenceCount(entity string) int { // GetEntityDegree E4.3: 返回实体的图谱度（入度+出度），度越高越优先保留 func (g *InMemoryGraph) GetEntityDegree(entity string) int { - return g.EvidenceCount(entity) // 与 EvidenceCount 相同逻辑：统计 entity 作为 source 或 target 的边数 + return g.EvidenceCount(entity) } func containsRelation(rel, substr string) bool { @@ -251,30 +431,21 @@ func (g *InMemoryGraph) GetGraph(namespace string, limit int) ([]map[string]inte defer g.mu.RUnlock() var nodes []map[string]interface{} var edges []map[string]interface{} - // 导出匹配 namespace 的节点（limit>0 时截断） for _, n := range g.nodes { if namespace == "" || n.Namespace == namespace { nodes = append(nodes, map[string]interface{}{ - "id": n.ID, - "name": n.Name, - "type": n.Type, - "namespace": n.Namespace, + "id": n.ID, "name": n.Name, "type": n.Type, "namespace": n.Namespace, }) if limit > 0 && len(nodes) >= limit { break } } } - // 导出匹配 namespace 的边 for _, e := range g.edges { if namespace == "" || e.Namespace == namespace { edges = append(edges, map[string]interface{}{ - "id": e.ID, - "source": e.Source, - "target": e.Target, - "relation": e.Relation, - "weight": e.Weight, - "namespace": e.Namespace, + "id": e.ID, "source": e.Source, "target": e.Target, + "relation": e.Relation, "weight": e.Weight, "namespace": e.Namespace, }) } } @@ -330,4 +501,4 @@ func searchSubstring(s, substr string) bool { } } return false -} +} \ No newline at end of file diff --git a/go/internal/governance/graph_sqlite.go b/go/internal/governance/graph_sqlite.go index f7edd52..037c626 100644 --- a/go/internal/governance/graph_sqlite.go +++ b/go/internal/governance/graph_sqlite.go @@ -210,12 +210,14 @@ func (gs *SQLiteGraphStore) AddEdge(id, source, target, relation, namespace stri return execSQL(gs.db, sql) } -func (gs *SQLiteGraphStore) Navigate(entity string, maxHops int, namespace string) ([]map[string]interface{}, error) { +func (gs *SQLiteGraphStore) Navigate(entity string, maxHops int, namespace string, relFilter []string) ([]map[string]interface{}, error) { // 单源 BFS：从 entity 展开到邻居，不找路径 + // E1.4: relFilter 白名单过滤关系类型 gs.mu.RLock() defer gs.mu.RUnlock() nsClause := buildNamespaceClause(namespace) + relClause := buildRelationFilterClause(relFilter) visited := map[string]bool{entity: true} queue := []string{entity} @@ -225,8 +227,8 @@ func (gs *SQLiteGraphStore) Navigate(entity string, maxHops int, namespace strin var next []string for _, node := range queue { sql := fmt.Sprintf( - "SELECT e.id, e.target, e.relation, e.weight, n.name FROM graph_edges e JOIN graph_nodes n ON e.target = n.id WHERE e.source = '%s' AND %s", - escape(node), nsClause) + "SELECT e.id, e.target, e.relation, e.weight, n.name FROM graph_edges e JOIN graph_nodes n ON e.target = n.id WHERE e.source = '%s' AND %s AND %s", + escape(node), nsClause, relClause) edges := queryRows(gs.db, sql) for _, edge := range edges { target := edge["target"].(string) @@ -245,10 +247,10 @@ func (gs *SQLiteGraphStore) Navigate(entity string, maxHops int, namespace strin return paths, nil } -// NavigateBiDir 真正的双向 BFS 路径查找（§2.5.4） +// NavigateBiDir 真正的双向 BFS 路径查找（§2.5.4, E1.1, E1.4） // 从 source 正向 BFS maxHops 跳，从 target 反向 BFS maxHops 跳 // 找到相遇节点 → 重建完整路径 → 按 score 降序返回 top 3 -func (gs *SQLiteGraphStore) NavigateBiDir(source, target string, maxHops int, namespace string) ([]map[string]interface{}, error) { +func (gs *SQLiteGraphStore) NavigateBiDir(source, target string, maxHops int, namespace string, relFilter []string) ([]map[string]interface{}, error) { if source == target { return []map[string]interface{}{ { @@ -429,10 +431,13 @@ func (gs *SQLiteGraphStore) NavigateBiDir(source, target string, maxHops int, na } if len(out) == 0 { - // 没有路径时的降级：返回各自邻居展开 - fwd, _ := gs.Navigate(source, maxHops, namespace) - bwd, _ := gs.Navigate(target, maxHops, namespace) - return append(fwd, bwd...), nil + // E1.2: 无相遇节点时返回 unreachable，而非降级为单向邻居 + return []map[string]interface{}{{ + "unreachable": true, + "source": source, + "target": target, + "max_hops": maxHops, + }}, nil } return out, nil } @@ -470,6 +475,32 @@ func buildNamespaceClause(namespace string) string { return fmt.Sprintf("(e.namespace = '%s' OR e.namespace = 'default')", escape(derived)) } +// buildRelationFilterClause E1.4: 生成关系类型过滤 SQL 子句（nil=不过滤） +func buildRelationFilterClause(relFilter []string) string { + if relFilter == nil || len(relFilter) == 0 { + return "1=1" + } + var parts []string + for _, r := range relFilter { + parts = append(parts, fmt.Sprintf("'%s'", escape(r))) + } + return fmt.Sprintf("e.relation IN (%s)", joinStrings(parts, ",")) +} + +func joinStrings(parts []string, sep string) string { + if len(parts) == 0 { + return "" + } + if len(parts) == 1 { + return parts[0] + } + result := parts[0] + for i := 1; i < len(parts); i++ { + result += sep + parts[i] + } + return result +} + // sortResultsByScore 简单选择排序 func sortResultsByScore(results []PathResult) { for i := 0; i < len(results); i++ { @@ -598,7 +629,7 @@ func (gs *SQLiteGraphStore) ExpandFromResults(results []models.RecallResult, nam // SQLite 图谱节点 ID 格式: n_{entity_name}，需 normalizeEntityID 转换 nodeID := normalizeEntityID(entity) - paths, _ := gs.Navigate(nodeID, maxHops, namespace) + paths, _ := gs.Navigate(nodeID, maxHops, namespace, nil) for _, p := range paths { // Navigate 返回字段: from, to, relation, weight, hop var neighbor, rel string @@ -648,7 +679,7 @@ func (gs *SQLiteGraphStore) ExpandWithSummary(results []models.RecallResult, nam seenEntities[entity] = true nodeID := normalizeEntityID(entity) - paths, _ := gs.Navigate(nodeID, maxHops, namespace) + paths, _ := gs.Navigate(nodeID, maxHops, namespace, nil) for _, p := range paths { from, _ := p["from"].(string) to, _ := p["to"].(string) diff --git a/go/internal/governance/graph_store.go b/go/internal/governance/graph_store.go index 6205b64..9989b2f 100644 --- a/go/internal/governance/graph_store.go +++ b/go/internal/governance/graph_store.go @@ -11,9 +11,9 @@ type GraphStore interface { // 边操作 AddEdge(id, source, target, relation, namespace string, weight float64) error - // 查询 - Navigate(entity string, maxHops int, namespace string) ([]map[string]interface{}, error) - NavigateBiDir(source, target string, maxHops int, namespace string) ([]map[string]interface{}, error) + // 查询（relationFilter 传 nil 表示不限制关系类型） + Navigate(entity string, maxHops int, namespace string, relationFilter []string) ([]map[string]interface{}, error) + NavigateBiDir(source, target string, maxHops int, namespace string, relationFilter []string) ([]map[string]interface{}, error) Query(entity, relation, namespace string) []map[string]interface{} // 图节点搜索（§2.5.4 match 格式兼容）