diff --git a/go/internal/api/server.go b/go/internal/api/server.go index 3688b9b..a3bbfbe 100644 --- a/go/internal/api/server.go +++ b/go/internal/api/server.go @@ -164,7 +164,7 @@ func NewServer() http.Handler { storage.CoOccurTrackerInstance = storage.NewCoOccurTracker(nil) // ─── V 值传播器（竞争性架构核心）───────────── - vPropagator := &selfoptimize.VPropagator{} + vPropagator := selfoptimize.VProp // 在蒸馏完成回调中记录 V 值 originalOnComplete := distill.OnDistillComplete distill.OnDistillComplete = func(input distill.DistillInput, result distill.DistillResult) { @@ -269,7 +269,14 @@ func NewServer() http.Handler { // 图谱可视化导出 mux.HandleFunc("/api/v1/graph/export", func(w http.ResponseWriter, r *http.Request) { ns := r.URL.Query().Get("namespace") - nodes, edges := graphStore.GetGraph(ns) + limitStr := r.URL.Query().Get("limit") + limit := 0 + if limitStr != "" { + if l, err := strconv.Atoi(limitStr); err == nil && l > 0 { + limit = l + } + } + nodes, edges := graphStore.GetGraph(ns, limit) respondJSON(w, 200, map[string]interface{}{ "nodes": nodes, "edges": edges, @@ -781,6 +788,20 @@ func NewServer() http.Handler { }() log.Println("[zhiyid] 多 Agent 架构 — Redis + FileGraph + EventBus + vLLM — 已启动") + + // Catch-all: return JSON for unknown paths + mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/" || r.URL.Path == "" { + respondJSON(w, 200, map[string]interface{}{ + "service": "zhiyid", "status": "ok", "version": "0.1.0", + }) + return + } + respondJSON(w, 404, map[string]interface{}{ + "error": "endpoint not found: " + r.URL.Path, + }) + }) + return middleware.Auth(mux) } diff --git a/go/internal/governance/graph_mem.go b/go/internal/governance/graph_mem.go index cd5651f..89bc43a 100644 --- a/go/internal/governance/graph_mem.go +++ b/go/internal/governance/graph_mem.go @@ -241,12 +241,12 @@ func containsRelation(rel, substr string) bool { } // GetGraph 内存图谱的导出 -func (g *InMemoryGraph) GetGraph(namespace string) ([]map[string]interface{}, []map[string]interface{}) { +func (g *InMemoryGraph) GetGraph(namespace string, limit int) ([]map[string]interface{}, []map[string]interface{}) { g.mu.RLock() defer g.mu.RUnlock() var nodes []map[string]interface{} var edges []map[string]interface{} - // 导出匹配 namespace 的节点 + // 导出匹配 namespace 的节点（limit>0 时截断） for _, n := range g.nodes { if namespace == "" || n.Namespace == namespace { nodes = append(nodes, map[string]interface{}{ @@ -255,6 +255,9 @@ func (g *InMemoryGraph) GetGraph(namespace string) ([]map[string]interface{}, [] "type": n.Type, "namespace": n.Namespace, }) + if limit > 0 && len(nodes) >= limit { + break + } } } // 导出匹配 namespace 的边 diff --git a/go/internal/governance/graph_sqlite.go b/go/internal/governance/graph_sqlite.go index b261dbc..1484d84 100644 --- a/go/internal/governance/graph_sqlite.go +++ b/go/internal/governance/graph_sqlite.go @@ -469,15 +469,23 @@ func (gs *SQLiteGraphStore) Prune(minWeight float64) { execSQL(gs.db, `DELETE FROM graph_nodes WHERE id NOT IN (SELECT DISTINCT source FROM graph_edges UNION SELECT DISTINCT target FROM graph_edges)`) } -func (gs *SQLiteGraphStore) GetGraph(namespace string) ([]map[string]interface{}, []map[string]interface{}) { +func (gs *SQLiteGraphStore) GetGraph(namespace string, limit int) ([]map[string]interface{}, []map[string]interface{}) { nodes := []map[string]interface{}{} edges := []map[string]interface{}{} var nodeSQL string if namespace == "" || namespace == "all" { - nodeSQL = "SELECT id, name, type, namespace, properties FROM graph_nodes" + if limit > 0 { + nodeSQL = fmt.Sprintf("SELECT id, name, type, namespace, properties FROM graph_nodes ORDER BY pagerank DESC LIMIT %d", limit) + } else { + nodeSQL = "SELECT id, name, type, namespace, properties FROM graph_nodes" + } } else { - nodeSQL = "SELECT id, name, type, namespace, properties FROM graph_nodes WHERE namespace = '" + escape(namespace) + "'" + if limit > 0 { + nodeSQL = fmt.Sprintf("SELECT id, name, type, namespace, properties FROM graph_nodes WHERE namespace = '%s' ORDER BY pagerank DESC LIMIT %d", escape(namespace), limit) + } else { + nodeSQL = fmt.Sprintf("SELECT id, name, type, namespace, properties FROM graph_nodes WHERE namespace = '%s'", escape(namespace)) + } } rows := queryRows(gs.db, nodeSQL) @@ -493,19 +501,28 @@ func (gs *SQLiteGraphStore) GetGraph(namespace string) ([]map[string]interface{} }) } - var edgeSQL string - if namespace == "" || namespace == "all" { - edgeSQL = "SELECT id, source, target, relation, weight, namespace FROM graph_edges" - } else { - edgeSQL = "SELECT id, source, target, relation, weight, namespace FROM graph_edges WHERE namespace = '" + escape(namespace) + "'" - } + // 只获取这些节点的边 + if len(nodes) > 0 { + nids := make([]string, len(nodes)) + for i, n := range nodes { + nids[i] = "'" + escape(n["id"].(string)) + "'" + } + nidList := strings.Join(nids, ",") - edgeRows := queryRows(gs.db, edgeSQL) - for _, e := range edgeRows { - edges = append(edges, map[string]interface{}{ - "id": e["id"], "source": e["source"], "target": e["target"], - "relation": e["relation"], "weight": e["weight"], "namespace": e["namespace"], - }) + var edgeSQL string + if namespace == "" || namespace == "all" { + edgeSQL = fmt.Sprintf("SELECT id, source, target, relation, weight, namespace FROM graph_edges WHERE source IN (%s) AND target IN (%s)", nidList, nidList) + } else { + edgeSQL = fmt.Sprintf("SELECT id, source, target, relation, weight, namespace FROM graph_edges WHERE namespace = '%s' AND source IN (%s) AND target IN (%s)", escape(namespace), nidList, nidList) + } + + edgeRows := queryRows(gs.db, edgeSQL) + for _, e := range edgeRows { + edges = append(edges, map[string]interface{}{ + "id": e["id"], "source": e["source"], "target": e["target"], + "relation": e["relation"], "weight": e["weight"], "namespace": e["namespace"], + }) + } } return nodes, edges } diff --git a/go/internal/governance/graph_store.go b/go/internal/governance/graph_store.go index 1d42f79..eba6ae0 100644 --- a/go/internal/governance/graph_store.go +++ b/go/internal/governance/graph_store.go @@ -32,6 +32,6 @@ type GraphStore interface { PageRank(damping float64, iterations int) map[string]float64 EvidenceCount(entity string) int - // 导出完整图谱（供可视化） - GetGraph(namespace string) (nodes []map[string]interface{}, edges []map[string]interface{}) + // 导出完整图谱（供可视化），limit≤0 时不限制 + GetGraph(namespace string, limit int) (nodes []map[string]interface{}, edges []map[string]interface{}) } diff --git a/go/internal/storage/lancedb_ipc.go b/go/internal/storage/lancedb_ipc.go index 9c954df..c8f9435 100644 --- a/go/internal/storage/lancedb_ipc.go +++ b/go/internal/storage/lancedb_ipc.go @@ -142,9 +142,20 @@ func (rc *RustLanceDBClient) Search(table string, vector []float32, topK int, na Category string `json:"category"` Namespace string `json:"namespace"` IsDeleted bool `json:"is_deleted"` + CreatedAt string `json:"created_at"` + UpdatedAt string `json:"updated_at"` } json.Unmarshal([]byte(resp.ReportJSON), &raw) + // 解析时间（Rust 返回 RFC3339 字符串 → time.Time） + parseTime := func(s string) time.Time { + t, err := time.Parse(time.RFC3339, s) + if err != nil { + return time.Now() + } + return t + } + // 客户端 namespace 过滤（Rust lancedb only_if 不可靠） out := make([]models.MemoryRecord, 0, topK) for _, r := range raw { @@ -156,6 +167,8 @@ func (rc *RustLanceDBClient) Search(table string, vector []float32, topK int, na } out = append(out, models.MemoryRecord{ ID: r.ID, Content: r.Content, Category: r.Category, Namespace: r.Namespace, + CreatedAt: parseTime(r.CreatedAt), + UpdatedAt: parseTime(r.UpdatedAt), }) if len(out) >= topK { break diff --git a/go/static/graph.html b/go/static/graph.html index a7a3a34..a5d98e9 100644 --- a/go/static/graph.html +++ b/go/static/graph.html @@ -348,8 +348,8 @@ try { const url = namespace - ? `${API_BASE}/api/v1/graph/export?namespace=${encodeURIComponent(namespace)}` - : `${API_BASE}/api/v1/graph/export`; + ? `${API_BASE}/api/v1/graph/export?namespace=${encodeURIComponent(namespace)}&limit=200` + : `${API_BASE}/api/v1/graph/export?limit=300`; const resp = await fetch(url, { headers: { 'X-API-Key': API_KEY } diff --git a/rust/src/lancedb_ops.rs b/rust/src/lancedb_ops.rs index 8ae96a1..61628f5 100644 --- a/rust/src/lancedb_ops.rs +++ b/rust/src/lancedb_ops.rs @@ -14,7 +14,7 @@ use std::sync::Arc; use serde::{Deserialize, Serialize}; -use arrow::array::{ArrayRef, BooleanArray, Float32Array, Int32Array, StringArray, FixedSizeListArray}; +use arrow::array::{ArrayRef, BooleanArray, FixedSizeListArray, Float32Array, Float64Array, Int64Array, StringArray}; use arrow::datatypes::{DataType, Field, Schema}; use arrow::record_batch::{RecordBatch, RecordBatchIterator}; use futures::StreamExt; @@ -24,9 +24,9 @@ use lancedb::query::{ExecutableQuery, QueryBase}; pub struct MemoryRecord { pub id: String, pub agent_id: String, pub namespace: String, pub content: String, pub category: String, pub vector: Vec, - pub tier: String, pub importance: f32, pub quality_score: f32, - pub recall_count: i32, pub useful_count: i32, pub not_useful_count: i32, - pub freshness: String, pub version: i32, + pub tier: String, pub importance: f64, pub quality_score: f64, + pub recall_count: i64, pub useful_count: i64, pub not_useful_count: i64, + pub freshness: String, pub version: i64, pub version_history: String, pub source: String, pub volatile_flag: bool, pub is_deleted: bool, pub depends_on: String, pub derived_from: String, @@ -47,13 +47,13 @@ fn arrow_schema() -> Schema { Field::new("category", DataType::Utf8, false), Field::new("vector", DataType::FixedSizeList(Arc::new(Field::new("item", DataType::Float32, true)), 1024), true), Field::new("tier", DataType::Utf8, false), - Field::new("importance", DataType::Float32, false), - Field::new("quality_score", DataType::Float32, false), - Field::new("recall_count", DataType::Int32, false), - Field::new("useful_count", DataType::Int32, false), - Field::new("not_useful_count", DataType::Int32, false), + Field::new("importance", DataType::Float64, false), + Field::new("quality_score", DataType::Float64, false), + Field::new("recall_count", DataType::Int64, false), + Field::new("useful_count", DataType::Int64, false), + Field::new("not_useful_count", DataType::Int64, false), Field::new("freshness", DataType::Utf8, false), - Field::new("version", DataType::Int32, false), + Field::new("version", DataType::Int64, false), Field::new("version_history", DataType::Utf8, true), Field::new("source", DataType::Utf8, true), Field::new("volatile_flag", DataType::Boolean, false), @@ -108,13 +108,13 @@ impl LanceDBOps { category: col_str(&batch, i, "category"), vector: vec![], tier: col_str(&batch, i, "tier"), - importance: col_f32(&batch, i, "importance"), - quality_score: col_f32(&batch, i, "quality_score"), - recall_count: col_i32(&batch, i, "recall_count"), - useful_count: col_i32(&batch, i, "useful_count"), - not_useful_count: col_i32(&batch, i, "not_useful_count"), + importance: col_f64(&batch, i, "importance"), + quality_score: col_f64(&batch, i, "quality_score"), + recall_count: col_i64(&batch, i, "recall_count"), + useful_count: col_i64(&batch, i, "useful_count"), + not_useful_count: col_i64(&batch, i, "not_useful_count"), freshness: col_str(&batch, i, "freshness"), - version: col_i32(&batch, i, "version"), + version: col_i64(&batch, i, "version"), version_history: String::new(), source: col_str(&batch, i, "source"), volatile_flag: col_bool(&batch, i, "volatile_flag"), @@ -162,13 +162,13 @@ impl LanceDBOps { } else { vec![0.0_f32; 1024] }; vectors.push(v); tiers.push(r["tier"].as_str().unwrap_or("normal")); - importances.push(r["importance"].as_f64().unwrap_or(1.0) as f32); - quality_scores.push(r["quality_score"].as_f64().unwrap_or(0.0) as f32); - recall_counts.push(r["recall_count"].as_i64().unwrap_or(0) as i32); - useful_counts.push(r["useful_count"].as_i64().unwrap_or(0) as i32); - not_useful_counts.push(r["not_useful_count"].as_i64().unwrap_or(0) as i32); + importances.push(r["importance"].as_f64().unwrap_or(1.0)); + quality_scores.push(r["quality_score"].as_f64().unwrap_or(0.0)); + recall_counts.push(r["recall_count"].as_i64().unwrap_or(0)); + useful_counts.push(r["useful_count"].as_i64().unwrap_or(0)); + not_useful_counts.push(r["not_useful_count"].as_i64().unwrap_or(0)); freshnesses.push(r["freshness"].as_str().unwrap_or("fresh")); - versions.push(r["version"].as_i64().unwrap_or(1) as i32); + versions.push(r["version"].as_i64().unwrap_or(1)); version_histories.push(r["version_history"].as_str().unwrap_or("[]")); sources.push(r["source"].as_str().unwrap_or("")); volatile_flags.push(r["volatile_flag"].as_bool().unwrap_or(false)); @@ -197,13 +197,13 @@ impl LanceDBOps { Arc::new(StringArray::from(categories)), Arc::new(vec_array), Arc::new(StringArray::from(tiers)), - Arc::new(Float32Array::from(importances)), - Arc::new(Float32Array::from(quality_scores)), - Arc::new(Int32Array::from(recall_counts)), - Arc::new(Int32Array::from(useful_counts)), - Arc::new(Int32Array::from(not_useful_counts)), + Arc::new(Float64Array::from(importances)), + Arc::new(Float64Array::from(quality_scores)), + Arc::new(Int64Array::from(recall_counts)), + Arc::new(Int64Array::from(useful_counts)), + Arc::new(Int64Array::from(not_useful_counts)), Arc::new(StringArray::from(freshnesses)), - Arc::new(Int32Array::from(versions)), + Arc::new(Int64Array::from(versions)), Arc::new(StringArray::from(version_histories)), Arc::new(StringArray::from(sources)), Arc::new(BooleanArray::from(volatile_flags)), @@ -250,13 +250,13 @@ impl LanceDBOps { category: col_str(&batch, i, "category"), vector: vec![], tier: col_str(&batch, i, "tier"), - importance: col_f32(&batch, i, "importance"), - quality_score: col_f32(&batch, i, "quality_score"), - recall_count: col_i32(&batch, i, "recall_count"), - useful_count: col_i32(&batch, i, "useful_count"), - not_useful_count: col_i32(&batch, i, "not_useful_count"), + importance: col_f64(&batch, i, "importance"), + quality_score: col_f64(&batch, i, "quality_score"), + recall_count: col_i64(&batch, i, "recall_count"), + useful_count: col_i64(&batch, i, "useful_count"), + not_useful_count: col_i64(&batch, i, "not_useful_count"), freshness: col_str(&batch, i, "freshness"), - version: col_i32(&batch, i, "version"), + version: col_i64(&batch, i, "version"), version_history: String::new(), source: col_str(&batch, i, "source"), volatile_flag: col_bool(&batch, i, "volatile_flag"), @@ -311,15 +311,15 @@ fn col_str(b: &RecordBatch, r: usize, c: &str) -> String { .map(|a| a.value(r).to_string()) .unwrap_or_default() } -fn col_f32(b: &RecordBatch, r: usize, c: &str) -> f32 { +fn col_f64(b: &RecordBatch, r: usize, c: &str) -> f64 { b.column_by_name(c) - .and_then(|col| col.as_any().downcast_ref::()) + .and_then(|col| col.as_any().downcast_ref::()) .map(|a| a.value(r)) .unwrap_or(0.0) } -fn col_i32(b: &RecordBatch, r: usize, c: &str) -> i32 { +fn col_i64(b: &RecordBatch, r: usize, c: &str) -> i64 { b.column_by_name(c) - .and_then(|col| col.as_any().downcast_ref::()) + .and_then(|col| col.as_any().downcast_ref::()) .map(|a| a.value(r)) .unwrap_or(0) }