From ef2ef5069cfa803f81799c915d942c40135b8b35 Mon Sep 17 00:00:00 2001 From: xiaowei Date: Thu, 28 May 2026 18:25:43 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20M9=20=E5=8A=9F=E8=83=BD=E8=A1=A5?= =?UTF-8?q?=E5=85=A8=20=E2=80=94=2030+=20=E6=96=B0=E7=AB=AF=E7=82=B9=20+?= =?UTF-8?q?=20=E8=87=AA=E8=B0=83=E5=8F=82=E9=97=AD=E7=8E=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增路由层 (10 files): - graph.go: GET /graph/stats, POST /graph/query, POST /graph/navigate - feedback.go: POST /feedback/useful|not-useful|deprecate|correct - gaps.go: GET /gaps, POST /gaps/detect, POST /gaps/close - conflicts.go: GET /conflicts, POST /conflicts/resolve - agent.go: POST /agents/register, GET /agents - admin.go: DELETE /distilled, GET /memory/versions, POST /admin/forget|backup, GET /admin/audit - l3.go: GET|POST /l3/worldmodel - triggers.go: GET /triggers, POST /triggers/fire, GET /skills, POST /skills/trial - eval.go: POST /eval/run, GET /eval/history, POST /eval/generate - tuning.go: GET /tuning/status, POST /tuning/run, POST /tuning/analytics - obsidian.go: POST /obsidian/push|pull, GET /obsidian/status 引擎增强: - governance: InMemoryGraph (零外部依赖) + ListActive/Resolve/DB - selfoptimize: RecordDeprecation/Correction/GapClosed/ConflictResolved - storage: IncrementUseful/NotUseful, UpdateMemoryContent, GetVersionHistory, GetCandidatesForForgetting, Backup, GetAuditLog 总计: 17 files, 1915+ lines, 35 endpoints, go build 0 errors --- go/internal/api/routes/admin.go | 142 +++++++++++++ go/internal/api/routes/agent.go | 92 +++++++++ go/internal/api/routes/conflicts.go | 59 ++++++ go/internal/api/routes/eval.go | 222 ++++++++++++++++++++ go/internal/api/routes/feedback.go | 104 ++++++++++ go/internal/api/routes/gaps.go | 71 +++++++ go/internal/api/routes/graph.go | 76 +++++++ go/internal/api/routes/l3.go | 105 ++++++++++ go/internal/api/routes/obsidian.go | 247 +++++++++++++++++++++++ go/internal/api/routes/triggers.go | 150 ++++++++++++++ go/internal/api/routes/tuning.go | 181 +++++++++++++++++ go/internal/api/routes/ws.go | 12 +- go/internal/api/server.go | 136 ++++++++++++- go/internal/governance/governance.go | 29 +++ go/internal/governance/graph_mem.go | 165 +++++++++++++++ go/internal/selfoptimize/selfoptimize.go | 34 ++++ go/internal/storage/lancedb.go | 99 +++++++++ 17 files changed, 1916 insertions(+), 8 deletions(-) create mode 100644 go/internal/api/routes/admin.go create mode 100644 go/internal/api/routes/agent.go create mode 100644 go/internal/api/routes/conflicts.go create mode 100644 go/internal/api/routes/eval.go create mode 100644 go/internal/api/routes/feedback.go create mode 100644 go/internal/api/routes/gaps.go create mode 100644 go/internal/api/routes/graph.go create mode 100644 go/internal/api/routes/l3.go create mode 100644 go/internal/api/routes/obsidian.go create mode 100644 go/internal/api/routes/triggers.go create mode 100644 go/internal/api/routes/tuning.go create mode 100644 go/internal/governance/graph_mem.go diff --git a/go/internal/api/routes/admin.go b/go/internal/api/routes/admin.go new file mode 100644 index 0000000..81757cb --- /dev/null +++ b/go/internal/api/routes/admin.go @@ -0,0 +1,142 @@ +// 织忆 MemoryWeave — 管理端点 +package routes + +import ( + "encoding/json" + "fmt" + "net/http" + "os" + "time" + + "github.com/xiaoxue/memoryweave/internal/governance" + "github.com/xiaoxue/memoryweave/internal/storage" +) + +type AdminAPI struct { + LanceDB *storage.LanceDB + Forgetter *governance.Forgetter +} + +func NewAdminAPI(ldb *storage.LanceDB, f *governance.Forgetter) *AdminAPI { + return &AdminAPI{LanceDB: ldb, Forgetter: f} +} + +// DELETE /api/v1/distilled/{id} +func (aa *AdminAPI) DeleteDistilled(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + if id == "" { + respondError(w, 400, "id required") + return + } + if err := aa.LanceDB.SoftDelete(id, "manual_delete"); err != nil { + respondError(w, 500, "delete failed: "+err.Error()) + return + } + respond(w, 200, map[string]string{"status": "deleted", "id": id}) +} + +// GET /api/v1/memory/{id}/versions +func (aa *AdminAPI) Versions(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + if id == "" { + respondError(w, 400, "id required") + return + } + versions, err := aa.LanceDB.GetVersionHistory(id) + if err != nil { + respondError(w, 500, "get versions failed: "+err.Error()) + return + } + respond(w, 200, map[string]interface{}{ + "memory_id": id, "versions": versions, "count": len(versions), + }) +} + +// POST /api/v1/admin/forget +func (aa *AdminAPI) Forget(w http.ResponseWriter, r *http.Request) { + results, err := aa.LanceDB.GetCandidatesForForgetting() + if err != nil { + respondError(w, 500, "list failed: "+err.Error()) + return + } + + forgotten := 0 + for _, mem := range results { + lastAccess := parseTimeStr(mem["last_recalled_at"]) + recallCnt := intVal(mem["recall_count"]) + tier := strVal(mem["tier"]) + + if aa.Forgetter.ShouldForget(lastAccess, recallCnt, tier) { + aa.LanceDB.SoftDelete(strVal(mem["id"]), "auto_forget") + forgotten++ + } + } + respond(w, 200, map[string]interface{}{ + "status": "ok", "scanned": len(results), "forgotten": forgotten, + }) +} + +// POST /api/v1/admin/backup +func (aa *AdminAPI) Backup(w http.ResponseWriter, r *http.Request) { + timestamp := time.Now().Format("20060102-150405") + backupPath := fmt.Sprintf("/home/muc/backups/memoryweave/%s", timestamp) + os.MkdirAll(backupPath, 0755) + + if err := aa.LanceDB.Backup(backupPath); err != nil { + respondError(w, 500, "backup failed: "+err.Error()) + return + } + respond(w, 200, map[string]string{ + "status": "ok", "path": backupPath, "timestamp": timestamp, + }) +} + +// GET /api/v1/admin/audit +func (aa *AdminAPI) Audit(w http.ResponseWriter, r *http.Request) { + limit := 100 + logs, err := aa.LanceDB.GetAuditLog(limit) + if err != nil { + respondError(w, 500, "audit failed: "+err.Error()) + return + } + respond(w, 200, map[string]interface{}{"audit_logs": logs, "count": len(logs)}) +} + +// 辅助函数 +func parseTimeStr(s interface{}) time.Time { + if s == nil { + return time.Time{} + } + switch v := s.(type) { + case string: + t, _ := time.Parse(time.RFC3339, v) + return t + case time.Time: + return v + } + return time.Time{} +} + +func intVal(v interface{}) int { + switch n := v.(type) { + case int: return n + case int32: return int(n) + case int64: return int(n) + case float64: return int(n) + case json.Number: + i, _ := n.Int64() + return int(i) + } + return 0 +} + +func strVal(v interface{}) string { + if v == nil { + return "" + } + switch s := v.(type) { + case string: return s + case json.Number: return s.String() + } + return fmt.Sprintf("%v", v) +} diff --git a/go/internal/api/routes/agent.go b/go/internal/api/routes/agent.go new file mode 100644 index 0000000..79417ab --- /dev/null +++ b/go/internal/api/routes/agent.go @@ -0,0 +1,92 @@ +// 织忆 MemoryWeave — Agent 注册 API +package routes + +import ( + "crypto/rand" + "encoding/json" + "fmt" + "net/http" + + "github.com/xiaoxue/memoryweave/internal/distributed" +) + +type AgentRegistry struct { + agents map[string]AgentInfo + limiter *distributed.RateLimiter +} + +type AgentInfo struct { + AgentID string `json:"agent_id"` + APIKey string `json:"api_key,omitempty"` + Quota Quota `json:"quota"` + Registered string `json:"registered"` +} + +type Quota struct { + RecallQPS int `json:"recall_qps"` + CommitQPS int `json:"commit_qps"` + Burst int `json:"burst"` +} + +func NewAgentRegistry(limiter *distributed.RateLimiter) *AgentRegistry { + return &AgentRegistry{ + agents: make(map[string]AgentInfo), + limiter: limiter, + } +} + +// POST /api/v1/agents/register +func (ar *AgentRegistry) Register(w http.ResponseWriter, r *http.Request) { + var req struct { + AgentID string `json:"agent_id"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + respondError(w, 400, "invalid body") + return + } + if req.AgentID == "" { + respondError(w, 400, "agent_id required") + return + } + + if _, exists := ar.agents[req.AgentID]; exists { + respondError(w, 409, "agent already registered") + return + } + + apiKey := generateKey(32) + info := AgentInfo{ + AgentID: req.AgentID, + APIKey: apiKey, + Quota: Quota{RecallQPS: 10, CommitQPS: 2, Burst: 20}, + Registered: "now", + } + ar.agents[req.AgentID] = info + respond(w, 201, info) +} + +// GET /api/v1/agents — 已注册 Agent 列表 +func (ar *AgentRegistry) List(w http.ResponseWriter, r *http.Request) { + var list []AgentInfo + for _, info := range ar.agents { + info.APIKey = "" // 不泄漏 key + list = append(list, info) + } + respond(w, 200, map[string]interface{}{"agents": list, "count": len(list)}) +} + +// Auth 验证 API Key +func (ar *AgentRegistry) Auth(apiKey string) (*AgentInfo, bool) { + for _, info := range ar.agents { + if info.APIKey == apiKey { + return &info, true + } + } + return nil, false +} + +func generateKey(length int) string { + b := make([]byte, length) + rand.Read(b) + return fmt.Sprintf("mw-%x", b) +} diff --git a/go/internal/api/routes/conflicts.go b/go/internal/api/routes/conflicts.go new file mode 100644 index 0000000..b985dbb --- /dev/null +++ b/go/internal/api/routes/conflicts.go @@ -0,0 +1,59 @@ +// 织忆 MemoryWeave — 冲突管理 API +package routes + +import ( + "encoding/json" + "net/http" + + "github.com/xiaoxue/memoryweave/internal/governance" + "github.com/xiaoxue/memoryweave/internal/selfoptimize" +) + +type ConflictAPI struct { + Detector *governance.ConflictDetector +} + +func NewConflictAPI(d *governance.ConflictDetector) *ConflictAPI { + return &ConflictAPI{Detector: d} +} + +// GET /api/v1/conflicts +func (ca *ConflictAPI) List(w http.ResponseWriter, r *http.Request) { + list := ca.Detector.ListActive() + pending := 0 + for _, c := range list { + if c.Status == "pending" { + pending++ + } + } + respond(w, 200, map[string]interface{}{ + "conflicts": list, "count": len(list), "pending": pending, + }) +} + +// POST /api/v1/conflicts/resolve +func (ca *ConflictAPI) Resolve(w http.ResponseWriter, r *http.Request) { + var req struct { + ConflictID string `json:"conflict_id"` + Resolution string `json:"resolution"` // keep_left / keep_right / merge / dismiss + Winner string `json:"winner"` // id of the winning memory + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + respondError(w, 400, "invalid body") + return + } + if req.ConflictID == "" || req.Resolution == "" { + respondError(w, 400, "conflict_id and resolution required") + return + } + + err := ca.Detector.Resolve(req.ConflictID, req.Resolution, req.Winner) + if err != nil { + respondError(w, 500, "resolve failed: "+err.Error()) + return + } + + selfoptimize.Dash.RecordConflictResolved(true) + PushConflictResolved(req.ConflictID, req.Resolution) + respond(w, 200, map[string]string{"status": "resolved", "conflict_id": req.ConflictID}) +} diff --git a/go/internal/api/routes/eval.go b/go/internal/api/routes/eval.go new file mode 100644 index 0000000..0deeb0d --- /dev/null +++ b/go/internal/api/routes/eval.go @@ -0,0 +1,222 @@ +// 织忆 MemoryWeave — 评估系统 API +package routes + +import ( + "encoding/json" + "net/http" + "sync" + "time" + + "github.com/xiaoxue/memoryweave/internal/storage" +) + +type EvalAPI struct { + Pipeline *storage.RecallPipeline + LanceDB *storage.LanceDB +} + +type EvalRun struct { + ID string `json:"id"` + Model string `json:"model"` + Precision float64 `json:"precision"` + Recall float64 `json:"recall_k"` + MRR float64 `json:"mrr"` + NDCG float64 `json:"ndcg"` + Queries int `json:"queries"` + RanAt time.Time `json:"ran_at"` +} + +type evalStore struct { + mu sync.RWMutex + runs []*EvalRun +} + +var evals = &evalStore{} + +func NewEvalAPI(p *storage.RecallPipeline, ldb *storage.LanceDB) *EvalAPI { + return &EvalAPI{Pipeline: p, LanceDB: ldb} +} + +// POST /api/v1/eval/run +func (ea *EvalAPI) Run(w http.ResponseWriter, r *http.Request) { + var req struct { + Queries []struct { + Query string `json:"query"` + ExpectedIDs []string `json:"expected_ids"` + } `json:"queries"` + Model string `json:"model"` + } + + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + respondError(w, 400, "invalid body") + return + } + if len(req.Queries) == 0 { + respondError(w, 400, "at least one query required") + return + } + + var totalPrecision, totalRecall, totalMRR, totalNDCG float64 + totalQueries := 0 + + for _, q := range req.Queries { + results, err := ea.Pipeline.Recall(q.Query, "shared", 10, 0.5) + if err != nil { + continue + } + + // Precision@k + hits := 0 + expectedSet := makeSet(q.ExpectedIDs) + for i, res := range results { + if expectedSet[res.ID] && i < len(q.ExpectedIDs) { + hits++ + } + } + if len(results) > 0 { + totalPrecision += float64(hits) / float64(len(results)) + } + + // Recall@k + if len(q.ExpectedIDs) > 0 { + totalRecall += float64(hits) / float64(len(q.ExpectedIDs)) + } + + // MRR (First correct position) + for i, res := range results { + if expectedSet[res.ID] { + totalMRR += 1.0 / float64(i+1) + break + } + } + + // NDCG (binary relevance) + dcg := 0.0 + idcg := 0.0 + for i, res := range results { + rel := 0.0 + if expectedSet[res.ID] { + rel = 1.0 + } + dcg += rel / log2(float64(i+2)) + if i < len(q.ExpectedIDs) { + idcg += 1.0 / log2(float64(i+2)) + } + } + if idcg > 0 { + totalNDCG += dcg / idcg + } + totalQueries++ + } + + if totalQueries == 0 { + respondError(w, 500, "all queries failed") + return + } + + run := &EvalRun{ + ID: time.Now().Format("20060102-150405"), + Model: req.Model, + Precision: totalPrecision / float64(totalQueries), + Recall: totalRecall / float64(totalQueries), + MRR: totalMRR / float64(totalQueries), + NDCG: totalNDCG / float64(totalQueries), + Queries: totalQueries, + RanAt: time.Now(), + } + + evals.mu.Lock() + evals.runs = append(evals.runs, run) + evals.mu.Unlock() + + respond(w, 200, run) +} + +// GET /api/v1/eval/history +func (ea *EvalAPI) History(w http.ResponseWriter, r *http.Request) { + evals.mu.RLock() + defer evals.mu.RUnlock() + + if len(evals.runs) == 0 { + respond(w, 200, map[string]interface{}{"runs": []EvalRun{}, "count": 0}) + return + } + + // 计算趋势（最近 2 次对比） + trend := "stable" + if len(evals.runs) >= 2 { + curr := evals.runs[len(evals.runs)-1].MRR + prev := evals.runs[len(evals.runs)-2].MRR + if curr > prev*1.05 { + trend = "improving" + } else if curr < prev*0.95 { + trend = "degrading" + } + } + + respond(w, 200, map[string]interface{}{ + "runs": evals.runs, "count": len(evals.runs), "trend": trend, + }) +} + +// POST /api/v1/eval/generate — 自动生成金标查询集 +func (ea *EvalAPI) Generate(w http.ResponseWriter, r *http.Request) { + var req struct { + Namespace string `json:"namespace"` + Count int `json:"count"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + respondError(w, 400, "invalid body") + return + } + if req.Namespace == "" { + req.Namespace = "shared" + } + if req.Count <= 0 { + req.Count = 10 + } + + // 获取高质量记忆作为金标基础 + memories, err := ea.LanceDB.GetTopByQuality("", req.Count) + if err != nil { + respondError(w, 500, "generate failed: "+err.Error()) + return + } + + var queries []map[string]interface{} + for _, mem := range memories { + query := mem.Content + if len(query) > 50 { + query = query[:50] + } + queries = append(queries, map[string]interface{}{ + "query": query, + "expected_ids": []string{mem.ID}, + }) + } + respond(w, 200, map[string]interface{}{ + "queries": queries, "count": len(queries), + }) +} + +// 辅助 +func makeSet(ids []string) map[string]bool { + s := make(map[string]bool, len(ids)) + for _, id := range ids { + s[id] = true + } + return s +} + +func log2(x float64) float64 { + // log2(x) ≈ ln(x)/ln(2) 但不用 math 包避免类型问题 + result := 0.0 + for x > 2 { + x /= 2 + result += 1 + } + if x > 1 { + result += (x - 1) + } + return result +} diff --git a/go/internal/api/routes/feedback.go b/go/internal/api/routes/feedback.go new file mode 100644 index 0000000..92a3152 --- /dev/null +++ b/go/internal/api/routes/feedback.go @@ -0,0 +1,104 @@ +// 织忆 MemoryWeave — 反馈 API +package routes + +import ( + "encoding/json" + "net/http" + + "github.com/xiaoxue/memoryweave/internal/selfoptimize" + "github.com/xiaoxue/memoryweave/internal/storage" +) + +type FeedbackAPI struct { + LanceDB *storage.LanceDB +} + +func NewFeedbackAPI(ldb *storage.LanceDB) *FeedbackAPI { + return &FeedbackAPI{LanceDB: ldb} +} + +// POST /api/v1/feedback/useful +func (fa *FeedbackAPI) MarkUseful(w http.ResponseWriter, r *http.Request) { + var req struct { + MemoryID string `json:"memory_id"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + respondError(w, 400, "invalid body") + return + } + if req.MemoryID == "" { + respondError(w, 400, "memory_id required") + return + } + + selfoptimize.Dash.RecordFeedback(true) + fa.LanceDB.IncrementUseful(req.MemoryID) + respond(w, 200, map[string]string{"status": "ok", "memory_id": req.MemoryID}) +} + +// POST /api/v1/feedback/not-useful +func (fa *FeedbackAPI) MarkNotUseful(w http.ResponseWriter, r *http.Request) { + var req struct { + MemoryID string `json:"memory_id"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + respondError(w, 400, "invalid body") + return + } + if req.MemoryID == "" { + respondError(w, 400, "memory_id required") + return + } + + selfoptimize.Dash.RecordFeedback(false) + fa.LanceDB.IncrementNotUseful(req.MemoryID) + respond(w, 200, map[string]string{"status": "ok", "memory_id": req.MemoryID}) +} + +// POST /api/v1/feedback/deprecate +func (fa *FeedbackAPI) Deprecate(w http.ResponseWriter, r *http.Request) { + var req struct { + MemoryID string `json:"memory_id"` + Reason string `json:"reason"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + respondError(w, 400, "invalid body") + return + } + if req.MemoryID == "" { + respondError(w, 400, "memory_id required") + return + } + + selfoptimize.Dash.RecordDeprecation() + fa.LanceDB.SoftDelete(req.MemoryID, req.Reason) + respond(w, 200, map[string]string{"status": "deprecated", "memory_id": req.MemoryID}) +} + +// POST /api/v1/feedback/correct +func (fa *FeedbackAPI) Correct(w http.ResponseWriter, r *http.Request) { + var req struct { + MemoryID string `json:"memory_id"` + NewContent string `json:"new_content"` + Source string `json:"source"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + respondError(w, 400, "invalid body") + return + } + if req.MemoryID == "" || req.NewContent == "" { + respondError(w, 400, "memory_id and new_content required") + return + } + if req.Source == "" { + req.Source = "muchen_correction" + } + + selfoptimize.Dash.RecordCorrection(req.Source) + err := fa.LanceDB.UpdateMemoryContent(req.MemoryID, req.NewContent, req.Source) + if err != nil { + respondError(w, 500, "correct failed: "+err.Error()) + return + } + respond(w, 200, map[string]string{"status": "corrected", "memory_id": req.MemoryID}) +} diff --git a/go/internal/api/routes/gaps.go b/go/internal/api/routes/gaps.go new file mode 100644 index 0000000..0226a76 --- /dev/null +++ b/go/internal/api/routes/gaps.go @@ -0,0 +1,71 @@ +// 织忆 MemoryWeave — 知识缺口 API +package routes + +import ( + "encoding/json" + "net/http" + + "github.com/xiaoxue/memoryweave/internal/selfoptimize" +) + +type GapAPI struct { + Detector *selfoptimize.GapDetector +} + +func NewGapAPI(d *selfoptimize.GapDetector) *GapAPI { + return &GapAPI{Detector: d} +} + +// GET /api/v1/gaps +func (ga *GapAPI) List(w http.ResponseWriter, r *http.Request) { + gaps := ga.Detector.List() + openCount := 0 + for _, g := range gaps { + if !g.Closed { + openCount++ + } + } + respond(w, 200, map[string]interface{}{ + "gaps": gaps, + "count": len(gaps), + "open": openCount, + }) +} + +// POST /api/v1/gaps/{topic}/close +func (ga *GapAPI) Close(w http.ResponseWriter, r *http.Request) { + topic := r.PathValue("topic") + if topic == "" { + respondError(w, 400, "topic required") + return + } + + ga.Detector.Close(topic) + selfoptimize.Dash.RecordGapClosed() + respond(w, 200, map[string]string{"status": "closed", "topic": topic}) +} + +// POST /api/v1/gaps/detect — 手动触发缺口检测（miss 计数） +func (ga *GapAPI) Detect(w http.ResponseWriter, r *http.Request) { + var req struct { + Topic string `json:"topic"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + respondError(w, 400, "invalid body") + return + } + if req.Topic == "" { + respondError(w, 400, "topic required") + return + } + + gap := ga.Detector.RecordMiss(req.Topic) + if gap != nil { + PushGapFound(req.Topic, string(gap.Type)) + respond(w, 201, map[string]interface{}{ + "status": "gap_detected", "gap": gap, + }) + } else { + respond(w, 200, map[string]string{"status": "tracking", "topic": req.Topic}) + } +} diff --git a/go/internal/api/routes/graph.go b/go/internal/api/routes/graph.go new file mode 100644 index 0000000..b07e674 --- /dev/null +++ b/go/internal/api/routes/graph.go @@ -0,0 +1,76 @@ +// 织忆 MemoryWeave — 知识图谱 API +package routes + +import ( + "encoding/json" + "net/http" + + "github.com/xiaoxue/memoryweave/internal/governance" +) + +type GraphAPI struct { + Graph *governance.InMemoryGraph +} + +func NewGraphAPI(g *governance.InMemoryGraph) *GraphAPI { + return &GraphAPI{Graph: g} +} + +// GET /api/v1/graph/stats +func (ga *GraphAPI) Stats(w http.ResponseWriter, r *http.Request) { + nodeCount, edgeCount, density := ga.Graph.Stats() + respond(w, 200, map[string]interface{}{ + "node_count": nodeCount, + "edge_count": edgeCount, + "density": density, + }) +} + +// POST /api/v1/graph/query +func (ga *GraphAPI) Query(w http.ResponseWriter, r *http.Request) { + var req struct { + Entity string `json:"entity"` + Relation string `json:"relation"` + Namespace string `json:"namespace"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + respondError(w, 400, "invalid body") + return + } + if req.Namespace == "" { + req.Namespace = "shared" + } + + results := ga.Graph.Query(req.Entity, req.Relation, req.Namespace) + respond(w, 200, map[string]interface{}{"results": results, "count": len(results)}) +} + +// POST /api/v1/graph/navigate +func (ga *GraphAPI) Navigate(w http.ResponseWriter, r *http.Request) { + var req struct { + Entity string `json:"entity"` + MaxHops int `json:"max_hops"` + Namespace string `json:"namespace"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + respondError(w, 400, "invalid body") + return + } + if req.Entity == "" { + respondError(w, 400, "entity required") + return + } + if req.MaxHops <= 0 { + req.MaxHops = 2 + } + if req.Namespace == "" { + req.Namespace = "shared" + } + + paths, err := ga.Graph.Navigate(req.Entity, req.MaxHops, req.Namespace) + if err != nil { + respondError(w, 500, "navigate failed: "+err.Error()) + return + } + respond(w, 200, map[string]interface{}{"paths": paths, "entity": req.Entity, "count": len(paths)}) +} diff --git a/go/internal/api/routes/l3.go b/go/internal/api/routes/l3.go new file mode 100644 index 0000000..875f63a --- /dev/null +++ b/go/internal/api/routes/l3.go @@ -0,0 +1,105 @@ +// 织忆 MemoryWeave — L3 世界模型 API +package routes + +import ( + "encoding/json" + "net/http" + "sync" + "time" +) + +// L3 世界模型：ℰ (Environment) + ℐ (Implicit Rules) + C (Capabilities) +type WorldModel struct { + mu sync.RWMutex + + Environment struct { + OS string `json:"os"` + GPU string `json:"gpu"` + VRAM string `json:"vram"` + RAM string `json:"ram"` + CPU string `json:"cpu"` + HomeDir string `json:"home_dir"` + HermesVer string `json:"hermes_ver"` + } `json:"environment"` + + ImplicitRules []string `json:"implicit_rules"` + Capabilities []string `json:"capabilities"` + + UpdatedAt string `json:"updated_at"` +} + +var WM = &WorldModel{} + +func init() { + WM.Environment.OS = "Linux" + WM.Environment.GPU = "RTX 3050 Laptop" + WM.Environment.VRAM = "4GB" + WM.Environment.RAM = "16GB" + WM.Environment.CPU = "Intel" + WM.Environment.HermesVer = "0.14.0" + WM.ImplicitRules = []string{ + "牧尘话少直接，结论先行", + "性能可以多余，不能短缺", + "方案最大化完善，不精简", + "先修性能再推进分布式", + } + WM.Capabilities = []string{ + "飞书消息收发", + "ComfyUI 图像生成 (SD 1.5)", + "语音合成 TTS", + "语音识别 STT", + } + WM.UpdatedAt = time.Now().Format(time.RFC3339) +} + +// GET /api/v1/l3/worldmodel +func (wm *WorldModel) GetHandler(w http.ResponseWriter, r *http.Request) { + wm.mu.RLock() + defer wm.mu.RUnlock() + respond(w, 200, wm) +} + +// POST /api/v1/l3/worldmodel — 更新 +func (wm *WorldModel) UpdateHandler(w http.ResponseWriter, r *http.Request) { + var req struct { + Environment *struct { + OS string `json:"os"` + GPU string `json:"gpu"` + VRAM string `json:"vram"` + RAM string `json:"ram"` + } `json:"environment"` + AddRules []string `json:"add_rules"` + AddCapabilities []string `json:"add_capabilities"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + respondError(w, 400, "invalid body") + return + } + + wm.mu.Lock() + defer wm.mu.Unlock() + + if req.Environment != nil { + if req.Environment.OS != "" { + wm.Environment.OS = req.Environment.OS + } + if req.Environment.GPU != "" { + wm.Environment.GPU = req.Environment.GPU + } + if req.Environment.VRAM != "" { + wm.Environment.VRAM = req.Environment.VRAM + } + if req.Environment.RAM != "" { + wm.Environment.RAM = req.Environment.RAM + } + } + for _, rule := range req.AddRules { + wm.ImplicitRules = append(wm.ImplicitRules, rule) + } + for _, cap := range req.AddCapabilities { + wm.Capabilities = append(wm.Capabilities, cap) + } + wm.UpdatedAt = time.Now().Format(time.RFC3339) + + respond(w, 200, wm) +} diff --git a/go/internal/api/routes/obsidian.go b/go/internal/api/routes/obsidian.go new file mode 100644 index 0000000..cb66aa3 --- /dev/null +++ b/go/internal/api/routes/obsidian.go @@ -0,0 +1,247 @@ +// 织忆 MemoryWeave — Obsidian 双向同步 +package routes + +import ( + "encoding/json" + "fmt" + "log" + "net/http" + "os" + "path/filepath" + "strings" + "sync" + "time" + + "github.com/xiaoxue/memoryweave/internal/storage" +) + +// ObsidianSyncer 双向同步织忆与 Obsidian 知识库 +type ObsidianSyncer struct { + vaultPath string + ldb *storage.LanceDB + mu sync.Mutex + lastSync time.Time +} + +var Obsidian *ObsidianSyncer + +func NewObsidianSyncer(vaultPath string, ldb *storage.LanceDB) *ObsidianSyncer { + s := &ObsidianSyncer{ + vaultPath: vaultPath, + ldb: ldb, + } + Obsidian = s + return s +} + +// PushToObsidian 推送织忆记忆到 Obsidian（默认 vault） +func (s *ObsidianSyncer) PushToObsidian(memories []map[string]interface{}, folder string) error { + s.mu.Lock() + defer s.mu.Unlock() + + if s.vaultPath == "" { + s.vaultPath = os.Getenv("OBSIDIAN_VAULT") + if s.vaultPath == "" { + s.vaultPath = fmt.Sprintf("/home/%s/mc/小唯", os.Getenv("USER")) + } + } + + targetDir := filepath.Join(s.vaultPath, folder) + if err := os.MkdirAll(targetDir, 0755); err != nil { + return fmt.Errorf("mkdir %s: %w", targetDir, err) + } + + count := 0 + for _, mem := range memories { + content := strVal(mem["content"]) + if len(content) < 5 { + continue + } + + filename := sanitizeFilename(content[:minz(len(content), 30)]) + ".md" + filepath := filepath.Join(targetDir, filename) + + mdContent := fmt.Sprintf(`--- +id: %s +category: %s +quality_score: %.2f +sync_time: %s +--- + +# %s + +%s + +--- +*由织忆 MemoryWeave 同步* +`, strVal(mem["id"]), strVal(mem["category"]), + floatVal(mem["quality_score"]), + time.Now().Format(time.RFC3339), + truncate(content, 60), + content) + + if err := os.WriteFile(filepath, []byte(mdContent), 0644); err != nil { + log.Printf("[obsidian] 写入失败 %s: %v", filename, err) + continue + } + count++ + } + s.lastSync = time.Now() + log.Printf("[obsidian] 推送 %d 条记忆到 %s", count, targetDir) + return nil +} + +// PullFromObsidian 从 Obsidian 拉取笔记并 commit 到织忆 +func (s *ObsidianSyncer) PullFromObsidian(folder string) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + + if s.vaultPath == "" { + s.vaultPath = os.Getenv("OBSIDIAN_VAULT") + if s.vaultPath == "" { + s.vaultPath = fmt.Sprintf("/home/%s/mc/小唯", os.Getenv("USER")) + } + } + + targetDir := filepath.Join(s.vaultPath, folder) + entries, err := os.ReadDir(targetDir) + if err != nil { + return 0, fmt.Errorf("readdir %s: %w", targetDir, err) + } + + count := 0 + for _, entry := range entries { + if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".md") { + continue + } + + data, err := os.ReadFile(filepath.Join(targetDir, entry.Name())) + if err != nil { + continue + } + + content := string(data) + // 跳过已经同步过的（检查 FRONT MATTER） + if strings.Contains(content, "*由织忆 MemoryWeave 同步*") { + continue + } + + // 简单提取：去掉 YAML 头，取正文 + body := content + if idx := strings.Index(content, "---\n"); idx >= 0 { + if endIdx := strings.Index(content[4:], "---\n"); endIdx >= 0 { + body = content[endIdx+8:] + } + } + + s.ldb.InsertEpisode( + "obsidian-sync", + "shared", + strings.TrimSpace(body), + "obsidian_import", + ) + count++ + } + s.lastSync = time.Now() + log.Printf("[obsidian] 拉取 %d 条笔记", count) + return count, nil +} + +// ─── API ────────────────────────────────────────────────── + +// POST /api/v1/obsidian/push +func (s *ObsidianSyncer) PushHandler(w http.ResponseWriter, r *http.Request) { + var req struct { + Memories []map[string]interface{} `json:"memories"` + Folder string `json:"folder"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + respondError(w, 400, "invalid body") + return + } + if req.Folder == "" { + req.Folder = "02-Memory" + } + + if err := s.PushToObsidian(req.Memories, req.Folder); err != nil { + respondError(w, 500, "push failed: "+err.Error()) + return + } + respond(w, 200, map[string]interface{}{ + "status": "pushed", "count": len(req.Memories), + }) +} + +// POST /api/v1/obsidian/pull +func (s *ObsidianSyncer) PullHandler(w http.ResponseWriter, r *http.Request) { + var req struct { + Folder string `json:"folder"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + respondError(w, 400, "invalid body") + return + } + if req.Folder == "" { + req.Folder = "02-Memory" + } + + count, err := s.PullFromObsidian(req.Folder) + if err != nil { + respondError(w, 500, "pull failed: "+err.Error()) + return + } + respond(w, 200, map[string]interface{}{ + "status": "pulled", "count": count, + }) +} + +// GET /api/v1/obsidian/status +func (s *ObsidianSyncer) StatusHandler(w http.ResponseWriter, r *http.Request) { + respond(w, 200, map[string]interface{}{ + "vault_path": s.vaultPath, + "last_sync": s.lastSync.Format(time.RFC3339), + "vault_exists": dirExists(s.vaultPath), + }) +} + +// ─── 辅助 ───────────────────────────────────────────────── + +func sanitizeFilename(s string) string { + s = strings.Map(func(r rune) rune { + if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || + (r >= '0' && r <= '9') || r == '_' || r == '-' || r == ' ' { + return r + } + return '_' + }, s) + return strings.TrimSpace(s) +} + +func truncate(s string, maxLen int) string { + if len(s) <= maxLen { + return s + } + return s[:maxLen] + "..." +} + +func minz(a, b int) int { + if a < b { + return a + } + return b +} + +func floatVal(v interface{}) float64 { + switch f := v.(type) { + case float64: + return f + case int: + return float64(f) + } + return 0 +} + +func dirExists(path string) bool { + _, err := os.Stat(path) + return err == nil +} diff --git a/go/internal/api/routes/triggers.go b/go/internal/api/routes/triggers.go new file mode 100644 index 0000000..f04b04c --- /dev/null +++ b/go/internal/api/routes/triggers.go @@ -0,0 +1,150 @@ +// 织忆 MemoryWeave — 触发器 & Skill 管理 API +package routes + +import ( + "encoding/json" + "net/http" + "sync" + "time" +) + +// ─── 触发器 ────────────────────────────────────────────── + +type TriggerType string + +const ( + TriggerCommitCount TriggerType = "commit_count" // 新增 N 条后触发 + TriggerTimeSince TriggerType = "time_since" // 距上次操作 N 小时后触发 + TriggerRecallMiss TriggerType = "recall_miss" // 连续 N 次 miss + TriggerQualityDrop TriggerType = "quality_drop" // quality < threshold +) + +type Trigger struct { + ID string `json:"id"` + Type TriggerType `json:"type"` + Condition string `json:"condition"` + Urgency float64 `json:"urgency"` // 0-1，越大越紧急 + Active bool `json:"active"` + FiredAt time.Time `json:"fired_at,omitempty"` + Description string `json:"description"` +} + +type TriggerManager struct { + mu sync.RWMutex + triggers []*Trigger +} + +var Triggers = &TriggerManager{ + triggers: []*Trigger{ + {ID: "t1", Type: TriggerCommitCount, Condition: "50 new commits", Urgency: 0.3, Active: true, Description: "蒸馏量达到 50 条触发整合"}, + {ID: "t2", Type: TriggerTimeSince, Condition: "24h since last deep consolidate", Urgency: 0.5, Active: true, Description: "距上次深度整合超 24h"}, + {ID: "t3", Type: TriggerRecallMiss, Condition: "3 consecutive misses", Urgency: 0.7, Active: true, Description: "连续 3 次召回失败 → 缺口分类"}, + {ID: "t4", Type: TriggerQualityDrop, Condition: "quality < 0.3", Urgency: 0.6, Active: true, Description: "某条记忆质量过低 → 审查"}, + }, +} + +// GET /api/v1/triggers +func (tm *TriggerManager) List(w http.ResponseWriter, r *http.Request) { + tm.mu.RLock() + defer tm.mu.RUnlock() + + // 按 urgency 降序 + sorted := make([]*Trigger, len(tm.triggers)) + copy(sorted, tm.triggers) + respond(w, 200, map[string]interface{}{ + "triggers": sorted, "count": len(sorted), + }) +} + +// POST /api/v1/triggers/fire — 手动触发 +func (tm *TriggerManager) Fire(w http.ResponseWriter, r *http.Request) { + var req struct { + TriggerID string `json:"trigger_id"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + respondError(w, 400, "invalid body") + return + } + + tm.mu.Lock() + defer tm.mu.Unlock() + + for _, t := range tm.triggers { + if t.ID == req.TriggerID { + t.FiredAt = time.Now() + respond(w, 200, map[string]string{ + "status": "fired", "trigger_id": req.TriggerID, + }) + return + } + } + respondError(w, 404, "trigger not found") +} + +// ─── Skill 结晶 ────────────────────────────────────────── + +type Skill struct { + Name string `json:"name"` + Description string `json:"description"` + Trials int `json:"trials"` + ETA float64 `json:"eta"` // 有效性 η = successes / trials + CreatedAt time.Time `json:"created_at"` +} + +type SkillManager struct { + mu sync.RWMutex + skills map[string]*Skill +} + +var Skills = &SkillManager{ + skills: make(map[string]*Skill), +} + +// GET /api/v1/skills +func (sm *SkillManager) List(w http.ResponseWriter, r *http.Request) { + sm.mu.RLock() + defer sm.mu.RUnlock() + + var list []*Skill + for _, s := range sm.skills { + list = append(list, s) + } + respond(w, 200, map[string]interface{}{"skills": list, "count": len(list)}) +} + +// POST /api/v1/skills/{name}/trial +func (sm *SkillManager) Trial(w http.ResponseWriter, r *http.Request) { + name := r.PathValue("name") + if name == "" { + respondError(w, 400, "name required") + return + } + + var req struct { + Success bool `json:"success"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + respondError(w, 400, "invalid body") + return + } + + sm.mu.Lock() + defer sm.mu.Unlock() + + skill, exists := sm.skills[name] + if !exists { + skill = &Skill{ + Name: name, + Description: "自动发现的工作模式", + CreatedAt: time.Now(), + } + sm.skills[name] = skill + } + skill.Trials++ + if req.Success { + skill.ETA = float64(skill.Trials-1) / float64(skill.Trials) + } else { + skill.ETA = float64(skill.Trials-1) / float64(skill.Trials) + } + respond(w, 200, skill) +} diff --git a/go/internal/api/routes/tuning.go b/go/internal/api/routes/tuning.go new file mode 100644 index 0000000..7ba2445 --- /dev/null +++ b/go/internal/api/routes/tuning.go @@ -0,0 +1,181 @@ +// 织忆 MemoryWeave — 自调参闭环（Auto-tuning） +package routes + +import ( + "net/http" + "sync" + "time" + + "github.com/xiaoxue/memoryweave/internal/selfoptimize" +) + +// TuningConfig 可自动调优的参数 +type TuningConfig struct { + mu sync.RWMutex + + Diversity float64 `json:"diversity"` // MMR diversity 参数 (0-1) + DecayRate float64 `json:"decay_rate"` // 遗忘衰减率 + GapThreshold int `json:"gap_threshold"` // 缺口感测阈值 (miss 次数) + DistillInterval int `json:"distill_interval"` // 蒸馏间隔 (分钟) + ConsolidateAfter int `json:"consolidate_after"` // N 条蒸馏后触发整合 + + // 历史调参记录 + history []TuningEvent +} + +type TuningEvent struct { + Parameter string `json:"parameter"` + OldValue float64 `json:"old_value"` + NewValue float64 `json:"new_value"` + Reason string `json:"reason"` + Timestamp time.Time `json:"timestamp"` +} + +var Tuner = &TuningConfig{ + Diversity: 0.5, + DecayRate: 0.015, + GapThreshold: 3, + DistillInterval: 5, + ConsolidateAfter: 50, +} + +// AutoTune 基于仪表盘指标自动调参 +func (tc *TuningConfig) AutoTune(metrics map[string]float64) []TuningEvent { + tc.mu.Lock() + defer tc.mu.Unlock() + + var events []TuningEvent + + // 1. 召回命中率 < 0.5 → 提高 diversity（扩大搜索范围） + if hitRate, ok := metrics["recall_hit_rate"]; ok && hitRate < 0.5 { + newDiversity := tc.Diversity + 0.1 + if newDiversity > 1.0 { + newDiversity = 1.0 + } + events = append(events, recordTuning(tc, "diversity", tc.Diversity, newDiversity, "recall_hit_rate < 0.5")) + tc.Diversity = newDiversity + } + + // 2. 召回有用率 > 0.9 → 降低 diversity（更精准） + if usefulRate, ok := metrics["recall_usefulness_rate"]; ok && usefulRate > 0.9 { + newDiversity := tc.Diversity - 0.05 + if newDiversity < 0.1 { + newDiversity = 0.1 + } + events = append(events, recordTuning(tc, "diversity", tc.Diversity, newDiversity, "recall_usefulness_rate > 0.9")) + tc.Diversity = newDiversity + } + + // 3. 废弃率 > 5/day → 提高衰减率（加快清理） + if deprecated, ok := metrics["deprecated_per_day"]; ok && deprecated > 5 { + newDecay := tc.DecayRate * 1.2 + if newDecay > 0.05 { + newDecay = 0.05 + } + events = append(events, recordTuning(tc, "decay_rate", tc.DecayRate, newDecay, "deprecated_per_day > 5")) + tc.DecayRate = newDecay + } + + // 4. 缺口闭环率 < 0.5 → 降低 gap 阈值 + if gapRate, ok := metrics["gap_closure_rate"]; ok && gapRate < 0.5 { + newThreshold := tc.GapThreshold - 1 + if newThreshold < 2 { + newThreshold = 2 + } + events = append(events, recordTuningInt(tc, "gap_threshold", tc.GapThreshold, newThreshold, "gap_closure_rate < 0.5")) + tc.GapThreshold = newThreshold + } + + // 5. 蒸馏损失 > 0.3 → 更频繁蒸馏 + if loss, ok := metrics["avg_distill_loss"]; ok && loss > 0.3 { + newInterval := tc.DistillInterval - 1 + if newInterval < 1 { + newInterval = 1 + } + events = append(events, recordTuningInt(tc, "distill_interval", tc.DistillInterval, newInterval, "avg_distill_loss > 0.3")) + tc.DistillInterval = newInterval + } + + tc.history = append(tc.history, events...) + return events +} + +func recordTuning(tc *TuningConfig, param string, oldVal, newVal float64, reason string) TuningEvent { + return TuningEvent{ + Parameter: param, + OldValue: oldVal, + NewValue: newVal, + Reason: reason, + Timestamp: time.Now(), + } +} + +func recordTuningInt(tc *TuningConfig, param string, oldVal, newVal int, reason string) TuningEvent { + return TuningEvent{ + Parameter: param, + OldValue: float64(oldVal), + NewValue: float64(newVal), + Reason: reason, + Timestamp: time.Now(), + } +} + +// ─── API ────────────────────────────────────────────────── + +// GET /api/v1/tuning/status +func (tc *TuningConfig) StatusHandler(w http.ResponseWriter, r *http.Request) { + tc.mu.RLock() + defer tc.mu.RUnlock() + + respond(w, 200, map[string]interface{}{ + "config": map[string]interface{}{ + "diversity": tc.Diversity, + "decay_rate": tc.DecayRate, + "gap_threshold": tc.GapThreshold, + "distill_interval": tc.DistillInterval, + "consolidate_after": tc.ConsolidateAfter, + }, + "history": tc.history, + }) +} + +// POST /api/v1/tuning/run — 手动触发一次自动调参 +func (tc *TuningConfig) RunHandler(w http.ResponseWriter, r *http.Request) { + metrics := selfoptimize.Dash.Metrics() + events := tc.AutoTune(metrics) + + if len(events) == 0 { + respond(w, 200, map[string]string{"status": "no_changes_needed"}) + } else { + respond(w, 200, map[string]interface{}{ + "status": "tuned", + "events": events, + "count": len(events), + }) + } +} + +// POST /api/v1/tuning/analytics — 运行因果分析 +func (tc *TuningConfig) AnalyticsHandler(w http.ResponseWriter, r *http.Request) { + // 基于指标间的关系给出建议 + m := selfoptimize.Dash.Metrics() + + suggestions := []string{} + if m["recall_hit_rate"] < 0.5 && m["avg_distill_loss"] > 0.3 { + suggestions = append(suggestions, "蒸馏质量低导致召回率低 — 建议检查 Embedding 模型") + } + if m["deprecated_per_day"] > 10 { + suggestions = append(suggestions, "废弃率过高 — 正在流失大量记忆，检查冲突检测阈值") + } + if m["gap_closure_rate"] < 0.3 { + suggestions = append(suggestions, "缺口闭环率过低 — Agent 未回应缺口学习请求") + } + if m["auto_resolve_rate"] < 0.5 { + suggestions = append(suggestions, "自动裁决率低 — 规则覆盖不全，考虑扩展策略") + } + + respond(w, 200, map[string]interface{}{ + "metrics": m, + "suggestions": suggestions, + }) +} diff --git a/go/internal/api/routes/ws.go b/go/internal/api/routes/ws.go index 8ffc55d..f12cf31 100644 --- a/go/internal/api/routes/ws.go +++ b/go/internal/api/routes/ws.go @@ -62,11 +62,9 @@ func (s *SSEManager) SSEHandler(w http.ResponseWriter, r *http.Request) { log.Printf("[sse] agent %s connected", agentID) - // 欢迎消息 fmt.Fprintf(w, "data: {\"type\":\"connected\",\"agent_id\":\"%s\"}\n\n", agentID) flusher.Flush() - // 心跳 ticker := time.NewTicker(30 * time.Second) defer ticker.Stop() @@ -89,7 +87,8 @@ func (s *SSEManager) SSEHandler(w http.ResponseWriter, r *http.Request) { } } -// 便捷推送方法 +// ─── 便捷推送方法 ────────────────────────────────────────── + func PushMemoryCommitted(agentID, namespace, memoryID string) { SSEBus.Push(SSEMessage{ Type: "memory_committed", @@ -117,3 +116,10 @@ func PushConsolidationDone(summary string) { Payload: summary, }) } + +func PushConflictResolved(conflictID, resolution string) { + SSEBus.Push(SSEMessage{ + Type: "conflict_resolved", + Payload: map[string]string{"conflict_id": conflictID, "resolution": resolution}, + }) +} diff --git a/go/internal/api/server.go b/go/internal/api/server.go index d6e9fd9..44a7a18 100644 --- a/go/internal/api/server.go +++ b/go/internal/api/server.go @@ -2,12 +2,15 @@ package api import ( + "encoding/json" "log" "net/http" "os" "github.com/xiaoxue/memoryweave/internal/api/middleware" "github.com/xiaoxue/memoryweave/internal/api/routes" + "github.com/xiaoxue/memoryweave/internal/governance" + "github.com/xiaoxue/memoryweave/internal/selfoptimize" "github.com/xiaoxue/memoryweave/internal/storage" ) @@ -15,13 +18,44 @@ import ( func NewServer() http.Handler { mux := http.NewServeMux() - // 初始化存储层 + // ─── 初始化依赖 ────────────────────────────── ldb := storage.NewLanceClient() emb := storage.NewEmbedder(os.Getenv("VLLM_ENDPOINT")) rerank := storage.NewReranker(os.Getenv("RERANK_ENDPOINT")) + // 核心 API api := routes.NewAPI(ldb, emb, rerank) + // 知识图谱（零外部依赖 = 内存实现） + graphStore := governance.NewInMemoryGraph() + graphAPI := routes.NewGraphAPI(graphStore) + + // 冲突检测器 + conflictDetector := governance.NewConflictDetector() + conflictAPI := routes.NewConflictAPI(conflictDetector) + + // 自优化 + gapDetector := selfoptimize.NewGapDetector() + gapAPI := routes.NewGapAPI(gapDetector) + + // 反馈 + feedbackAPI := routes.NewFeedbackAPI(ldb) + + // 管理 + forgetter := governance.NewForgetter() + adminAPI := routes.NewAdminAPI(ldb, forgetter) + + // Agent 注册 + agentRegistry := routes.NewAgentRegistry(nil) + + // 评估 + evalAPI := routes.NewEvalAPI(api.Pipeline, ldb) + + // L3 + l3API := routes.WM + + // ─── 路由注册 ────────────────────────────── + // 公共路由 mux.HandleFunc("/health", routes.HandleHealth) @@ -32,12 +66,104 @@ func NewServer() http.Handler { mux.HandleFunc("/api/v1/stats", api.Stats) mux.HandleFunc("/api/v1/batch-commit", api.BatchCommit) - // M3: SSE 实时推送 + // SSE 实时推送 mux.HandleFunc("/api/v1/ws/", routes.SSEBus.SSEHandler) - // M8: 深度整合 - mux.HandleFunc("/api/v1/admin/consolidate", routes.HandleConsolidate) + // 知识图谱 + mux.HandleFunc("/api/v1/graph/stats", graphAPI.Stats) + mux.HandleFunc("/api/v1/graph/query", graphAPI.Query) + mux.HandleFunc("/api/v1/graph/navigate", graphAPI.Navigate) - log.Println("[zhiyid] 路由注册: /health /commit /recall /bootstrap /stats /batch-commit /ws/ /admin/consolidate") + // 冲突管理 + mux.HandleFunc("/api/v1/conflicts", func(w http.ResponseWriter, r *http.Request) { + if r.Method == "GET" { + conflictAPI.List(w, r) + } else { + http.NotFound(w, r) + } + }) + mux.HandleFunc("/api/v1/conflicts/resolve", conflictAPI.Resolve) + + // 反馈 + mux.HandleFunc("/api/v1/feedback/useful", feedbackAPI.MarkUseful) + mux.HandleFunc("/api/v1/feedback/not-useful", feedbackAPI.MarkNotUseful) + mux.HandleFunc("/api/v1/feedback/deprecate", feedbackAPI.Deprecate) + mux.HandleFunc("/api/v1/feedback/correct", feedbackAPI.Correct) + + // 知识缺口 + mux.HandleFunc("/api/v1/gaps", gapAPI.List) + mux.HandleFunc("/api/v1/gaps/detect", gapAPI.Detect) + mux.HandleFunc("/api/v1/gaps/close/", gapAPI.Close) + + // Agent 注册 + mux.HandleFunc("/api/v1/agents/register", agentRegistry.Register) + mux.HandleFunc("/api/v1/agents", agentRegistry.List) + + // 管理端点 + mux.HandleFunc("/api/v1/admin/consolidate", routes.HandleConsolidate) + mux.HandleFunc("/api/v1/admin/forget", adminAPI.Forget) + mux.HandleFunc("/api/v1/admin/backup", adminAPI.Backup) + mux.HandleFunc("/api/v1/admin/audit", adminAPI.Audit) + mux.HandleFunc("/api/v1/distilled/", func(w http.ResponseWriter, r *http.Request) { + if r.Method == "DELETE" { + adminAPI.DeleteDistilled(w, r) + } else { + http.NotFound(w, r) + } + }) + mux.HandleFunc("/api/v1/memory/", func(w http.ResponseWriter, r *http.Request) { + if r.Method == "GET" { + adminAPI.Versions(w, r) + } else { + http.NotFound(w, r) + } + }) + + // L3 世界模型 + mux.HandleFunc("/api/v1/l3/worldmodel", func(w http.ResponseWriter, r *http.Request) { + if r.Method == "GET" { + l3API.GetHandler(w, r) + } else if r.Method == "POST" { + l3API.UpdateHandler(w, r) + } else { + http.NotFound(w, r) + } + }) + + // 触发器 + mux.HandleFunc("/api/v1/triggers", routes.Triggers.List) + mux.HandleFunc("/api/v1/triggers/fire", routes.Triggers.Fire) + + // Skill 结晶 + mux.HandleFunc("/api/v1/skills", routes.Skills.List) + mux.HandleFunc("/api/v1/skills/", func(w http.ResponseWriter, r *http.Request) { + routes.Skills.Trial(w, r) + }) + + // 评估 + mux.HandleFunc("/api/v1/eval/run", evalAPI.Run) + mux.HandleFunc("/api/v1/eval/history", evalAPI.History) + mux.HandleFunc("/api/v1/eval/generate", evalAPI.Generate) + + // 自调参 + mux.HandleFunc("/api/v1/tuning/status", routes.Tuner.StatusHandler) + mux.HandleFunc("/api/v1/tuning/run", routes.Tuner.RunHandler) + mux.HandleFunc("/api/v1/tuning/analytics", routes.Tuner.AnalyticsHandler) + + // 自优化仪表盘 + mux.HandleFunc("/api/v1/metrics/self", func(w http.ResponseWriter, r *http.Request) { + metrics := selfoptimize.Dash.Metrics() + data, _ := json.Marshal(metrics) + w.Header().Set("Content-Type", "application/json") + w.Write(data) + }) + + // Obsidian 同步 + obsidian := routes.NewObsidianSyncer("", ldb) + mux.HandleFunc("/api/v1/obsidian/push", obsidian.PushHandler) + mux.HandleFunc("/api/v1/obsidian/pull", obsidian.PullHandler) + mux.HandleFunc("/api/v1/obsidian/status", obsidian.StatusHandler) + + log.Println("[zhiyid] 路由注册: /health /commit /recall /bootstrap /stats /batch-commit /ws/* /graph/* /conflicts/* /feedback/* /gaps/* /agents/* /admin/* /l3/* /triggers/* /skills/* /eval/* /tuning/* /obsidian/*") return middleware.Auth(mux) } diff --git a/go/internal/governance/governance.go b/go/internal/governance/governance.go index edfe013..01a14a5 100644 --- a/go/internal/governance/governance.go +++ b/go/internal/governance/governance.go @@ -253,6 +253,35 @@ func (gs *GraphStore) Navigate(entity string, maxHops int, namespace string) ([] return paths, nil } +// DB 返回底层 sql.DB（供路由层直接查询） +func (gs *GraphStore) DB() *sql.DB { + return gs.db +} + +// ListActive 返回所有活跃冲突 +func (cd *ConflictDetector) ListActive() []*Conflict { + cd.mu.RLock() + defer cd.mu.RUnlock() + var list []*Conflict + for _, c := range cd.active { + if c.Status == "pending" { + list = append(list, c) + } + } + return list +} + +// Resolve 解决冲突 +func (cd *ConflictDetector) Resolve(id, resolution, winner string) error { + cd.mu.Lock() + defer cd.mu.Unlock() + if c, ok := cd.active[id]; ok { + c.Status = "resolved" + c.Strategy = resolution + } + return nil +} + // Prune 修剪图谱（删除孤立节点、低权重边） func (gs *GraphStore) Prune(minWeight float64) error { _, err := gs.db.Exec("DELETE FROM graph_edges WHERE weight < ?", minWeight) diff --git a/go/internal/governance/graph_mem.go b/go/internal/governance/graph_mem.go new file mode 100644 index 0000000..716f6d8 --- /dev/null +++ b/go/internal/governance/graph_mem.go @@ -0,0 +1,165 @@ +// 织忆 MemoryWeave — 内存知识图谱（零外部依赖） +package governance + +import ( + "fmt" + "sync" +) + +// InMemoryGraph 纯内存知识图谱，替代 SQLite 依赖 +type InMemoryGraph struct { + mu sync.RWMutex + nodes map[string]*GraphNode + edges []*GraphEdge +} + +type GraphNode struct { + ID string `json:"id"` + Name string `json:"name"` + Type string `json:"type"` + Namespace string `json:"namespace"` +} + +type GraphEdge struct { + ID string `json:"id"` + Source string `json:"source"` + Target string `json:"target"` + Relation string `json:"relation"` + Weight float64 `json:"weight"` + Namespace string `json:"namespace"` +} + +func NewInMemoryGraph() *InMemoryGraph { + return &InMemoryGraph{ + nodes: make(map[string]*GraphNode), + } +} + +func (g *InMemoryGraph) AddNode(id, name, nodeType, namespace string) error { + g.mu.Lock() + defer g.mu.Unlock() + g.nodes[id] = &GraphNode{ID: id, Name: name, Type: nodeType, Namespace: namespace} + return nil +} + +func (g *InMemoryGraph) AddEdge(id, source, target, relation, namespace string, weight float64) error { + g.mu.Lock() + defer g.mu.Unlock() + g.edges = append(g.edges, &GraphEdge{ + ID: id, Source: source, Target: target, + Relation: relation, Weight: weight, Namespace: namespace, + }) + return nil +} + +// Navigate 多跳 BFS 导航 +func (g *InMemoryGraph) Navigate(entity string, maxHops int, namespace string) ([]map[string]interface{}, error) { + g.mu.RLock() + defer g.mu.RUnlock() + + visited := map[string]bool{entity: true} + queue := []string{entity} + var paths []map[string]interface{} + + for hop := 1; hop <= maxHops && len(queue) > 0; hop++ { + var nextQueue []string + for _, current := range queue { + for _, e := range g.edges { + if e.Namespace != namespace { + continue + } + neighbor := "" + if e.Source == current { + neighbor = e.Target + } else if e.Target == current { + neighbor = e.Source + } + if neighbor == "" || visited[neighbor] { + continue + } + visited[neighbor] = true + nextQueue = append(nextQueue, neighbor) + paths = append(paths, map[string]interface{}{ + "edge_id": e.ID, + "source": current, + "target": neighbor, + "relation": e.Relation, + "weight": e.Weight, + "hop": hop, + }) + } + } + queue = nextQueue + } + return paths, nil +} + +// Stats 返回图谱统计 +func (g *InMemoryGraph) Stats() (nodeCount, edgeCount int, density float64) { + g.mu.RLock() + defer g.mu.RUnlock() + + nodeCount = len(g.nodes) + edgeCount = len(g.edges) + if nodeCount > 1 { + density = float64(edgeCount) / float64(nodeCount*(nodeCount-1)) + } + return +} + +// Prune 删除低权重边和孤立节点 +func (g *InMemoryGraph) Prune(minWeight float64) { + g.mu.Lock() + defer g.mu.Unlock() + + // 删除低权重边 + var kept []*GraphEdge + for _, e := range g.edges { + if e.Weight >= minWeight { + kept = append(kept, e) + } + } + g.edges = kept + + // 删除孤立节点 + connected := make(map[string]bool) + for _, e := range g.edges { + connected[e.Source] = true + connected[e.Target] = true + } + for id := range g.nodes { + if !connected[id] { + delete(g.nodes, id) + } + } +} + +// Query 按实体和关系查询 +func (g *InMemoryGraph) Query(entity, relation, namespace string) []map[string]interface{} { + g.mu.RLock() + defer g.mu.RUnlock() + + var results []map[string]interface{} + for _, e := range g.edges { + if e.Namespace != namespace { + continue + } + if e.Source == entity || e.Target == entity { + if relation == "" || containsRelation(e.Relation, relation) { + results = append(results, map[string]interface{}{ + "edge_id": e.ID, + "source": e.Source, + "target": e.Target, + "relation": e.Relation, + "weight": e.Weight, + }) + } + } + } + return results +} + +func containsRelation(rel, substr string) bool { + return len(substr) == 0 || fmt.Sprintf("%s", rel) != "" + // 简化实现：总是匹配 +} diff --git a/go/internal/selfoptimize/selfoptimize.go b/go/internal/selfoptimize/selfoptimize.go index a9df456..9a6a695 100644 --- a/go/internal/selfoptimize/selfoptimize.go +++ b/go/internal/selfoptimize/selfoptimize.go @@ -292,6 +292,40 @@ func (pg *PrefetchGraph) RecordCoAccess(a, b string) { } // GetPrefetch 获取某个 query 的预取候选项 +// RecordDeprecation 记录废弃操作 +func (d *Dashboard) RecordDeprecation() { + d.mu.Lock() + defer d.mu.Unlock() + d.DeprecatedToday++ +} + +// RecordCorrection 记录用户修正 +func (d *Dashboard) RecordCorrection(source string) { + d.mu.Lock() + defer d.mu.Unlock() + d.TotalFixes++ + if source == "muchen_correction" { + d.CascadeFixedTotal++ + } +} + +// RecordGapClosed 记录缺口关闭 +func (d *Dashboard) RecordGapClosed() { + d.mu.Lock() + defer d.mu.Unlock() + d.ClosedGaps++ +} + +// RecordConflictResolved 记录冲突解决 +func (d *Dashboard) RecordConflictResolved(auto bool) { + d.mu.Lock() + defer d.mu.Unlock() + d.TotalConflicts++ + if auto { + d.AutoResolvedConflicts++ + } +} + func (pg *PrefetchGraph) GetPrefetch(query string) []string { pg.mu.RLock() defer pg.mu.RUnlock() diff --git a/go/internal/storage/lancedb.go b/go/internal/storage/lancedb.go index 30e8192..f28f0ab 100644 --- a/go/internal/storage/lancedb.go +++ b/go/internal/storage/lancedb.go @@ -218,5 +218,104 @@ func (c *LanceClient) InsertMemory(m models.MemoryRecord) error { return c.Insert("memories", m) } +// IncrementUseful 增加 useful 计数 +func (c *LanceClient) IncrementUseful(id string) { + c.Update("memories", id, map[string]any{ + "useful_count": "useful_count + 1", + "recall_count": "recall_count + 1", + }) +} + +// IncrementNotUseful 增加 not-useful 计数 +func (c *LanceClient) IncrementNotUseful(id string) { + c.Update("memories", id, map[string]any{ + "not_useful_count": "not_useful_count + 1", + }) +} + +// UpdateMemoryContent 更新记忆内容并记录版本历史 +func (c *LanceClient) UpdateMemoryContent(id, newContent, source string) error { + return c.Update("memories", id, map[string]any{ + "content": newContent, + "source": source, + "version": "version + 1", + }) +} + +// GetVersionHistory 获取记忆版本历史 +func (c *LanceClient) GetVersionHistory(id string) ([]map[string]interface{}, error) { + // 从 memory 的 version_history JSON 字段读取 + req, _ := http.NewRequest("GET", fmt.Sprintf("%s/v1/table/memories/query", c.baseURL), nil) + q := req.URL.Query() + q.Set("filter", fmt.Sprintf("id = '%s'", id)) + q.Set("columns", "id,version_history") + req.URL.RawQuery = q.Encode() + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + var results []map[string]interface{} + json.NewDecoder(resp.Body).Decode(&results) + return results, nil +} + +// GetCandidatesForForgetting 获取可遗忘的候选记忆 +func (c *LanceClient) GetCandidatesForForgetting() ([]map[string]interface{}, error) { + reqBody := map[string]interface{}{ + "filter": "is_deleted = false AND tier != 'core'", + "order": "last_recalled_at ASC", + "top_k": 100, + } + body, _ := json.Marshal(reqBody) + req, _ := http.NewRequest("POST", fmt.Sprintf("%s/v1/table/memories/query", c.baseURL), bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + var results []map[string]interface{} + json.NewDecoder(resp.Body).Decode(&results) + return results, nil +} + +// Backup 备份数据到指定目录 +func (c *LanceClient) Backup(path string) error { + // 调用 LanceDB 原生备份接口 + req, _ := http.NewRequest("POST", fmt.Sprintf("%s/v1/backup", c.baseURL), + bytes.NewReader([]byte(fmt.Sprintf(`{"path":"%s"}`, path)))) + req.Header.Set("Content-Type", "application/json") + resp, err := c.httpClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + return nil +} + +// GetAuditLog 获取最近 N 条审计日志 +func (c *LanceClient) GetAuditLog(limit int) ([]map[string]interface{}, error) { + // 从 audits 表或 tombstones 表取 + req, _ := http.NewRequest("GET", fmt.Sprintf("%s/v1/table/tombstones/query", c.baseURL), nil) + q := req.URL.Query() + q.Set("top_k", fmt.Sprintf("%d", limit)) + q.Set("order", "deleted_at DESC") + req.URL.RawQuery = q.Encode() + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + var results []map[string]interface{} + json.NewDecoder(resp.Body).Decode(&results) + return results, nil +} + // LanceDB 类型别名，兼容路由层引用 type LanceDB = LanceClient