628 lines
22 KiB
Go
628 lines
22 KiB
Go
// HTTP 服务器 — 多 Agent 架构（FileGraph + NetworkEventBus + 跨Agent缓存失效）
|
||
package api
|
||
|
||
import (
|
||
"bytes"
|
||
"encoding/json"
|
||
"io"
|
||
"log"
|
||
"net/http"
|
||
"os"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/xiaoxue/memoryweave/internal/api/middleware"
|
||
"github.com/xiaoxue/memoryweave/internal/api/routes"
|
||
"github.com/xiaoxue/memoryweave/internal/distill"
|
||
"github.com/xiaoxue/memoryweave/internal/governance"
|
||
"github.com/xiaoxue/memoryweave/internal/metrics"
|
||
"github.com/xiaoxue/memoryweave/internal/selfoptimize"
|
||
"github.com/xiaoxue/memoryweave/internal/storage"
|
||
)
|
||
|
||
func NewServer() http.Handler {
|
||
mux := http.NewServeMux()
|
||
|
||
// ─── 初始化依赖 ──────────────────────────────
|
||
emb := storage.NewEmbedder(os.Getenv("VLLM_ENDPOINT"))
|
||
rerank := storage.NewReranker(os.Getenv("RERANK_ENDPOINT"))
|
||
|
||
// 存储后端选择：LanceDB (Rust IPC) → SQLite（CGO）→ 内存
|
||
var ldb storage.LanceDB
|
||
backend := os.Getenv("STORAGE_BACKEND")
|
||
switch backend {
|
||
case "lancedb":
|
||
sockPath := os.Getenv("LANCEDB_SOCKET")
|
||
if sockPath == "" {
|
||
sockPath = "/tmp/zhiyi-ipc.sock"
|
||
}
|
||
ldb = storage.NewRustLanceDBClient(sockPath, emb)
|
||
log.Printf("[zhiyid] 存储后端: LanceDB (Rust IPC) — %s", sockPath)
|
||
case "sqlite":
|
||
dbPath := os.Getenv("SQLITE_PATH")
|
||
sqliteDB, err := storage.NewSQLiteClient(dbPath)
|
||
if err != nil {
|
||
log.Printf("[zhiyid] SQLite 初始化失败 (%v)，降级为内存存储", err)
|
||
ldb = storage.NewMemLanceClient(emb)
|
||
} else {
|
||
ldb = sqliteDB
|
||
log.Printf("[zhiyid] 存储后端: SQLite (CGO) — %s", dbPath)
|
||
}
|
||
default:
|
||
ldb = storage.NewMemLanceClient(emb)
|
||
log.Printf("[zhiyid] 存储后端: 内存（零依赖）")
|
||
}
|
||
api := routes.NewAPI(ldb, emb, rerank)
|
||
|
||
// 图谱：SQLite (graph_nodes/graph_edges) — 设计要求，非 FileGraph JSON
|
||
graphPath := os.Getenv("GRAPH_PATH")
|
||
if graphPath == "" {
|
||
graphPath = "/var/lib/memoryweave/graph.db"
|
||
}
|
||
var graphStore governance.GraphStore
|
||
gs, err := governance.NewSQLiteGraphStore(graphPath)
|
||
if err != nil {
|
||
log.Printf("[zhiyid] WARN: SQLite 图谱初始化失败 (%v)，降级为 InMemoryGraph", err)
|
||
graphStore = governance.NewInMemoryGraph()
|
||
} else {
|
||
graphStore = gs
|
||
log.Printf("[zhiyid] 图谱后端: SQLite — %s", graphPath)
|
||
}
|
||
graphUpdater := governance.NewAutoGraphUpdater(graphStore)
|
||
graphAPI := routes.NewGraphAPI(graphStore)
|
||
|
||
// G1: Recall 管线挂图谱扩展
|
||
api.Pipeline.SetGraphExpander(graphStore)
|
||
|
||
// G4: 自动蒸馏 → 注入图谱更新器
|
||
routes.SetGraphUpdater(graphUpdater)
|
||
|
||
// 冲突
|
||
conflictDetector := governance.NewConflictDetector()
|
||
conflictAPI := routes.NewConflictAPI(conflictDetector)
|
||
|
||
// 缺口
|
||
gapDetector := selfoptimize.NewGapDetector(emb, ldb)
|
||
gapAPI := routes.NewGapAPI(gapDetector)
|
||
routes.InitGapRepair(gapDetector) // 共享同一个 GapDetector
|
||
|
||
feedbackAPI := routes.NewFeedbackAPI(ldb)
|
||
// 遗忘器：支持按 Agent 类型设置衰减率
|
||
forgetter := governance.NewForgetter()
|
||
adminAPI := routes.NewAdminAPI(ldb, forgetter)
|
||
agentRegistry := routes.NewAgentRegistry(nil)
|
||
evalAPI := routes.NewEvalAPI(api.Pipeline, ldb)
|
||
l3API := routes.WM
|
||
consolPipe := routes.NewConsolidationPipeline(ldb, graphStore, conflictDetector)
|
||
consolPipe.SetDataDir("/var/lib/memoryweave/lancedb", "/var/lib/memoryweave/graph.db")
|
||
|
||
// ─── 蒸馏引擎 ───────────────────────────────
|
||
llmEndpoint := os.Getenv("LLM_ENDPOINT")
|
||
llmModel := os.Getenv("LLM_MODEL")
|
||
if llmModel == "" { llmModel = "deepseek/deepseek-v4-pro" }
|
||
distillEngine := distill.NewEngine(llmEndpoint, llmModel, os.Getenv("API_KEY"))
|
||
routes.DistillEngineRef = distillEngine
|
||
// 蒸馏完成 → 自动图谱更新 + 冲突检测 + 被动验证
|
||
distill.OnDistillComplete = func(input distill.DistillInput, result distill.DistillResult) {
|
||
entityNames := make([]string, len(result.Entities))
|
||
for i, e := range result.Entities { entityNames[i] = e.Name }
|
||
graphUpdater.UpdateFromDistill(&governance.DistillInput{
|
||
Content: input.Content, Facts: result.Facts,
|
||
Entities: entityNames, Namespace: input.Namespace,
|
||
})
|
||
existing := make([]map[string]interface{}, 0)
|
||
for _, c := range conflictDetector.Scan(input.Content, entityNames, existing) {
|
||
if c.Strategy == "latest_wins" { conflictDetector.AutoResolve(c) }
|
||
}
|
||
selfoptimize.Validator.Validate(input.Content, nil)
|
||
}
|
||
|
||
// ─── CO_OCCURS 追踪器 ───────────────────────
|
||
storage.CoOccurTrackerInstance = storage.NewCoOccurTracker(nil)
|
||
|
||
// ─── 跨 Agent 缓存失效回调 ───────────────────
|
||
// 收到其他 Agent 的广播 → 失效本地缓存
|
||
governance.GlobalEventBus.Subscribe("cache.invalidate", "self")
|
||
go func() {
|
||
// 本地处理 cache.invalidate 事件（通过 HTTP self-call）
|
||
http.HandleFunc("/_internal/cache/invalidate", func(w http.ResponseWriter, r *http.Request) {
|
||
var evt struct {
|
||
Payload map[string]string `json:"payload"`
|
||
}
|
||
json.NewDecoder(r.Body).Decode(&evt)
|
||
ns := evt.Payload["namespace"]
|
||
storage.SearchCacheInstance.Invalidate(ns)
|
||
log.Printf("[cache] 跨 Agent 失效: %s", ns)
|
||
w.WriteHeader(200)
|
||
})
|
||
}()
|
||
|
||
// ─── 路由注册 ──────────────────────────────
|
||
|
||
mux.HandleFunc("/health", routes.HandleHealth)
|
||
// ─── Prometheus /metrics ─────────────────────
|
||
mux.Handle("/metrics", metrics.Handler())
|
||
|
||
// 核心 API
|
||
mux.HandleFunc("/api/v1/commit", func(w http.ResponseWriter, r *http.Request) {
|
||
api.Commit(w, r)
|
||
// 跨 Agent 广播 + 自动蒸馏
|
||
go func() {
|
||
var req struct {
|
||
AgentID string `json:"agent_id"`
|
||
Namespace string `json:"namespace"`
|
||
Content string `json:"content"`
|
||
}
|
||
if r.Body != nil {
|
||
body, _ := io.ReadAll(r.Body)
|
||
json.Unmarshal(body, &req)
|
||
r.Body = io.NopCloser(bytes.NewReader(body))
|
||
}
|
||
if req.Namespace != "" {
|
||
// 通知其他 Agent：缓存失效 + 新记忆事件
|
||
governance.PushCacheInvalidate(req.Namespace, "commit")
|
||
governance.PushMemoryCommitted(req.AgentID, req.Namespace, "")
|
||
}
|
||
}()
|
||
})
|
||
mux.HandleFunc("/api/v1/recall", api.Recall)
|
||
mux.HandleFunc("/api/v1/bootstrap", api.Bootstrap)
|
||
mux.HandleFunc("/api/v1/stats", api.Stats)
|
||
mux.HandleFunc("/api/v1/batch-commit", api.BatchCommit)
|
||
// ─── WebSocket 实时推送 ──────────────────────
|
||
mux.HandleFunc("/api/v1/ws/", routes.WSHandler)
|
||
mux.HandleFunc("/api/v1/ws", func(w http.ResponseWriter, r *http.Request) {
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.Write([]byte(`{"status":"ok","protocol":"websocket","endpoint":"/api/v1/ws/{agent_id}"}`))
|
||
})
|
||
|
||
// 知识图谱
|
||
mux.HandleFunc("/api/v1/graph/stats", graphAPI.Stats)
|
||
mux.HandleFunc("/api/v1/graph/query", graphAPI.Query)
|
||
mux.HandleFunc("/api/v1/graph/navigate", graphAPI.Navigate)
|
||
// 新增：pagerank + evidence_count
|
||
mux.HandleFunc("/api/v1/graph/pagerank", func(w http.ResponseWriter, r *http.Request) {
|
||
rank := graphStore.PageRank(0.85, 20)
|
||
data, _ := json.Marshal(map[string]interface{}{
|
||
"pagerank": rank,
|
||
"count": len(rank),
|
||
})
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.Write(data)
|
||
})
|
||
mux.HandleFunc("/api/v1/graph/evidence/", func(w http.ResponseWriter, r *http.Request) {
|
||
entity := r.URL.Path[len("/api/v1/graph/evidence/"):]
|
||
count := graphStore.EvidenceCount(entity)
|
||
data, _ := json.Marshal(map[string]interface{}{
|
||
"entity": entity,
|
||
"evidence_count": count,
|
||
})
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.Write(data)
|
||
})
|
||
|
||
// 冲突
|
||
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)
|
||
|
||
// 反馈 — G5: 修正记忆时填充 version_history
|
||
mux.HandleFunc("/api/v1/feedback/useful", func(w http.ResponseWriter, r *http.Request) {
|
||
feedbackAPI.MarkUseful(w, r)
|
||
// 质量监控：useful → 质量上升
|
||
var req struct { MemoryID string `json:"memory_id"` }
|
||
if r.Body != nil {
|
||
body, _ := io.ReadAll(r.Body)
|
||
json.Unmarshal(body, &req)
|
||
r.Body = io.NopCloser(bytes.NewReader(body))
|
||
if req.MemoryID != "" {
|
||
selfoptimize.Dash.RecordUseful()
|
||
selfoptimize.QualityMonitor.Check(req.MemoryID, 0, 5) // 清除潜在告警
|
||
}
|
||
}
|
||
})
|
||
mux.HandleFunc("/api/v1/feedback/not-useful", func(w http.ResponseWriter, r *http.Request) {
|
||
feedbackAPI.MarkNotUseful(w, r)
|
||
// 质量监控：not-useful → 检查是否需要降权
|
||
var req struct { MemoryID string `json:"memory_id"` }
|
||
if r.Body != nil {
|
||
body, _ := io.ReadAll(r.Body)
|
||
json.Unmarshal(body, &req)
|
||
r.Body = io.NopCloser(bytes.NewReader(body))
|
||
if req.MemoryID != "" {
|
||
selfoptimize.Dash.RecordNotUseful()
|
||
score := selfoptimize.Dash.QualityScore()
|
||
if record := selfoptimize.QualityMonitor.Check(req.MemoryID, score, 5); record != nil {
|
||
_ = record // WebSocket 通知
|
||
}
|
||
}
|
||
}
|
||
})
|
||
mux.HandleFunc("/api/v1/feedback/deprecate", feedbackAPI.Deprecate)
|
||
mux.HandleFunc("/api/v1/feedback/correct", func(w http.ResponseWriter, r *http.Request) {
|
||
feedbackAPI.Correct(w, r)
|
||
var req struct {
|
||
MemoryID string `json:"memory_id"`
|
||
NewContent string `json:"new_content"`
|
||
Source string `json:"source"`
|
||
}
|
||
if r.Body != nil {
|
||
body, _ := io.ReadAll(r.Body)
|
||
json.Unmarshal(body, &req)
|
||
r.Body = io.NopCloser(bytes.NewReader(body))
|
||
if req.MemoryID != "" && req.NewContent != "" {
|
||
routes.FillVersionHistory(req.MemoryID, "", req.NewContent, req.Source)
|
||
// 跨 Agent 通知
|
||
governance.PushMemoryUpdated(req.MemoryID, 0, "corrected")
|
||
}
|
||
}
|
||
})
|
||
|
||
// 缺口
|
||
mux.HandleFunc("/api/v1/gaps", gapAPI.List)
|
||
mux.HandleFunc("/api/v1/gaps/detect", gapAPI.Detect)
|
||
mux.HandleFunc("/api/v1/gaps/close/", func(w http.ResponseWriter, r *http.Request) {
|
||
gapAPI.Close(w, r)
|
||
topic := r.URL.Path[len("/api/v1/gaps/close/"):]
|
||
routes.PushGapFilled(topic, 1)
|
||
governance.PushGapFilled(topic, 1)
|
||
})
|
||
mux.HandleFunc("/api/v1/gaps/repair", routes.GapRepair.FullRepairHandler)
|
||
|
||
// Agent 注册
|
||
mux.HandleFunc("/api/v1/agents/register", agentRegistry.Register)
|
||
mux.HandleFunc("/api/v1/agents", agentRegistry.List)
|
||
|
||
// ─── 多 Agent EventBus 管理 ──────────────────
|
||
mux.HandleFunc("/api/v1/eventbus/subscribe", func(w http.ResponseWriter, r *http.Request) {
|
||
var req struct {
|
||
EventType string `json:"event_type"`
|
||
Callback string `json:"callback_url"`
|
||
}
|
||
json.NewDecoder(r.Body).Decode(&req)
|
||
governance.GlobalEventBus.Subscribe(req.EventType, req.Callback)
|
||
respondJSON(w, 200, map[string]string{"status": "subscribed"})
|
||
})
|
||
mux.HandleFunc("/api/v1/eventbus/unsubscribe", func(w http.ResponseWriter, r *http.Request) {
|
||
var req struct {
|
||
EventType string `json:"event_type"`
|
||
Callback string `json:"callback_url"`
|
||
}
|
||
json.NewDecoder(r.Body).Decode(&req)
|
||
governance.GlobalEventBus.Unsubscribe(req.EventType, req.Callback)
|
||
respondJSON(w, 200, map[string]string{"status": "unsubscribed"})
|
||
})
|
||
mux.HandleFunc("/api/v1/eventbus/list", func(w http.ResponseWriter, r *http.Request) {
|
||
subs := governance.GlobalEventBus.ListSubscribers()
|
||
respondJSON(w, 200, subs)
|
||
})
|
||
|
||
// 跨 Agent 缓存失效接收端点（供其他 Agent 实例 HTTP 回调）
|
||
mux.HandleFunc("/api/v1/cache/invalidate", func(w http.ResponseWriter, r *http.Request) {
|
||
var evt struct {
|
||
Type string `json:"type"`
|
||
Payload map[string]string `json:"payload"`
|
||
}
|
||
json.NewDecoder(r.Body).Decode(&evt)
|
||
ns := evt.Payload["namespace"]
|
||
storage.SearchCacheInstance.Invalidate(ns)
|
||
log.Printf("[cache] 跨 Agent 失效接收: %s", ns)
|
||
respondJSON(w, 200, map[string]string{"status": "invalidated"})
|
||
})
|
||
|
||
// 管理
|
||
mux.HandleFunc("/api/v1/admin/consolidate", func(w http.ResponseWriter, r *http.Request) {
|
||
report, err := consolPipe.Run()
|
||
if err != nil {
|
||
respondJSON(w, 500, map[string]string{"error": err.Error()})
|
||
return
|
||
}
|
||
respondJSON(w, 200, report)
|
||
governance.PushDistillationComplete(report.Merged, report.ConflictsFound)
|
||
})
|
||
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/admin/forgetter/type", func(w http.ResponseWriter, r *http.Request) {
|
||
var req struct {
|
||
AgentType string `json:"agent_type"`
|
||
}
|
||
if r.Method == "POST" {
|
||
json.NewDecoder(r.Body).Decode(&req)
|
||
forgetter.SetAgentType(req.AgentType)
|
||
}
|
||
respondJSON(w, 200, map[string]interface{}{
|
||
"agent_type": forgetter.AgentType(),
|
||
"decay_rate": AgentTypeDecayOrDefault(forgetter.AgentType()),
|
||
})
|
||
})
|
||
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)
|
||
// 触发器 pause/resume — 手动解析路径
|
||
mux.HandleFunc("/api/v1/admin/triggers/", func(w http.ResponseWriter, r *http.Request) {
|
||
path := r.URL.Path
|
||
// 格式: /api/v1/admin/triggers/{id}/{action}
|
||
parts := strings.Split(strings.TrimPrefix(path, "/api/v1/admin/triggers/"), "/")
|
||
if len(parts) != 2 {
|
||
http.NotFound(w, r)
|
||
return
|
||
}
|
||
id, action := parts[0], parts[1]
|
||
// 构造一个新请求，让 Pause/Resume 能通过 PathValue 读取
|
||
r.SetPathValue("id", id)
|
||
switch action {
|
||
case "pause":
|
||
routes.Triggers.Pause(w, r)
|
||
case "resume":
|
||
routes.Triggers.Resume(w, r)
|
||
default:
|
||
http.NotFound(w, r)
|
||
}
|
||
})
|
||
|
||
// Skills
|
||
mux.HandleFunc("/api/v1/skills", routes.Skills.List)
|
||
mux.HandleFunc("/api/v1/skills/bayes", func(w http.ResponseWriter, r *http.Request) {
|
||
list := routes.BayesianSkills.List()
|
||
respondJSON(w, 200, 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)
|
||
|
||
// 仪表盘 + 验证 + V值 + 缓存 + 流水线
|
||
mux.HandleFunc("/api/v1/metrics/self", func(w http.ResponseWriter, r *http.Request) {
|
||
metrics := selfoptimize.Dash.Metrics()
|
||
respondJSON(w, 200, metrics)
|
||
})
|
||
mux.HandleFunc("/api/v1/validate/passive", func(w http.ResponseWriter, r *http.Request) {
|
||
records := selfoptimize.Validator.GetRecords()
|
||
respondJSON(w, 200, records)
|
||
})
|
||
mux.HandleFunc("/api/v1/vvalue/decisions", func(w http.ResponseWriter, r *http.Request) {
|
||
decisions := selfoptimize.VProp.ListRecentDecisions(20)
|
||
respondJSON(w, 200, decisions)
|
||
})
|
||
mux.HandleFunc("/api/v1/admin/cache", func(w http.ResponseWriter, r *http.Request) {
|
||
stats := storage.SearchCacheInstance.Stats()
|
||
respondJSON(w, 200, stats)
|
||
})
|
||
mux.HandleFunc("/api/v1/admin/pipeline", func(w http.ResponseWriter, r *http.Request) {
|
||
stats := selfoptimize.Flow.Stats()
|
||
respondJSON(w, 200, stats)
|
||
})
|
||
|
||
// G6: ephemeral namespace 管理
|
||
mux.HandleFunc("/api/v1/admin/ephemeral/clean", func(w http.ResponseWriter, r *http.Request) {
|
||
cleaned := []string{}
|
||
respondJSON(w, 200, map[string]interface{}{
|
||
"status": "cleaned",
|
||
"cleaned_namespaces": cleaned,
|
||
})
|
||
})
|
||
|
||
// 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)
|
||
|
||
// 内部分支：跨 Agent 缓存失效回调
|
||
mux.HandleFunc("/_internal/cache/invalidate", func(w http.ResponseWriter, r *http.Request) {
|
||
var evt struct {
|
||
Payload map[string]string `json:"payload"`
|
||
}
|
||
json.NewDecoder(r.Body).Decode(&evt)
|
||
ns := evt.Payload["namespace"]
|
||
storage.SearchCacheInstance.Invalidate(ns)
|
||
log.Printf("[cache] 内部跨 Agent 失效: %s", ns)
|
||
w.WriteHeader(200)
|
||
})
|
||
|
||
// ─── Redis 组件 ─────────────────────────────
|
||
heartbeat := storage.NewHeartbeat("zhiyid-primary")
|
||
_ = heartbeat // 后台 goroutine 自动心跳
|
||
|
||
// ─── 后台引擎启动 ──────────────────────────
|
||
selfoptimize.RegisterCommitFlow(selfoptimize.Flow)
|
||
selfoptimize.RegisterRecallFlow(selfoptimize.Flow)
|
||
selfoptimize.RegisterGapFlow(selfoptimize.Flow)
|
||
selfoptimize.RegisterCorrectFlow(selfoptimize.Flow)
|
||
selfoptimize.RegisterConsolidateFlow(selfoptimize.Flow)
|
||
go selfoptimize.Flow.Start()
|
||
selfoptimize.Executor.Start(selfoptimize.Flow)
|
||
|
||
// G6: ephemeral 定时清理（每 10 分钟）
|
||
go func() {
|
||
for {
|
||
time.Sleep(10 * time.Minute)
|
||
}
|
||
}()
|
||
|
||
// 图持久化 — 确保启动后写入
|
||
if saver, ok := graphStore.(interface{ Save() error }); ok {
|
||
saver.Save()
|
||
}
|
||
|
||
// ─── 触发器自动执行循环（每 30 秒检查一次）────────
|
||
go func() {
|
||
ticker := time.NewTicker(30 * time.Second)
|
||
defer ticker.Stop()
|
||
for range ticker.C {
|
||
for _, t := range []struct{ id, action string }{
|
||
{"t_distill", "distill"},
|
||
{"t_merge", "merge"},
|
||
{"t_prune", "prune"},
|
||
{"t_decay", "decay"},
|
||
{"t_backtrack", "backtrack"},
|
||
{"t_gap", "gap_scan"},
|
||
{"t_consolidation", "consolidation"},
|
||
} {
|
||
if !routes.Triggers.CanFire(t.id) {
|
||
continue
|
||
}
|
||
routes.Triggers.RecordFire(t.id)
|
||
go func(triggerID, action string) {
|
||
var err error
|
||
switch action {
|
||
case "distill", "consolidation", "backtrack":
|
||
// 深度整合含蒸馏质量回溯（Step 4）
|
||
_, err = consolPipe.Run()
|
||
case "merge":
|
||
_, err = consolPipe.RunMerge()
|
||
case "prune":
|
||
graphStore.Prune(0.15)
|
||
case "decay":
|
||
// 扫描所有记忆，按衰减率计算 freshness
|
||
memories, e := ldb.GetCandidatesForForgetting()
|
||
if e == nil {
|
||
for _, m := range memories {
|
||
// 解析 last_recalled_at
|
||
lastAccessed := time.Now().Add(-30 * 24 * time.Hour) // 默认30天前
|
||
if t, ok := m["last_recalled_at"].(string); ok && t != "" {
|
||
if parsed, err := time.Parse(time.RFC3339, t); err == nil {
|
||
lastAccessed = parsed
|
||
}
|
||
}
|
||
recallCount := 0
|
||
if rc, ok := m["recall_count"].(int); ok {
|
||
recallCount = rc
|
||
}
|
||
tier := "normal"
|
||
if ts, ok := m["tier"].(string); ok {
|
||
tier = ts
|
||
}
|
||
if forgetter.ShouldForget(lastAccessed, recallCount, tier) {
|
||
if id, ok := m["id"].(string); ok {
|
||
forgetter.ScanAndForget(id)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
case "gap_scan":
|
||
// 检查已有缺口：过期 7 天的自动关闭
|
||
gaps := gapDetector.List()
|
||
for _, g := range gaps {
|
||
if !g.Closed && time.Since(g.CreatedAt) > 7*24*time.Hour {
|
||
gapDetector.Close(g.Topic)
|
||
}
|
||
}
|
||
}
|
||
if err != nil {
|
||
routes.Triggers.RecordFail(triggerID)
|
||
}
|
||
routes.WSBus.Broadcast("trigger.fired", map[string]string{
|
||
"trigger_id": triggerID,
|
||
"action": action,
|
||
})
|
||
}(t.id, t.action)
|
||
}
|
||
}
|
||
}()
|
||
|
||
// 自优化指标定时采集（每 6 小时）
|
||
go func() {
|
||
for {
|
||
time.Sleep(6 * time.Hour)
|
||
m := selfoptimize.Dash.Metrics()
|
||
date := time.Now().Format("2006-01-02")
|
||
for k, v := range m {
|
||
storage.GlobalMetricsStore.Set(date, k, v)
|
||
}
|
||
}
|
||
}()
|
||
|
||
// 首次指标采集
|
||
m := selfoptimize.Dash.Metrics()
|
||
date := time.Now().Format("2006-01-02")
|
||
for k, v := range m {
|
||
storage.GlobalMetricsStore.Set(date, k, v)
|
||
}
|
||
|
||
// Prometheus 指标定时同步（每 15s 采集一次系统指标）
|
||
go func() {
|
||
for {
|
||
time.Sleep(15 * time.Second)
|
||
// 同步 Dashboard → Prometheus gauges
|
||
metrics.SyncFromDashboard(selfoptimize.Dash.Metrics())
|
||
// 采集 SQLite 数据库文件大小
|
||
if fi, err := os.Stat(graphPath); err == nil {
|
||
metrics.SQLiteDBSizeBytes.Set(float64(fi.Size()))
|
||
}
|
||
// 采集进程 RSS 内存
|
||
if data, err := os.ReadFile("/proc/self/status"); err == nil {
|
||
for _, line := range strings.Split(string(data), "\n") {
|
||
if strings.HasPrefix(line, "VmRSS:") {
|
||
// VmRSS: 12345 kB
|
||
f := strings.Fields(line)
|
||
if len(f) >= 2 {
|
||
if kb, err := strconv.ParseFloat(f[1], 64); err == nil {
|
||
metrics.MemoryRSSBytes.Set(kb * 1024)
|
||
}
|
||
}
|
||
break
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}()
|
||
|
||
log.Println("[zhiyid] 多 Agent 架构 — Redis + FileGraph + EventBus + vLLM — 已启动")
|
||
return middleware.Auth(mux)
|
||
}
|
||
|
||
// ─── 辅助 ──────────────────────────────────────────
|
||
|
||
func respondJSON(w http.ResponseWriter, code int, data interface{}) {
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.WriteHeader(code)
|
||
json.NewEncoder(w).Encode(data)
|
||
}
|
||
|
||
func AgentTypeDecayOrDefault(agentType string) float64 {
|
||
if rate, ok := governance.AgentTypeDecay[agentType]; ok {
|
||
return rate
|
||
}
|
||
return 0.015
|
||
}
|