223 lines
7.2 KiB
Go
223 lines
7.2 KiB
Go
// 织忆 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)
|
||
}
|
||
|
||
// CleanupScopedNodes 透传（inner 不支持该能力时静默返回 0；生产后端 SQLiteGraphStore 已实现）
|
||
func (c *cachedGraphStore) CleanupScopedNodes(dryRun bool, namespaces, nameContains []string) (int, []string, error) {
|
||
type scoped interface {
|
||
CleanupScopedNodes(dryRun bool, namespaces, nameContains []string) (int, []string, error)
|
||
}
|
||
if s, ok := c.inner.(scoped); ok {
|
||
return s.CleanupScopedNodes(dryRun, namespaces, nameContains)
|
||
}
|
||
return 0, nil, nil
|
||
}
|
||
|
||
// P0: FallbackTextSearch 透传到内层
|
||
func (c *cachedGraphStore) FallbackTextSearch(query, namespace string, limit int) []map[string]interface{} {
|
||
return c.inner.FallbackTextSearch(query, namespace, limit)
|
||
}
|
||
|
||
// P2: 信任评分透传
|
||
func (c *cachedGraphStore) AddEdgeFeedback(edgeID string, helpful bool) error {
|
||
return c.inner.AddEdgeFeedback(edgeID, helpful)
|
||
}
|
||
func (c *cachedGraphStore) IncrementEdgeRetrieval(edgeID string) error {
|
||
return c.inner.IncrementEdgeRetrieval(edgeID)
|
||
}
|
||
func (c *cachedGraphStore) UpdateEdgeTrustScores() error {
|
||
return c.inner.UpdateEdgeTrustScores()
|
||
}
|
||
|
||
// ── 缓存内部方法 ──
|
||
|
||
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)
|
||
}
|
||
} |