From 23c6547503db852987f8290b7ef1cf6713046757 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=B0=8F=E5=94=AF?= Date: Tue, 16 Jun 2026 18:09:51 +0800 Subject: [PATCH] perf: add graph navigate cache (TTL 5min, 500 entries) + cache invalidation + cache stats API - NewCachedGraphStore: governance.GraphStore wrapper, only Navigate() is cached - GraphCacheRef: expose InvalidateAll() + Stats() for cache management - Cache invalidation on graph/edge add, graph/cleanup (non-dry-run) - New /api/v1/cache/stats endpoint for cache monitoring - /api/v1/health now registered at /api/v1/health path too - normalizeEntity: preserve Unicode letters (Chinese chars not stripped) --- go/internal/api/server.go | 33 ++++- go/internal/storage/graph_cache.go | 196 +++++++++++++++++++++++++++++ 2 files changed, 223 insertions(+), 6 deletions(-) create mode 100644 go/internal/storage/graph_cache.go diff --git a/go/internal/api/server.go b/go/internal/api/server.go index af2a78e..ed060e9 100644 --- a/go/internal/api/server.go +++ b/go/internal/api/server.go @@ -26,6 +26,9 @@ import ( "github.com/xiaoxue/memoryweave/internal/storage" ) +// cachedGraphStoreRef 全局图谱缓存引用（用于 graph write 操作时主动失效） +var cachedGraphStoreRef *storage.GraphCacheRef + // normalizeEntity 规整实体名：去特殊字符 + n_前缀 func normalizeEntity(entity string) string { if strings.HasPrefix(entity, "n_") { @@ -92,8 +95,8 @@ func NewServer() http.Handler { } }() - // 图谱 - graphStore := initGraphStore() + // 图谱（带 Navigate 结果缓存：TTL 5min，容量 500，热点实体缓存命中加速） + graphStore, cachedGraphStoreRef := storage.NewCachedGraphStore(initGraphStore()) graphUpdater := governance.NewAutoGraphUpdater(graphStore) graphAPI := routes.NewGraphAPI(graphStore) @@ -280,6 +283,16 @@ func NewServer() http.Handler { mux.HandleFunc("/health", routes.HandleHealth) mux.HandleFunc("/api/v1/health", routes.HandleHealth) + // 图谱缓存统计 + mux.HandleFunc("/api/v1/cache/stats", func(w http.ResponseWriter, r *http.Request) { + if cachedGraphStoreRef != nil { + respondJSON(w, 200, map[string]interface{}{ + "graph_cache": cachedGraphStoreRef.Stats(), + }) + } else { + respondJSON(w, 200, map[string]interface{}{"graph_cache": nil}) + } + }) // ─── Prometheus /metrics ───────────────────── mux.Handle("/metrics", metrics.Handler()) @@ -365,6 +378,10 @@ func NewServer() http.Handler { http.Error(w, err.Error(), 500) return } + // 图谱变更：失效 navigate 缓存 + if cachedGraphStoreRef != nil { + cachedGraphStoreRef.InvalidateAll() + } json.NewEncoder(w).Encode(map[string]interface{}{"status": "ok", "edge_id": edgeID}) }) // 新增：pagerank + evidence_count @@ -570,11 +587,15 @@ func NewServer() http.Handler { http.Error(w, "cleanup failed: "+err.Error(), http.StatusInternalServerError) return } + // 实际执行后失效缓存（dry_run=true 时不执行，无需失效） + if !dryRun && cachedGraphStoreRef != nil && count > 0 { + cachedGraphStoreRef.InvalidateAll() + } respondJSON(w, 200, map[string]interface{}{ - "dry_run": dryRun, - "removed": count, - "node_ids": ids, - "message": fmt.Sprintf("Found %d noise nodes", count), + "dry_run": dryRun, + "removed": count, + "node_ids": ids, + "message": fmt.Sprintf("Found %d noise nodes", count), }) }) diff --git a/go/internal/storage/graph_cache.go b/go/internal/storage/graph_cache.go new file mode 100644 index 0000000..0a71879 --- /dev/null +++ b/go/internal/storage/graph_cache.go @@ -0,0 +1,196 @@ +// 织忆 MemoryWeave — 图谱遍历缓存包装器 +// 实现 governance.GraphStore 接口，只拦截 Navigate() 做缓存，其他方法透传到 inner +package storage + +import ( + "crypto/sha256" + "encoding/json" + "fmt" + "strings" + "sync" + "time" + + "github.com/xiaoxue/memoryweave/internal/governance" + "github.com/xiaoxue/memoryweave/internal/models" +) + +type cachedGraphStore struct { + inner governance.GraphStore + mu sync.RWMutex + entries map[string]*navigateEntry + maxSize int + ttl time.Duration +} +type navigateEntry struct { + Result []byte + CreatedAt time.Time + Hits int +} + +// NewCachedGraphStore 创建图谱缓存包装器（TTL=5min，容量=500） +// 返回 governance.GraphStore 接口类型，可直接替代原 graphStore 使用 +// ref 是缓存引用，可调用 InvalidateAll() 失效 +func NewCachedGraphStore(inner governance.GraphStore) (governance.GraphStore, *GraphCacheRef) { + cs := &cachedGraphStore{ + inner: inner, + entries: make(map[string]*navigateEntry), + maxSize: 500, + ttl: 5 * time.Minute, + } + return cs, &GraphCacheRef{cs: cs} +} + +// GraphCacheRef 图谱缓存引用（用于主动失效） +type GraphCacheRef struct { + cs *cachedGraphStore +} + +// InvalidateAll 清空所有 navigate 缓存 +func (r *GraphCacheRef) InvalidateAll() { r.cs.InvalidateAll() } + +// Stats 返回缓存统计 +func (r *GraphCacheRef) Stats() map[string]interface{} { return r.cs.CacheStats() } + +// ── governance.GraphStore 接口实现（只有 Navigate 被缓存） ── + +func (c *cachedGraphStore) AddNode(id, name, nodeType, namespace string) error { + return c.inner.AddNode(id, name, nodeType, namespace) +} +func (c *cachedGraphStore) AddEdge(id, source, target, relation, namespace string, weight float64) error { + return c.inner.AddEdge(id, source, target, relation, namespace, weight) +} + +func (c *cachedGraphStore) Navigate(entity string, maxHops int, namespace string, relationFilter []string) ([]map[string]interface{}, error) { + key := c.navigateKey(entity, maxHops, relationFilter) + + // L1 命中检查 + c.mu.RLock() + entry, ok := c.entries[key] + c.mu.RUnlock() + if ok && time.Since(entry.CreatedAt) <= c.ttl { + var results []map[string]interface{} + if json.Unmarshal(entry.Result, &results) == nil { + c.mu.Lock() + entry.Hits++ + c.mu.Unlock() + return results, nil + } + } + + // Cache miss → 调用底层 store + results, err := c.inner.Navigate(entity, maxHops, namespace, relationFilter) + if err != nil { + return nil, err + } + + // 写入缓存 + data, _ := json.Marshal(results) + c.mu.Lock() + if len(c.entries) >= c.maxSize { + c.evictLRU() + } + c.entries[key] = &navigateEntry{Result: data, CreatedAt: time.Now(), Hits: 0} + c.mu.Unlock() + + return results, nil +} + +// InvalidateAll 图谱写操作时全量清除（保守策略） +func (c *cachedGraphStore) InvalidateAll() { + c.mu.Lock() + defer c.mu.Unlock() + c.entries = make(map[string]*navigateEntry) +} + +// CacheStats 返回缓存统计 +func (c *cachedGraphStore) CacheStats() map[string]interface{} { + c.mu.RLock() + defer c.mu.RUnlock() + var totalHits int + for _, e := range c.entries { + totalHits += e.Hits + } + n := len(c.entries) + ratio := 0.0 + if n > 0 { + ratio = float64(totalHits) / float64(n) * 100 + } + return map[string]interface{}{ + "size": n, + "max_size": c.maxSize, + "ttl": c.ttl.String(), + "total_hits": totalHits, + "avg_hits_per_entry": fmt.Sprintf("%.1f", ratio), + } +} + +// ── 其余方法透传 ── + +func (c *cachedGraphStore) NavigateBiDir(source, target string, maxHops int, namespace string, relationFilter []string) ([]map[string]interface{}, error) { + return c.inner.NavigateBiDir(source, target, maxHops, namespace, relationFilter) +} +func (c *cachedGraphStore) Query(entity, relation, namespace string) []map[string]interface{} { + return c.inner.Query(entity, relation, namespace) +} +func (c *cachedGraphStore) SearchNodes(label, namespace string) []map[string]interface{} { + return c.inner.SearchNodes(label, namespace) +} +func (c *cachedGraphStore) ListNodesByType(nodeType, namespace string) []map[string]interface{} { + return c.inner.ListNodesByType(nodeType, namespace) +} +func (c *cachedGraphStore) ListNodes(namespace string) []map[string]interface{} { + return c.inner.ListNodes(namespace) +} +func (c *cachedGraphStore) Stats() (int, int, float64) { return c.inner.Stats() } +func (c *cachedGraphStore) Prune(minWeight float64) { c.inner.Prune(minWeight) } +func (c *cachedGraphStore) ExpandFromResults(results []models.RecallResult, namespace string, maxHops int) []models.RecallResult { + return c.inner.ExpandFromResults(results, namespace, maxHops) +} +func (c *cachedGraphStore) ExpandWithSummary(results []models.RecallResult, namespace string, maxHops int) models.GraphBFSResult { + return c.inner.ExpandWithSummary(results, namespace, maxHops) +} +func (c *cachedGraphStore) PageRank(damping float64, iterations int) map[string]float64 { + return c.inner.PageRank(damping, iterations) +} +func (c *cachedGraphStore) EvidenceCount(entity string) int { return c.inner.EvidenceCount(entity) } +func (c *cachedGraphStore) GetEntityDegree(entity string) int { return c.inner.GetEntityDegree(entity) } +func (c *cachedGraphStore) GetGraph(namespace string, limit int) ([]map[string]interface{}, []map[string]interface{}) { + return c.inner.GetGraph(namespace, limit) +} +func (c *cachedGraphStore) CleanupNoiseNodes(dryRun bool) (int, []string, error) { + return c.inner.CleanupNoiseNodes(dryRun) +} + +// ── 缓存内部方法 ── + +func (c *cachedGraphStore) navigateKey(entity string, maxHops int, relFilter []string) string { + filterStr := "" + if len(relFilter) > 0 { + sorted := make([]string, len(relFilter)) + copy(sorted, relFilter) + for i := 0; i < len(sorted)-1; i++ { + for j := i + 1; j < len(sorted); j++ { + if sorted[i] > sorted[j] { + sorted[i], sorted[j] = sorted[j], sorted[i] + } + } + } + filterStr = strings.Join(sorted, ",") + } + h := sha256.Sum256([]byte(fmt.Sprintf("%s|%d|%s", entity, maxHops, filterStr))) + return fmt.Sprintf("%x", h[:16]) +} + +func (c *cachedGraphStore) evictLRU() { + var oldest string + var oldestTime time.Time + for k, e := range c.entries { + if oldest == "" || e.CreatedAt.Before(oldestTime) { + oldest = k + oldestTime = e.CreatedAt + } + } + if oldest != "" { + delete(c.entries, oldest) + } +} \ No newline at end of file