memoryweave/go/internal/api/server.go

220 lines
7.9 KiB
Go

// HTTP 服务器 — 路由注册与启动
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/distributed"
"github.com/xiaoxue/memoryweave/internal/governance"
"github.com/xiaoxue/memoryweave/internal/selfoptimize"
"github.com/xiaoxue/memoryweave/internal/storage"
)
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 := routes.NewAPI(ldb, emb, rerank)
graphStore := governance.NewInMemoryGraph()
graphUpdater := governance.NewAutoGraphUpdater(graphStore)
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)
agentRegistry := routes.NewAgentRegistry(nil)
evalAPI := routes.NewEvalAPI(api.Pipeline, ldb)
l3API := routes.WM
// Consolidation 完整流水线
consolPipe := routes.NewConsolidationPipeline(ldb, graphStore, conflictDetector)
// Metrics handler
metricsHandler := distributed.NewMetricsHandler(selfoptimize.Dash, func() map[string]interface{} {
s, _ := ldb.Stats()
return s
})
// ─── 路由注册 ──────────────────────────────
mux.HandleFunc("/health", routes.HandleHealth)
mux.Handle("/metrics", metricsHandler) // Prometheus
// 核心 API
mux.HandleFunc("/api/v1/commit", func(w http.ResponseWriter, r *http.Request) {
api.Commit(w, r)
// 自动触发蒸馏 + 图谱更新
_ = graphUpdater // TODO: wire distill result
})
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)
mux.HandleFunc("/api/v1/ws/", routes.SSEBus.SSEHandler)
// 知识图谱
mux.HandleFunc("/api/v1/graph/stats", graphAPI.Stats)
mux.HandleFunc("/api/v1/graph/query", graphAPI.Query)
mux.HandleFunc("/api/v1/graph/navigate", graphAPI.Navigate)
// 冲突
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)
mux.HandleFunc("/api/v1/gaps/repair", routes.GapRepair.RepairHandler)
// Agent 注册
mux.HandleFunc("/api/v1/agents/register", agentRegistry.Register)
mux.HandleFunc("/api/v1/agents", agentRegistry.List)
// 管理
mux.HandleFunc("/api/v1/admin/consolidate", func(w http.ResponseWriter, r *http.Request) {
report, err := consolPipe.Run()
if err != nil {
data, _ := json.Marshal(map[string]string{"error": err.Error()})
w.WriteHeader(500)
w.Write(data)
return
}
data, _ := json.Marshal(report)
w.Header().Set("Content-Type", "application/json")
w.Write(data)
})
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)
// 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()
data, _ := json.Marshal(list)
w.Header().Set("Content-Type", "application/json")
w.Write(data)
})
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)
})
// 被动验证器
mux.HandleFunc("/api/v1/validate/passive", func(w http.ResponseWriter, r *http.Request) {
records := selfoptimize.Validator.GetRecords()
data, _ := json.Marshal(records)
w.Header().Set("Content-Type", "application/json")
w.Write(data)
})
// V 值
mux.HandleFunc("/api/v1/vvalue/decisions", func(w http.ResponseWriter, r *http.Request) {
decisions := selfoptimize.VProp.ListRecentDecisions(20)
data, _ := json.Marshal(decisions)
w.Header().Set("Content-Type", "application/json")
w.Write(data)
})
// 搜索缓存
mux.HandleFunc("/api/v1/admin/cache", func(w http.ResponseWriter, r *http.Request) {
stats := storage.SearchCacheInstance.Stats()
data, _ := json.Marshal(stats)
w.Header().Set("Content-Type", "application/json")
w.Write(data)
})
// 流水线
mux.HandleFunc("/api/v1/admin/pipeline", func(w http.ResponseWriter, r *http.Request) {
stats := selfoptimize.Flow.Stats()
data, _ := json.Marshal(stats)
w.Header().Set("Content-Type", "application/json")
w.Write(data)
})
// IPC 状态
mux.HandleFunc("/api/v1/admin/ipc", func(w http.ResponseWriter, r *http.Request) {
data, _ := json.Marshal(map[string]string{"socket": "/tmp/zhiyi-consolidate.sock"})
w.Header().Set("Content-Type", "application/json")
w.Write(data)
})
// Obsidian
obsidian := routes.NewObsidianSyncer("", ldb)
carrier := routes.NewObsidianCarrier("")
mux.HandleFunc("/api/v1/obsidian/push", obsidian.PushHandler)
mux.HandleFunc("/api/v1/obsidian/pull", obsidian.PullHandler)
mux.HandleFunc("/api/v1/obsidian/status", obsidian.StatusHandler)
// ─── 启动后台引擎 ──────────────────────────────
// 注册自动化流程
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)
_ = carrier // Obsidian carrier 就绪
log.Println("[zhiyid] 全路由注册完成 + 后台引擎启动")
return middleware.Auth(mux)
}