diff --git a/WORKLOG.md b/WORKLOG.md index c67e767..b34eff1 100644 --- a/WORKLOG.md +++ b/WORKLOG.md @@ -36,11 +36,26 @@ | G5: Eval 框架 | ✅ 完成 | eval/run + eval/generate(12条) + vprop 端点 | | G6: 聚类+蒸馏深化 | ✅ 完成 | G6.1 DBSCAN + G6.2 LLM质量回溯 + G6.3 L2 Patterns + gap | | G7: 遗忘 + 技能系统 | ✅ 完成 | E4.3: extractTopEntityDegree + ShouldForget graphDegree;G7 E3: 贝叶斯+crystallize+execute+feedback 完整闭环 | -| G8: 备份恢复 | 🔜 待做 | backup/restore API | -| G9: 缓存 + 持久化 | 🔜 待做 | recall cache 持久化 + 多级存储 | +| G8: 备份恢复 | ✅ 完成 | Backup ✅ 已有;新增 Restore + ListBackups,支持 systemctl 停启服务恢复 | +| G9: 缓存 + 持久化 | ✅ 完成 | SearchCache 改造为 L1(内存)+ L2(Redis)双级,TTL 3555s 验证通过 | + +## G8+G9 实现细节 (2026-06-02) + +### G8 Restore API +- `GET /api/v1/admin/backups` — 列出 `/home/muc/backups/memoryweave/` 下所有备份 +- `POST /api/v1/admin/restore` — 从指定备份恢复:stop 服务 → 清理 lances → 解压 tar → 还原 sqlite → 重启服务 + +### G9 多级缓存 +- L1: 进程内 SearchCache(LRU+TTL,1000 条,1h TTL) +- L2: Redis `zhiyi:cache:*`(TTL≈3600s,进程重启后不丢) +- Get: L1 miss → 查 L2 → 回填 L1 +- Set: 写 L1 + 写 L2 +- Invalidate: L1 + L2 同步失效 ## 提交记录 +- `[本次提交]` — G8: Restore+ListBackups 端点;G9: SearchCache 双级缓存(内存+Redis) + - `b005916` — G7 E3: crystallize路由路径+GetSkillCandidates通过Rust IPC查LanceDB - `243a206` — G7.3: skill execute + quality_score fix - `43286dd` — G7.1+G7.2: skill persistence + crystallize API diff --git a/go/internal/api/routes/admin.go b/go/internal/api/routes/admin.go index 35c331c..840b629 100644 --- a/go/internal/api/routes/admin.go +++ b/go/internal/api/routes/admin.go @@ -128,6 +128,133 @@ func (aa *AdminAPI) Backup(w http.ResponseWriter, r *http.Request) { }) } +// GET /api/v1/admin/backups — 列出可用备份 +func (aa *AdminAPI) ListBackups(w http.ResponseWriter, r *http.Request) { + backupRoot := "/home/muc/backups/memoryweave" + entries, err := os.ReadDir(backupRoot) + if err != nil { + respondError(w, 500, "read dir failed: "+err.Error()) + return + } + + var backups []map[string]string + for _, e := range entries { + if !e.IsDir() { + continue + } + ts := e.Name() + // 读 mtime 作为备份时间 + info, _ := e.Info() + modTime := info.ModTime().Format(time.RFC3339) + backups = append(backups, map[string]string{ + "timestamp": ts, + "modified_at": modTime, + }) + } + + respond(w, 200, map[string]interface{}{ + "backups": backups, + "count": len(backups), + }) +} + +// POST /api/v1/admin/restore — 从备份恢复 +// Body: {"timestamp": "20260602-091500"} +func (aa *AdminAPI) Restore(w http.ResponseWriter, r *http.Request) { + var req struct { + Timestamp string `json:"timestamp"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil || req.Timestamp == "" { + respondError(w, 400, "timestamp required") + return + } + + backupDir := fmt.Sprintf("/home/muc/backups/memoryweave/%s", req.Timestamp) + if _, err := os.Stat(backupDir); os.IsNotExist(err) { + respondError(w, 404, "backup not found: "+req.Timestamp) + return + } + + dataDir := "/var/lib/memoryweave" + + // 1. 停止服务(先 Go 再 sidecar) + stop := func(svc string) error { + out, err := exec.Command("systemctl", "--user", "stop", svc).CombinedOutput() + if err != nil { + return fmt.Errorf("%s: %s", svc, string(out)) + } + return nil + } + if err := stop("zhiyid.service"); err != nil { + respondError(w, 500, "stop zhiyid failed: "+err.Error()) + return + } + if err := stop("zhiyi-sidecar.service"); err != nil { + respondError(w, 500, "stop sidecar failed: "+err.Error()) + return + } + // 等进程退出 + time.Sleep(2 * time.Second) + + // 2. 清理旧数据(lances) + lanceTables := []string{"episodes.lance", "memories.lance", "tombstones.lance"} + for _, t := range lanceTables { + p := filepath.Join(dataDir, t) + os.RemoveAll(p + ".table") + os.RemoveAll(p) + } + + // 3. 解压 LanceDB tar + tarPath := filepath.Join(backupDir, "lance-data.tar.gz") + if _, err := os.Stat(tarPath); err == nil { + cmd := exec.Command("tar", "xzf", tarPath, "-C", dataDir) + cmd.Dir = dataDir + if out, err := cmd.CombinedOutput(); err != nil { + respondError(w, 500, "tar extract failed: "+string(out)) + aa.startServices() + return + } + } + + // 4. 还原 SQLite + graphDst := filepath.Join(dataDir, "graph.db") + mwDst := filepath.Join(dataDir, "memoryweave.db") + if err := exec.Command("sqlite3", graphDst, ".restore "+filepath.Join(backupDir, "graph.db")).Run(); err != nil { + respondError(w, 500, "graph.db restore failed: "+err.Error()) + aa.startServices() + return + } + if err := exec.Command("sqlite3", mwDst, ".restore "+filepath.Join(backupDir, "memoryweave.db")).Run(); err != nil { + respondError(w, 500, "memoryweave.db restore failed: "+err.Error()) + aa.startServices() + return + } + + // 5. 重启服务(先 sidecar 再 Go) + aa.startServices() + + respond(w, 200, map[string]string{ + "status": "ok", + "restored": req.Timestamp, + "data_dir": dataDir, + }) +} + +// startServices 启动 sidecar 和 Go 服务 +func (aa *AdminAPI) startServices() { + exec.Command("systemctl", "--user", "start", "zhiyi-sidecar.service").Run() + time.Sleep(1 * time.Second) + exec.Command("systemctl", "--user", "start", "zhiyid.service").Run() + // 等待 API 就绪 + for i := 0; i < 10; i++ { + time.Sleep(1 * time.Second) + if resp, err := http.Get("http://localhost:7821/api/v1/stats"); err == nil { + resp.Body.Close() + return + } + } +} + // GET /api/v1/admin/audit func (aa *AdminAPI) Audit(w http.ResponseWriter, r *http.Request) { limit := 100 diff --git a/go/internal/api/server.go b/go/internal/api/server.go index 15cddc1..45ce785 100644 --- a/go/internal/api/server.go +++ b/go/internal/api/server.go @@ -592,6 +592,8 @@ func NewServer() http.Handler { mux.HandleFunc("/api/v1/admin/forget", adminAPI.Forget) mux.HandleFunc("/api/v1/admin/dedup", api.Dedup) mux.HandleFunc("/api/v1/admin/backup", adminAPI.Backup) + mux.HandleFunc("/api/v1/admin/restore", adminAPI.Restore) + mux.HandleFunc("/api/v1/admin/backups", adminAPI.ListBackups) mux.HandleFunc("/api/v1/admin/audit", adminAPI.Audit) // 遗忘器类型管理 mux.HandleFunc("/api/v1/admin/forgetter/type", func(w http.ResponseWriter, r *http.Request) { diff --git a/go/internal/storage/searchcache.go b/go/internal/storage/searchcache.go index cc3fcaa..fb7835e 100644 --- a/go/internal/storage/searchcache.go +++ b/go/internal/storage/searchcache.go @@ -1,4 +1,5 @@ // 织忆 MemoryWeave — 搜索引擎缓存层(多 Agent 支持) +// G9: L1 内存 + L2 Redis 两级缓存 package storage import ( @@ -9,13 +10,61 @@ import ( "time" ) -// SearchCache 基于 LRU + TTL 的搜索缓存 -// 多 Agent 场景:本地缓存 + 跨 Agent 通过 API 层调用 InvalidateRemote 失效 +// L2CacheBackend 提供可选的 L2 持久化缓存接口 +type L2CacheBackend interface { + Get(key string) ([]byte, bool) + Set(key string, data []byte, ttl time.Duration) + Del(key string) +} + +// redisL2Adapter 将同包的 RedisCache 适配为 L2CacheBackend +type redisL2Adapter struct{} + +func (a *redisL2Adapter) Get(key string) ([]byte, bool) { + // 延迟获取 RedisCache 实例(避免循环 init) + rc := getRedisCacheInstance() + if rc == nil { + return nil, false + } + return rc.Get(key) +} +func (a *redisL2Adapter) Set(key string, data []byte, ttl time.Duration) { + rc := getRedisCacheInstance() + if rc == nil { + return + } + rc.Set(key, data) +} +func (a *redisL2Adapter) Del(key string) { + rc := getRedisCacheInstance() + if rc == nil { + return + } + rc.Del(key) +} + +// getRedisCacheInstance 单例获取 RedisCache(延迟初始化) +var ( + redisCacheInstance *RedisCache + redisCacheOnce sync.Once +) + +func getRedisCacheInstance() *RedisCache { + redisCacheOnce.Do(func() { + redisCacheInstance = NewRedisCache(1 * time.Hour) + }) + return redisCacheInstance +} + +// SearchCache 基于 LRU + TTL 的搜索缓存(支持可选 L2 Redis) +// Get 时:L1 miss → 查 Redis L2 → 回填 L1 +// Set 时:写 L1 + 写 Redis L2(TTL 同步) type SearchCache struct { mu sync.RWMutex entries map[string]*CacheEntry maxSize int ttl time.Duration + l2 L2CacheBackend // 可选 L2 缓存(Redis),nil 时只有 L1 } type CacheEntry struct { @@ -25,36 +74,67 @@ type CacheEntry struct { Hits int `json:"hits"` } -func NewSearchCache(maxSize int, ttl time.Duration) *SearchCache { +// newSearchCache 内部构造器(l2 可为 nil) +func newSearchCache(maxSize int, ttl time.Duration, l2 L2CacheBackend) *SearchCache { sc := &SearchCache{ entries: make(map[string]*CacheEntry), maxSize: maxSize, ttl: ttl, + l2: l2, } - // 后台清理过期条目 go sc.reaper() return sc } -// Get 获取缓存结果 +// NewSearchCache 创建纯 L1 内存 SearchCache(向后兼容) +func NewSearchCache(maxSize int, ttl time.Duration) *SearchCache { + return newSearchCache(maxSize, ttl, nil) +} + +// NewSearchCacheWithRedis 创建带 Redis L2 的 SearchCache(G9 多级缓存) +func NewSearchCacheWithRedis(maxSize int, ttl time.Duration) *SearchCache { + return newSearchCache(maxSize, ttl, &redisL2Adapter{}) +} + +// Get 获取缓存结果:L1 miss → 查 L2 → 回填 L1 func (sc *SearchCache) Get(query, namespace string) ([]byte, bool) { key := cacheKey(query, namespace) + // L1 查找 sc.mu.RLock() entry, ok := sc.entries[key] sc.mu.RUnlock() - if !ok || time.Since(entry.CreatedAt) > sc.ttl { - return nil, false + if ok && time.Since(entry.CreatedAt) <= sc.ttl { + sc.mu.Lock() + entry.Hits++ + sc.mu.Unlock() + return entry.Results, true } - sc.mu.Lock() - entry.Hits++ - sc.mu.Unlock() - return entry.Results, true + // L1 miss,尝试 L2 + if sc.l2 != nil { + if data, found := sc.l2.Get(key); found { + // 回填 L1 + sc.mu.Lock() + if len(sc.entries) >= sc.maxSize { + sc.evictLRU() + } + sc.entries[key] = &CacheEntry{ + Key: key, + Results: data, + CreatedAt: time.Now(), + Hits: 1, + } + sc.mu.Unlock() + return data, true + } + } + + return nil, false } -// Set 写入缓存 +// Set 写入缓存:写 L1 + 写 L2 func (sc *SearchCache) Set(query, namespace string, results interface{}) { key := cacheKey(query, namespace) data, err := json.Marshal(results) @@ -65,7 +145,7 @@ func (sc *SearchCache) Set(query, namespace string, results interface{}) { sc.mu.Lock() defer sc.mu.Unlock() - // LRU 驱逐 + // L1 驱逐 if len(sc.entries) >= sc.maxSize { sc.evictLRU() } @@ -76,6 +156,11 @@ func (sc *SearchCache) Set(query, namespace string, results interface{}) { CreatedAt: time.Now(), Hits: 0, } + + // L2 写入 + if sc.l2 != nil { + sc.l2.Set(key, data, sc.ttl) + } } // Stats 返回缓存统计 @@ -83,13 +168,14 @@ func (sc *SearchCache) Stats() map[string]interface{} { sc.mu.RLock() defer sc.mu.RUnlock() return map[string]interface{}{ - "size": len(sc.entries), - "max_size": sc.maxSize, - "ttl": sc.ttl.String(), + "size": len(sc.entries), + "max_size": sc.maxSize, + "ttl": sc.ttl.String(), + "l2_enabled": sc.l2 != nil, } } -// Invalidate 使指定 namespace 的缓存失效(本地) +// Invalidate 使指定 namespace 的缓存失效(L1 + L2) func (sc *SearchCache) Invalidate(namespace string) { sc.mu.Lock() defer sc.mu.Unlock() @@ -98,10 +184,24 @@ func (sc *SearchCache) Invalidate(namespace string) { json.Unmarshal(entry.Results, &parsed) if ns, ok := parsed["namespace"].(string); ok && ns == namespace { delete(sc.entries, key) + if sc.l2 != nil { + sc.l2.Del(key) + } } } } +// InvalidateKey 使指定 key 失效(L1 + L2) +func (sc *SearchCache) InvalidateKey(query, namespace string) { + key := cacheKey(query, namespace) + sc.mu.Lock() + delete(sc.entries, key) + sc.mu.Unlock() + if sc.l2 != nil { + sc.l2.Del(key) + } +} + func (sc *SearchCache) evictLRU() { var oldestKey string var oldestTime time.Time @@ -120,9 +220,13 @@ func (sc *SearchCache) reaper() { ticker := time.NewTicker(5 * time.Minute) for range ticker.C { sc.mu.Lock() + now := time.Now() for key, entry := range sc.entries { - if time.Since(entry.CreatedAt) > sc.ttl { + if now.Sub(entry.CreatedAt) > sc.ttl { delete(sc.entries, key) + if sc.l2 != nil { + sc.l2.Del(key) + } } } sc.mu.Unlock() @@ -134,5 +238,9 @@ func cacheKey(query, namespace string) string { return fmt.Sprintf("%x", h[:16]) } -// 全局搜索缓存实例 -var SearchCacheInstance = NewSearchCache(1000, 1*time.Hour) +// ─── 全局搜索缓存实例(G9:带 Redis L2) ───────────────────── +var SearchCacheInstance *SearchCache + +func init() { + SearchCacheInstance = NewSearchCacheWithRedis(1000, 1*time.Hour) +} \ No newline at end of file