xiaoxue_admin
/
memoryweave
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: Search IPC 捕获 created_at/updated_at，recall 时间戳不再为零值 

lancedb_ipc.go Search() 反序列化 Rust 侧车返回结果时，
匿名 struct 缺少 created_at/updated_at 字段，导致所有
recall 结果 timestamp 为 time.Time{}（0001-01-01T00:00:00Z）。

修复：添加 CreatedAt/UpdatedAt string 字段 + 解析 RFC3339。
同时清理测试记忆。
This commit is contained in:
xiaowei 2026-05-30 10:31:25 +08:00
parent af14d7aa5f
commit ff935587bd
7 changed files with 115 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
go/internal/api/server.go
View File

@ -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)
}

7
go/internal/governance/graph_mem.go
View File

@ -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 的边

47
go/internal/governance/graph_sqlite.go
View File

@ -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
}

4
go/internal/governance/graph_store.go
View File

@ -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{})
}

13
go/internal/storage/lancedb_ipc.go
View File

@ -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

4
go/static/graph.html
View File

@ -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 }

76
rust/src/lancedb_ops.rs
View File

@ -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<f32>,
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::<Float32Array>())
.and_then(|col| col.as_any().downcast_ref::<Float64Array>())
.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::<Int32Array>())
.and_then(|col| col.as_any().downcast_ref::<Int64Array>())
.map(|a| a.value(r))
.unwrap_or(0)
}