1011 lines
30 KiB
Go
1011 lines
30 KiB
Go
// 织忆 MemoryWeave — SQLite 知识图谱存储（CGO，实现 GraphStore 接口）
|
||
// 设计要求: SQLite 仅存 graph_nodes / graph_edges / version_history
|
||
// 不存 memories / episodes / tombstones（那属于 LanceDB 职责）
|
||
package governance
|
||
|
||
/*
|
||
#cgo LDFLAGS: -lsqlite3
|
||
#include <sqlite3.h>
|
||
#include <stdlib.h>
|
||
|
||
// busy_timeout 是 Go/C 混合文件中直接使用 C 函数
|
||
// sqlite3_busy_timeout 在 sqlite3.h 中声明
|
||
*/
|
||
import "C"
|
||
|
||
import (
|
||
"encoding/json"
|
||
"fmt"
|
||
"os"
|
||
"strings"
|
||
"sync"
|
||
"unicode"
|
||
"unicode/utf8"
|
||
"unsafe"
|
||
|
||
"github.com/xiaoxue/memoryweave/internal/models"
|
||
)
|
||
|
||
// SQLiteGraphStore CGO 直连 SQLite，独立于主存储的 LanceDB SQLite 文件
|
||
type SQLiteGraphStore struct {
|
||
mu sync.RWMutex
|
||
db *C.sqlite3
|
||
path string
|
||
}
|
||
|
||
func NewSQLiteGraphStore(dbPath string) (*SQLiteGraphStore, error) {
|
||
if dbPath == "" {
|
||
dbPath = "/var/lib/memoryweave/graph.db"
|
||
}
|
||
|
||
// 清理 stale WAL/SHM 文件（防止旧进程崩溃后留下这些文件导致锁失败）
|
||
// 只有当主 db 文件存在时才清理（避免误删新建库的场景）
|
||
if _, err := os.Stat(dbPath); err == nil {
|
||
for _, suffix := range []string{"-wal", "-shm", "-journal"} {
|
||
f := dbPath + suffix
|
||
if _, err := os.Stat(f); err == nil {
|
||
os.Remove(f)
|
||
}
|
||
}
|
||
}
|
||
|
||
cPath := C.CString(dbPath)
|
||
defer C.free(unsafe.Pointer(cPath))
|
||
|
||
var db *C.sqlite3
|
||
rc := C.sqlite3_open(cPath, &db)
|
||
if rc != C.SQLITE_OK {
|
||
msg := C.GoString(C.sqlite3_errmsg(db))
|
||
C.sqlite3_close(db)
|
||
return nil, fmt.Errorf("sqlite open graph: %s", msg)
|
||
}
|
||
|
||
// WAL 模式：写操作不阻塞读，大幅降低图谱导航超时概率
|
||
_ = execSQL(db, "PRAGMA journal_mode=WAL;")
|
||
// busy_timeout 降为 3s（WAL 模式下读不阻塞写，3s 足够）
|
||
C.sqlite3_busy_timeout(db, 3000)
|
||
|
||
gs := &SQLiteGraphStore{db: db, path: dbPath}
|
||
if err := gs.migrate(); err != nil {
|
||
C.sqlite3_close(db)
|
||
return nil, err
|
||
}
|
||
return gs, nil
|
||
}
|
||
|
||
func (gs *SQLiteGraphStore) migrate() error {
|
||
sqls := []string{
|
||
`CREATE TABLE IF NOT EXISTS graph_nodes (
|
||
id TEXT PRIMARY KEY,
|
||
name TEXT NOT NULL,
|
||
type TEXT NOT NULL,
|
||
namespace TEXT NOT NULL DEFAULT '',
|
||
properties TEXT DEFAULT '{}',
|
||
pagerank REAL DEFAULT 1.0,
|
||
created_at TEXT NOT NULL,
|
||
last_updated_at TEXT NOT NULL DEFAULT ''
|
||
)`,
|
||
`CREATE TABLE IF NOT EXISTS graph_edges (
|
||
id TEXT PRIMARY KEY,
|
||
source TEXT NOT NULL,
|
||
target TEXT NOT NULL,
|
||
relation TEXT NOT NULL,
|
||
weight REAL DEFAULT 1.0,
|
||
evidence_count INTEGER DEFAULT 1,
|
||
namespace TEXT NOT NULL DEFAULT '',
|
||
created_at TEXT NOT NULL,
|
||
FOREIGN KEY (source) REFERENCES graph_nodes(id),
|
||
FOREIGN KEY (target) REFERENCES graph_nodes(id)
|
||
)`,
|
||
`CREATE TABLE IF NOT EXISTS version_history (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
memory_id TEXT NOT NULL,
|
||
version INTEGER NOT NULL,
|
||
content TEXT NOT NULL,
|
||
updated_by TEXT DEFAULT '',
|
||
source TEXT DEFAULT '',
|
||
trigger TEXT DEFAULT '',
|
||
reason TEXT DEFAULT '',
|
||
timestamp TEXT NOT NULL
|
||
)`,
|
||
`CREATE INDEX IF NOT EXISTS idx_gn_namespace ON graph_nodes(namespace)`,
|
||
`CREATE INDEX IF NOT EXISTS idx_ge_source ON graph_edges(source)`,
|
||
`CREATE INDEX IF NOT EXISTS idx_ge_target ON graph_edges(target)`,
|
||
`CREATE INDEX IF NOT EXISTS idx_ge_namespace ON graph_edges(namespace)`,
|
||
`CREATE INDEX IF NOT EXISTS idx_vh_memory ON version_history(memory_id)`,
|
||
}
|
||
for _, s := range sqls {
|
||
cSQL := C.CString(s)
|
||
rc := C.sqlite3_exec(gs.db, cSQL, nil, nil, nil)
|
||
C.free(unsafe.Pointer(cSQL))
|
||
if rc != C.SQLITE_OK {
|
||
return fmt.Errorf("migrate graph: %s", C.GoString(C.sqlite3_errmsg(gs.db)))
|
||
}
|
||
}
|
||
|
||
// 迁移：已有库可能缺少 pagerank 列（DESIGN.md 要求但原建表 SQL 漏了）
|
||
gs.migrateAddColumn("graph_nodes", "pagerank", "REAL DEFAULT 1.0")
|
||
// 迁移：缺少 last_updated_at 列（sidecar 的 consolidate prune 步骤需要）
|
||
gs.migrateAddColumn("graph_nodes", "last_updated_at", "TEXT NOT NULL DEFAULT ''")
|
||
// 为已有行初始化 last_updated_at = created_at
|
||
sqlInit := `UPDATE graph_nodes SET last_updated_at = created_at WHERE last_updated_at = '';`
|
||
cInit := C.CString(sqlInit)
|
||
C.sqlite3_exec(gs.db, cInit, nil, nil, nil)
|
||
C.free(unsafe.Pointer(cInit))
|
||
|
||
// 修复孤儿边：自动补充缺失的节点
|
||
gs.repairOrphanEdges()
|
||
|
||
return nil
|
||
}
|
||
|
||
func (gs *SQLiteGraphStore) repairOrphanEdges() {
|
||
sql := `INSERT OR IGNORE INTO graph_nodes (id, name, type, namespace, created_at)
|
||
SELECT DISTINCT source, source, 'entity', COALESCE(namespace,''), datetime('now')
|
||
FROM graph_edges WHERE source NOT IN (SELECT id FROM graph_nodes);
|
||
INSERT OR IGNORE INTO graph_nodes (id, name, type, namespace, created_at)
|
||
SELECT DISTINCT target, target, 'entity', COALESCE(namespace,''), datetime('now')
|
||
FROM graph_edges WHERE target NOT IN (SELECT id FROM graph_nodes);`
|
||
cSQL := C.CString(sql)
|
||
C.sqlite3_exec(gs.db, cSQL, nil, nil, nil)
|
||
C.free(unsafe.Pointer(cSQL))
|
||
}
|
||
|
||
// migrateAddColumn 幂等添加列（列存在时忽略错误）
|
||
func (gs *SQLiteGraphStore) migrateAddColumn(table, column, def string) {
|
||
sql := fmt.Sprintf("ALTER TABLE %s ADD COLUMN %s %s", table, column, def)
|
||
cSQL := C.CString(sql)
|
||
rc := C.sqlite3_exec(gs.db, cSQL, nil, nil, nil)
|
||
C.free(unsafe.Pointer(cSQL))
|
||
// SQLITE_ERROR 表示列已存在，忽略即可
|
||
if rc != C.SQLITE_OK && rc != C.SQLITE_ERROR {
|
||
// log but don't fail
|
||
}
|
||
}
|
||
|
||
func (gs *SQLiteGraphStore) Close() error {
|
||
if gs.db != nil {
|
||
C.sqlite3_close(gs.db)
|
||
gs.db = nil
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ─── GraphStore 接口实现 ────────────────────────────────
|
||
|
||
func (gs *SQLiteGraphStore) AddNode(id, name, nodeType, namespace string) error {
|
||
gs.mu.Lock()
|
||
defer gs.mu.Unlock()
|
||
sql := fmt.Sprintf(
|
||
"INSERT OR REPLACE INTO graph_nodes (id, name, type, namespace, created_at) VALUES ('%s', '%s', '%s', '%s', datetime('now'))",
|
||
escape(id), escape(name), escape(nodeType), escape(namespace))
|
||
return execSQL(gs.db, sql)
|
||
}
|
||
|
||
func (gs *SQLiteGraphStore) AddEdge(id, source, target, relation, namespace string, weight float64) error {
|
||
gs.mu.Lock()
|
||
defer gs.mu.Unlock()
|
||
|
||
// 自动创建缺失的源/目标节点；已存在的升级 namespace（default → 具体ns）
|
||
for _, node := range []struct{ id, name string }{{source, source}, {target, target}} {
|
||
// 先尝试 INSERT — 如果节点不存在就创建
|
||
sql := fmt.Sprintf(
|
||
"INSERT OR IGNORE INTO graph_nodes (id, name, type, namespace, created_at) VALUES ('%s', '%s', 'entity', '%s', datetime('now'))",
|
||
escape(node.id), escape(node.name), escape(namespace))
|
||
if err := execSQL(gs.db, sql); err != nil {
|
||
return err
|
||
}
|
||
// 如果 namespace 非空非 default，顺手升级 default 节点
|
||
if namespace != "" && namespace != "default" {
|
||
updateSQL := fmt.Sprintf(
|
||
"UPDATE graph_nodes SET namespace='%s' WHERE id='%s' AND namespace='default'",
|
||
escape(namespace), escape(node.id))
|
||
_ = execSQL(gs.db, updateSQL) // 忽略错误（节点不存在）不影响
|
||
}
|
||
}
|
||
|
||
sql := fmt.Sprintf(
|
||
"INSERT OR REPLACE INTO graph_edges (id, source, target, relation, weight, namespace, created_at) VALUES ('%s', '%s', '%s', '%s', %f, '%s', datetime('now'))",
|
||
escape(id), escape(source), escape(target), escape(relation), weight, escape(namespace))
|
||
return execSQL(gs.db, sql)
|
||
}
|
||
|
||
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}
|
||
var paths []map[string]interface{}
|
||
|
||
for hop := 1; hop <= maxHops && len(queue) > 0; hop++ {
|
||
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 AND %s",
|
||
escape(node), nsClause, relClause)
|
||
edges := queryRows(gs.db, sql)
|
||
for _, edge := range edges {
|
||
target := edge["target"].(string)
|
||
if !visited[target] {
|
||
visited[target] = true
|
||
next = append(next, target)
|
||
paths = append(paths, map[string]interface{}{
|
||
"from": node, "to": target, "relation": edge["relation"],
|
||
"weight": edge["weight"], "hop": hop,
|
||
})
|
||
}
|
||
}
|
||
}
|
||
queue = next
|
||
}
|
||
return paths, nil
|
||
}
|
||
|
||
// 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, relFilter []string) ([]map[string]interface{}, error) {
|
||
if source == target {
|
||
return []map[string]interface{}{
|
||
{
|
||
"nodes": []string{source},
|
||
"edges": []struct{}{},
|
||
"score": 1.0,
|
||
},
|
||
}, nil
|
||
}
|
||
|
||
gs.mu.RLock()
|
||
defer gs.mu.RUnlock()
|
||
|
||
nsClause := buildNamespaceClause(namespace)
|
||
|
||
// 正向 BFS 数据结构
|
||
type fwdNode struct {
|
||
parent string
|
||
edgeID string
|
||
relation string
|
||
weight float64
|
||
hop int
|
||
pathProd float64 // 累积权重乘积
|
||
}
|
||
fwd := make(map[string]*fwdNode)
|
||
fwd[source] = &fwdNode{hop: 0, pathProd: 1.0}
|
||
fwdQ := []string{source}
|
||
fwdVisited := map[string]bool{source: true}
|
||
|
||
// 反向 BFS 数据结构
|
||
type bwdNode struct {
|
||
parent string
|
||
edgeID string
|
||
relation string
|
||
weight float64
|
||
hop int
|
||
pathProd float64
|
||
}
|
||
bwd := make(map[string]*bwdNode)
|
||
bwd[target] = &bwdNode{hop: 0, pathProd: 1.0}
|
||
bwdQ := []string{target}
|
||
bwdVisited := map[string]bool{target: true}
|
||
|
||
// 正向步长：ceil(maxHops/2)
|
||
fwdHops := (maxHops + 1) / 2
|
||
// 反向步长：floor(maxHops/2)
|
||
bwdHops := maxHops / 2
|
||
|
||
// 执行正向 BFS
|
||
for hop := 1; hop <= fwdHops && len(fwdQ) > 0; hop++ {
|
||
var next []string
|
||
for _, node := range fwdQ {
|
||
sql := fmt.Sprintf(
|
||
"SELECT e.id, e.target, e.relation, e.weight FROM graph_edges e WHERE e.source = '%s' AND %s",
|
||
escape(node), nsClause)
|
||
edges := queryRows(gs.db, sql)
|
||
for _, edge := range edges {
|
||
targetID := edge["target"].(string)
|
||
if fwdVisited[targetID] {
|
||
continue
|
||
}
|
||
fwdVisited[targetID] = true
|
||
w := edge["weight"].(float64)
|
||
fwd[targetID] = &fwdNode{
|
||
parent: node,
|
||
edgeID: edge["id"].(string),
|
||
relation: edge["relation"].(string),
|
||
weight: w,
|
||
hop: hop,
|
||
pathProd: fwd[node].pathProd * w,
|
||
}
|
||
next = append(next, targetID)
|
||
}
|
||
}
|
||
fwdQ = next
|
||
}
|
||
|
||
// 执行反向 BFS
|
||
for hop := 1; hop <= bwdHops && len(bwdQ) > 0; hop++ {
|
||
var next []string
|
||
for _, node := range bwdQ {
|
||
// 反向要找所有指向 node 的边（target = node）
|
||
sql := fmt.Sprintf(
|
||
"SELECT e.id, e.source, e.relation, e.weight FROM graph_edges e WHERE e.target = '%s' AND %s",
|
||
escape(node), nsClause)
|
||
edges := queryRows(gs.db, sql)
|
||
for _, edge := range edges {
|
||
srcID := edge["source"].(string)
|
||
if bwdVisited[srcID] {
|
||
continue
|
||
}
|
||
bwdVisited[srcID] = true
|
||
w := edge["weight"].(float64)
|
||
bwd[srcID] = &bwdNode{
|
||
parent: node,
|
||
edgeID: edge["id"].(string),
|
||
relation: edge["relation"].(string),
|
||
weight: w,
|
||
hop: hop,
|
||
pathProd: bwd[node].pathProd * w,
|
||
}
|
||
next = append(next, srcID)
|
||
}
|
||
}
|
||
bwdQ = next
|
||
}
|
||
|
||
// 寻找相遇节点
|
||
var results []PathResult
|
||
|
||
for nodeID, fn := range fwd {
|
||
if bn, ok := bwd[nodeID]; ok {
|
||
// 重建从 source → nodeID 的路径（正向）
|
||
fwdPathNodes := []string{nodeID}
|
||
fwdPathEdges := make([]map[string]interface{}, 0)
|
||
cur := nodeID
|
||
for cur != source {
|
||
n := fwd[cur]
|
||
if n == nil {
|
||
break
|
||
}
|
||
fwdPathEdges = append(fwdPathEdges, map[string]interface{}{
|
||
"source": n.parent, "target": cur,
|
||
"relation": n.relation, "weight": n.weight,
|
||
})
|
||
fwdPathNodes = append(fwdPathNodes, n.parent)
|
||
cur = n.parent
|
||
}
|
||
// 逆序：从 source 到 meeting
|
||
for i, j := 0, len(fwdPathNodes)-1; i < j; i, j = i+1, j-1 {
|
||
fwdPathNodes[i], fwdPathNodes[j] = fwdPathNodes[j], fwdPathNodes[i]
|
||
}
|
||
for i, j := 0, len(fwdPathEdges)-1; i < j; i, j = i+1, j-1 {
|
||
fwdPathEdges[i], fwdPathEdges[j] = fwdPathEdges[j], fwdPathEdges[i]
|
||
}
|
||
|
||
// 重建从 nodeID → target 的路径（反向，翻转方向）
|
||
cur = nodeID
|
||
for cur != target {
|
||
n := bwd[cur]
|
||
if n == nil {
|
||
break
|
||
}
|
||
// 反向 BFS 的 parent 是 target 侧，所以边是从 cur → n.parent
|
||
fwdPathEdges = append(fwdPathEdges, map[string]interface{}{
|
||
"source": cur, "target": n.parent,
|
||
"relation": n.relation, "weight": n.weight,
|
||
})
|
||
fwdPathNodes = append(fwdPathNodes, n.parent)
|
||
cur = n.parent
|
||
}
|
||
|
||
score := fn.pathProd * bn.pathProd
|
||
if score > 0 {
|
||
results = append(results, PathResult{
|
||
Nodes: fwdPathNodes,
|
||
Edges: fwdPathEdges,
|
||
Score: score,
|
||
})
|
||
}
|
||
}
|
||
}
|
||
|
||
// 按 score 降序排序，取 top 3
|
||
sortResultsByScore(results)
|
||
|
||
// 转换为接口格式
|
||
out := make([]map[string]interface{}, 0, len(results))
|
||
for _, r := range results {
|
||
if len(out) >= 3 {
|
||
break
|
||
}
|
||
out = append(out, map[string]interface{}{
|
||
"nodes": r.Nodes,
|
||
"edges": r.Edges,
|
||
"score": r.Score,
|
||
})
|
||
}
|
||
|
||
if len(out) == 0 {
|
||
// E1.2: 无相遇节点时返回 unreachable，而非降级为单向邻居
|
||
return []map[string]interface{}{{
|
||
"unreachable": true,
|
||
"source": source,
|
||
"target": target,
|
||
"max_hops": maxHops,
|
||
}}, nil
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// PathResult 路径查找结果
|
||
type PathResult struct {
|
||
Nodes []string
|
||
Edges []map[string]interface{}
|
||
Score float64
|
||
}
|
||
|
||
// deriveNamespaceForGraph 将 namespace 转为图谱中的实际格式
|
||
// hermes → hermes-main, shared → shared, default → default
|
||
func deriveNamespaceForGraph(ns string) string {
|
||
if ns == "" {
|
||
return ""
|
||
}
|
||
// already full form
|
||
if strings.HasSuffix(ns, "-main") || ns == "shared" || ns == "default" {
|
||
return ns
|
||
}
|
||
// bare name → full form (hermes → hermes-main)
|
||
return ns + "-main"
|
||
}
|
||
|
||
func buildNamespaceClause(namespace string) string {
|
||
if namespace == "" {
|
||
return "1=1"
|
||
}
|
||
// 确保用图谱中的实际格式
|
||
derived := deriveNamespaceForGraph(namespace)
|
||
if derived == "shared" {
|
||
return "(e.namespace = 'shared')"
|
||
}
|
||
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++ {
|
||
for j := i + 1; j < len(results); j++ {
|
||
if results[j].Score > results[i].Score {
|
||
results[i], results[j] = results[j], results[i]
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
func (gs *SQLiteGraphStore) Query(entity, relation, namespace string) []map[string]interface{} {
|
||
gs.mu.RLock()
|
||
defer gs.mu.RUnlock()
|
||
// 命名空间兼容
|
||
nsClause := fmt.Sprintf("e.namespace IN ('%s', 'default')", escape(namespace))
|
||
if namespace == "" {
|
||
nsClause = "1=1"
|
||
}
|
||
sql := fmt.Sprintf(
|
||
"SELECT e.source, e.target, e.relation, e.weight FROM graph_edges e WHERE e.relation = '%s' AND %s AND (e.source = '%s' OR e.target = '%s')",
|
||
escape(relation), nsClause, escape(entity), escape(entity))
|
||
return queryRows(gs.db, sql)
|
||
}
|
||
|
||
func (gs *SQLiteGraphStore) Stats() (nodeCount, edgeCount int, density float64) {
|
||
gs.mu.RLock()
|
||
defer gs.mu.RUnlock()
|
||
|
||
nodeCount = queryInt(gs.db, "SELECT COUNT(*) FROM graph_nodes")
|
||
edgeCount = queryInt(gs.db, "SELECT COUNT(*) FROM graph_edges")
|
||
if nodeCount > 0 {
|
||
maxEdges := nodeCount * (nodeCount - 1)
|
||
density = float64(edgeCount) / float64(maxEdges)
|
||
if density > 1 {
|
||
density = 1
|
||
}
|
||
}
|
||
return
|
||
}
|
||
|
||
func (gs *SQLiteGraphStore) Prune(minWeight float64) {
|
||
gs.mu.Lock()
|
||
defer gs.mu.Unlock()
|
||
execSQL(gs.db, fmt.Sprintf("DELETE FROM graph_edges WHERE weight < %f", minWeight))
|
||
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, limit int) ([]map[string]interface{}, []map[string]interface{}) {
|
||
nodes := []map[string]interface{}{}
|
||
edges := []map[string]interface{}{}
|
||
|
||
var nodeSQL string
|
||
if namespace == "" || namespace == "all" {
|
||
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 {
|
||
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)
|
||
for _, n := range rows {
|
||
props := n["properties"].(string)
|
||
var properties map[string]interface{}
|
||
if props != "" {
|
||
json.Unmarshal([]byte(props), &properties)
|
||
}
|
||
nodes = append(nodes, map[string]interface{}{
|
||
"id": n["id"], "name": n["name"], "type": n["type"],
|
||
"namespace": n["namespace"], "properties": properties,
|
||
})
|
||
}
|
||
|
||
// 只获取这些节点的边
|
||
if len(nodes) > 0 {
|
||
nids := make([]string, len(nodes))
|
||
for i, n := range nodes {
|
||
nids[i] = "'" + escape(n["id"].(string)) + "'"
|
||
}
|
||
nidList := strings.Join(nids, ",")
|
||
|
||
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
|
||
}
|
||
|
||
func (gs *SQLiteGraphStore) ExpandFromResults(results []models.RecallResult, namespace string, maxHops int) []models.RecallResult {
|
||
// 从 recall 结果提取实体，展开图谱邻居
|
||
expanded := make([]models.RecallResult, 0, len(results)*2)
|
||
expanded = append(expanded, results...)
|
||
|
||
seen := make(map[string]bool) // 追踪已访问的记忆 ID
|
||
seenEntities := make(map[string]bool) // 追踪已访问的实体（用于去重扩展结果）
|
||
for _, r := range results {
|
||
if seen[r.ID] {
|
||
continue
|
||
}
|
||
seen[r.ID] = true
|
||
|
||
entities := extractPotentialEntities(r.Content)
|
||
for _, entity := range entities {
|
||
if seenEntities[entity] {
|
||
continue
|
||
}
|
||
seenEntities[entity] = true
|
||
|
||
// SQLite 图谱节点 ID 格式: n_{entity_name}，需 normalizeEntityID 转换
|
||
nodeID := normalizeEntityID(entity)
|
||
paths, _ := gs.Navigate(nodeID, maxHops, namespace, nil)
|
||
for _, p := range paths {
|
||
// Navigate 返回字段: from, to, relation, weight, hop
|
||
var neighbor, rel string
|
||
from, _ := p["from"].(string)
|
||
to, _ := p["to"].(string)
|
||
relVal, _ := p["relation"].(string)
|
||
if from == entity && to != "" {
|
||
neighbor = to
|
||
rel = relVal
|
||
} else if to == entity && from != "" {
|
||
neighbor = from
|
||
rel = "↩ " + relVal
|
||
}
|
||
if neighbor == "" {
|
||
continue
|
||
}
|
||
if seenEntities[neighbor] {
|
||
continue
|
||
}
|
||
seenEntities[neighbor] = true
|
||
expanded = append(expanded, models.RecallResult{
|
||
Content: fmt.Sprintf("[graph] %s --[%s]--> %s", entity, rel, neighbor),
|
||
Score: r.Score * 0.5,
|
||
})
|
||
}
|
||
}
|
||
}
|
||
return expanded
|
||
}
|
||
|
||
// ExpandWithSummary BFS 扩展 + 生成汇总语句 — E1 图谱导航增强
|
||
// 从 recall 结果提取实体，进行多跳扩展，返回扩展关系列表和一句话汇总
|
||
func (gs *SQLiteGraphStore) ExpandWithSummary(results []models.RecallResult, namespace string, maxHops int) models.GraphBFSResult {
|
||
if maxHops <= 0 {
|
||
maxHops = 2
|
||
}
|
||
|
||
seenEntities := make(map[string]bool)
|
||
var relations []models.ExpandedRelation
|
||
|
||
for _, r := range results {
|
||
entities := extractPotentialEntities(r.Content)
|
||
for _, entity := range entities {
|
||
if seenEntities[entity] {
|
||
continue
|
||
}
|
||
seenEntities[entity] = true
|
||
|
||
nodeID := normalizeEntityID(entity)
|
||
paths, _ := gs.Navigate(nodeID, maxHops, namespace, nil)
|
||
for _, p := range paths {
|
||
from, _ := p["from"].(string)
|
||
to, _ := p["to"].(string)
|
||
rel, _ := p["relation"].(string)
|
||
weight, _ := p["weight"].(float64)
|
||
hop, _ := p["hop"].(int)
|
||
|
||
// 归一化显示名（去掉 n_ 前缀）
|
||
fromName := strings.TrimPrefix(from, "n_")
|
||
toName := strings.TrimPrefix(to, "n_")
|
||
|
||
rel = strings.TrimSpace(rel)
|
||
if rel == "" {
|
||
rel = "RELATED_TO"
|
||
}
|
||
|
||
relations = append(relations, models.ExpandedRelation{
|
||
From: fromName,
|
||
To: toName,
|
||
Relation: rel,
|
||
Hops: hop,
|
||
Weight: weight,
|
||
Score: r.Score * weight,
|
||
})
|
||
}
|
||
}
|
||
}
|
||
|
||
// 生成汇总语句
|
||
summary := buildBFSSummary(relations)
|
||
return models.GraphBFSResult{
|
||
ExpandedRelations: relations,
|
||
Summary: summary,
|
||
}
|
||
}
|
||
|
||
// extractPotentialEntities 从文本中提取可能作为图谱实体的关键词（支持中文连续字符）
|
||
func extractPotentialEntities(text string) []string {
|
||
var entities []string
|
||
seen := make(map[string]bool)
|
||
runes := []rune(text)
|
||
for i := 0; i < len(runes); {
|
||
r := runes[i]
|
||
// 中文字符：收集连续的中文字符序列（2-8字）
|
||
if r >= 0x4E00 && r <= 0x9FFF {
|
||
start := i
|
||
i++
|
||
for i < len(runes) && runes[i] >= 0x4E00 && runes[i] <= 0x9FFF {
|
||
i++
|
||
}
|
||
chinese := string(runes[start:i])
|
||
// 不等式：2 <= len(chinese) <= 8（字符数，非字节数）
|
||
if utf8.RuneCountInString(chinese) >= 2 && utf8.RuneCountInString(chinese) <= 8 && !seen[chinese] {
|
||
seen[chinese] = true
|
||
entities = append(entities, chinese)
|
||
}
|
||
continue
|
||
}
|
||
// 非中文字符，收集整个单词
|
||
start := i
|
||
for i < len(runes) {
|
||
r2 := runes[i]
|
||
if r2 >= 0x4E00 && r2 <= 0x9FFF {
|
||
break // 遇到汉字则停止
|
||
}
|
||
i++
|
||
}
|
||
if i-start < 2 {
|
||
continue
|
||
}
|
||
w := string(runes[start:i])
|
||
// 去除首尾标点
|
||
w = strings.Trim(w, ",.;:!?，。；：！？、\"'()（）[]【】")
|
||
if len(w) < 2 {
|
||
continue
|
||
}
|
||
// 英文大写字母开头的词
|
||
first := []rune(w)
|
||
if len(first) > 0 && first[0] >= 'A' && first[0] <= 'Z' {
|
||
lower := strings.ToLower(w)
|
||
if !seen[lower] {
|
||
seen[lower] = true
|
||
entities = append(entities, w) // 保留原始大小写
|
||
}
|
||
}
|
||
}
|
||
return entities
|
||
}
|
||
|
||
// normalizeEntityID 将自由文本转为实体 ID 格式
|
||
func normalizeEntityID(name string) string {
|
||
clean := strings.Map(func(r rune) rune {
|
||
if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') || r == '_' || r == '-' || r == ' ' {
|
||
return r
|
||
}
|
||
if unicode.IsLetter(r) {
|
||
return r
|
||
}
|
||
return '_'
|
||
}, strings.TrimSpace(name))
|
||
clean = strings.ToLower(clean)
|
||
clean = strings.ReplaceAll(clean, " ", "_")
|
||
for strings.Contains(clean, "__") {
|
||
clean = strings.ReplaceAll(clean, "__", "_")
|
||
}
|
||
clean = strings.Trim(clean, "_")
|
||
if clean == "" {
|
||
return "n_unknown"
|
||
}
|
||
return "n_" + clean
|
||
}
|
||
|
||
// SearchNodes 按 label 模糊搜索节点（§2.5.4 match 格式兼容）
|
||
func (gs *SQLiteGraphStore) SearchNodes(label, namespace string) []map[string]interface{} {
|
||
gs.mu.RLock()
|
||
defer gs.mu.RUnlock()
|
||
nsClause := "1=1"
|
||
if namespace != "" {
|
||
nsClause = fmt.Sprintf("namespace = '%s'", escape(namespace))
|
||
}
|
||
sql := fmt.Sprintf(
|
||
"SELECT id, name, type, namespace, pagerank FROM graph_nodes WHERE name LIKE '%%%s%%' AND %s LIMIT 100",
|
||
escape(label), nsClause)
|
||
return queryRows(gs.db, sql)
|
||
}
|
||
|
||
// ListNodesByType 按 type 列出节点
|
||
func (gs *SQLiteGraphStore) ListNodesByType(nodeType, namespace string) []map[string]interface{} {
|
||
gs.mu.RLock()
|
||
defer gs.mu.RUnlock()
|
||
nsClause := "1=1"
|
||
if namespace != "" {
|
||
nsClause = fmt.Sprintf("namespace = '%s'", escape(namespace))
|
||
}
|
||
sql := fmt.Sprintf(
|
||
"SELECT id, name, type, namespace, pagerank FROM graph_nodes WHERE type = '%s' AND %s ORDER BY pagerank DESC LIMIT 100",
|
||
escape(nodeType), nsClause)
|
||
return queryRows(gs.db, sql)
|
||
}
|
||
|
||
// ListNodes 列出所有节点
|
||
func (gs *SQLiteGraphStore) ListNodes(namespace string) []map[string]interface{} {
|
||
gs.mu.RLock()
|
||
defer gs.mu.RUnlock()
|
||
nsClause := "1=1"
|
||
if namespace != "" {
|
||
nsClause = fmt.Sprintf("namespace = '%s'", escape(namespace))
|
||
}
|
||
sql := fmt.Sprintf(
|
||
"SELECT id, name, type, namespace, pagerank FROM graph_nodes WHERE %s ORDER BY pagerank DESC LIMIT 100", nsClause)
|
||
return queryRows(gs.db, sql)
|
||
}
|
||
|
||
func (gs *SQLiteGraphStore) PageRank(damping float64, iterations int) map[string]float64 {
|
||
gs.mu.RLock()
|
||
defer gs.mu.RUnlock()
|
||
|
||
// 获取所有节点
|
||
nodes := queryStrSlice(gs.db, "SELECT id FROM graph_nodes")
|
||
n := float64(len(nodes))
|
||
if n == 0 {
|
||
return nil
|
||
}
|
||
|
||
ranks := make(map[string]float64)
|
||
base := (1.0 - damping) / n
|
||
for _, id := range nodes {
|
||
ranks[id] = 1.0 / n
|
||
}
|
||
|
||
outEdges := make(map[string][]struct {
|
||
target string
|
||
weight float64
|
||
})
|
||
|
||
for _, node := range nodes {
|
||
sql := fmt.Sprintf("SELECT target, weight FROM graph_edges WHERE source = '%s'", escape(node))
|
||
rows := queryRows(gs.db, sql)
|
||
for _, r := range rows {
|
||
outEdges[node] = append(outEdges[node], struct {
|
||
target string
|
||
weight float64
|
||
}{r["target"].(string), r["weight"].(float64)})
|
||
}
|
||
}
|
||
|
||
for iter := 0; iter < iterations; iter++ {
|
||
newRanks := make(map[string]float64)
|
||
for _, node := range nodes {
|
||
rank := base
|
||
for src, edges := range outEdges {
|
||
totalWt := 0.0
|
||
for _, e := range edges {
|
||
totalWt += e.weight
|
||
}
|
||
for _, e := range edges {
|
||
if e.target == node && totalWt > 0 {
|
||
rank += damping * ranks[src] * e.weight / totalWt
|
||
}
|
||
}
|
||
}
|
||
newRanks[node] = rank
|
||
}
|
||
ranks = newRanks
|
||
}
|
||
return ranks
|
||
}
|
||
|
||
func (gs *SQLiteGraphStore) EvidenceCount(entity string) int {
|
||
gs.mu.RLock()
|
||
defer gs.mu.RUnlock()
|
||
var sum int
|
||
for _, r := range queryRows(gs.db, fmt.Sprintf("SELECT SUM(evidence_count) as s FROM graph_edges WHERE source = '%s' OR target = '%s'", escape(entity), escape(entity))) {
|
||
if v, ok := r["s"]; ok {
|
||
switch x := v.(type) {
|
||
case int:
|
||
sum += x
|
||
case int64:
|
||
sum += int(x)
|
||
case float64:
|
||
sum += int(x)
|
||
}
|
||
}
|
||
}
|
||
return sum
|
||
}
|
||
|
||
// GetEntityDegree E4.3: 返回实体的图谱度（入度+出度），度越高越优先保留
|
||
func (gs *SQLiteGraphStore) GetEntityDegree(entity string) int {
|
||
return gs.EvidenceCount(entity)
|
||
}
|
||
|
||
// ─── CGO 工具 ──────────────────────────────────────────
|
||
|
||
// UpdatePageRanks 批量更新节点的 pagerank 值（§2.5.5）
|
||
func (gs *SQLiteGraphStore) UpdatePageRanks(ranks map[string]float64) {
|
||
gs.mu.Lock()
|
||
defer gs.mu.Unlock()
|
||
for nodeID, rank := range ranks {
|
||
sql := fmt.Sprintf("UPDATE graph_nodes SET pagerank = %f WHERE id = '%s'", rank, escape(nodeID))
|
||
_ = execSQL(gs.db, sql)
|
||
}
|
||
}
|
||
|
||
func execSQL(db *C.sqlite3, sql string) error {
|
||
cSQL := C.CString(sql)
|
||
defer C.free(unsafe.Pointer(cSQL))
|
||
rc := C.sqlite3_exec(db, cSQL, nil, nil, nil)
|
||
if rc != C.SQLITE_OK {
|
||
return fmt.Errorf("sqlite: %s", C.GoString(C.sqlite3_errmsg(db)))
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func queryInt(db *C.sqlite3, sql string) int {
|
||
cSQL := C.CString(sql)
|
||
defer C.free(unsafe.Pointer(cSQL))
|
||
var stmt *C.sqlite3_stmt
|
||
rc := C.sqlite3_prepare_v2(db, cSQL, C.int(len(sql)), &stmt, nil)
|
||
if rc != C.SQLITE_OK {
|
||
return 0
|
||
}
|
||
defer C.sqlite3_finalize(stmt)
|
||
if C.sqlite3_step(stmt) == C.SQLITE_ROW {
|
||
return int(C.sqlite3_column_int(stmt, 0))
|
||
}
|
||
return 0
|
||
}
|
||
|
||
func queryStrSlice(db *C.sqlite3, sql string) []string {
|
||
var result []string
|
||
cSQL := C.CString(sql)
|
||
defer C.free(unsafe.Pointer(cSQL))
|
||
var stmt *C.sqlite3_stmt
|
||
rc := C.sqlite3_prepare_v2(db, cSQL, C.int(len(sql)), &stmt, nil)
|
||
if rc != C.SQLITE_OK {
|
||
return result
|
||
}
|
||
defer C.sqlite3_finalize(stmt)
|
||
for C.sqlite3_step(stmt) == C.SQLITE_ROW {
|
||
result = append(result, C.GoString((*C.char)(unsafe.Pointer(C.sqlite3_column_text(stmt, 0)))))
|
||
}
|
||
return result
|
||
}
|
||
|
||
func queryRows(db *C.sqlite3, sql string) []map[string]interface{} {
|
||
var results []map[string]interface{}
|
||
cSQL := C.CString(sql)
|
||
defer C.free(unsafe.Pointer(cSQL))
|
||
var stmt *C.sqlite3_stmt
|
||
rc := C.sqlite3_prepare_v2(db, cSQL, C.int(len(sql)), &stmt, nil)
|
||
if rc != C.SQLITE_OK {
|
||
return results
|
||
}
|
||
defer C.sqlite3_finalize(stmt)
|
||
|
||
colCount := int(C.sqlite3_column_count(stmt))
|
||
for C.sqlite3_step(stmt) == C.SQLITE_ROW {
|
||
row := make(map[string]interface{})
|
||
for i := 0; i < colCount; i++ {
|
||
name := C.GoString((*C.char)(unsafe.Pointer(C.sqlite3_column_name(stmt, C.int(i)))))
|
||
switch C.sqlite3_column_type(stmt, C.int(i)) {
|
||
case C.SQLITE_INTEGER:
|
||
row[name] = int(C.sqlite3_column_int(stmt, C.int(i)))
|
||
case C.SQLITE_FLOAT:
|
||
row[name] = float64(C.sqlite3_column_double(stmt, C.int(i)))
|
||
case C.SQLITE_TEXT:
|
||
row[name] = C.GoString((*C.char)(unsafe.Pointer(C.sqlite3_column_text(stmt, C.int(i)))))
|
||
default:
|
||
row[name] = nil
|
||
}
|
||
}
|
||
results = append(results, row)
|
||
}
|
||
return results
|
||
}
|
||
|
||
func escape(s string) string {
|
||
result := ""
|
||
for _, ch := range s {
|
||
if ch == '\'' {
|
||
result += "''"
|
||
} else {
|
||
result += string(ch)
|
||
}
|
||
}
|
||
return result
|
||
}
|