1351 lines
46 KiB
Go
1351 lines
46 KiB
Go
// HTTP 服务器 — 多 Agent 架构（FileGraph + NetworkEventBus + 跨Agent缓存失效）
|
||
package api
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"log"
|
||
"net/http"
|
||
"os"
|
||
"runtime"
|
||
"sort"
|
||
"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/models"
|
||
"github.com/xiaoxue/memoryweave/internal/selfoptimize"
|
||
"github.com/xiaoxue/memoryweave/internal/storage"
|
||
)
|
||
|
||
// cachedGraphStoreRef 全局图谱缓存引用（用于 graph write 操作时主动失效）
|
||
var cachedGraphStoreRef *storage.GraphCacheRef
|
||
|
||
// normalizeEntity 规整实体名：去特殊字符 + n_前缀
|
||
func normalizeEntity(entity string) string {
|
||
if strings.HasPrefix(entity, "n_") {
|
||
entity = entity[2:]
|
||
}
|
||
// 保留中文、字母、数字、下划线、连字符，过滤其他特殊字符
|
||
clean := strings.Map(func(r rune) rune {
|
||
if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') || r == '_' || r == '-' ||
|
||
(r >= 0x4e00 && r <= 0x9fa5) { // 中文 Unicode 范围
|
||
return r
|
||
}
|
||
return '_'
|
||
}, strings.ToLower(strings.TrimSpace(entity)))
|
||
// 合并连续下划线
|
||
for strings.Contains(clean, "__") {
|
||
clean = strings.ReplaceAll(clean, "__", "_")
|
||
}
|
||
return strings.Trim(clean, "_")
|
||
}
|
||
|
||
// skillByNameHandler GET/DELETE /api/v1/skills/{name} 的 method 分支分发器
|
||
func skillByNameHandler(w http.ResponseWriter, r *http.Request) {
|
||
switch r.Method {
|
||
case http.MethodGet:
|
||
routes.Skills.Get(w, r)
|
||
case http.MethodDelete:
|
||
routes.Skills.Delete(w, r)
|
||
default:
|
||
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
||
}
|
||
}
|
||
|
||
func NewServer() http.Handler {
|
||
mux := http.NewServeMux()
|
||
|
||
// ─── 初始化依赖 ──────────────────────────────
|
||
emb := storage.NewEmbedder(os.Getenv("VLLM_ENDPOINT"))
|
||
rerank := storage.NewReranker(os.Getenv("RERANK_ENDPOINT"), emb)
|
||
|
||
// 存储后端
|
||
ldb := initStorageBackend(os.Getenv("STORAGE_BACKEND"), emb)
|
||
|
||
// 启动时数据目录一致性检查（防止路径混乱导致读取废弃数据）
|
||
runStartupChecks(os.Getenv("STORAGE_BACKEND"))
|
||
|
||
// G7.2: 注册 LDB getter（供 skill crystallize 使用）
|
||
routes.RegisterLDBGetter(func() storage.LanceDB { return ldb })
|
||
|
||
// 冲突检测器（E4.1: Commit 时矛盾检测依赖此实例）
|
||
conflictDetector := governance.NewConflictDetector()
|
||
api := routes.NewAPI(ldb, emb, rerank, conflictDetector)
|
||
|
||
// 启动时初始化 Prometheus 指标
|
||
go func() {
|
||
time.Sleep(1 * time.Second) // 等待 Rust sidecar 就绪
|
||
if stats, err := ldb.Stats(); err == nil {
|
||
if total, ok := stats["total_memories"].(float64); ok {
|
||
metrics.TotalMemories.Set(total)
|
||
}
|
||
if total, ok := stats["total_episodes"].(float64); ok {
|
||
metrics.TotalEpisodes.Set(total)
|
||
}
|
||
log.Printf("[metrics] 初始化指标: memories=%.0f", stats["total_memories"])
|
||
}
|
||
}()
|
||
|
||
// 图谱（带 Navigate 结果缓存：TTL 5min，容量 500，热点实体缓存命中加速）
|
||
graphStore, cachedGraphStoreRef := storage.NewCachedGraphStore(initGraphStore())
|
||
graphUpdater := governance.NewAutoGraphUpdater(graphStore)
|
||
graphAPI := routes.NewGraphAPI(graphStore)
|
||
|
||
// G1: Recall 管线挂图谱扩展
|
||
api.Pipeline.SetGraphExpander(graphStore)
|
||
|
||
// G4: 自动蒸馏 → 注入图谱更新器
|
||
routes.SetGraphUpdater(graphUpdater)
|
||
|
||
// 冲突
|
||
conflictAPI := routes.NewConflictAPI(conflictDetector)
|
||
|
||
// ─── 缺口
|
||
gapDetector := selfoptimize.NewGapDetector(emb, ldb)
|
||
gapDetector.EnableGapRedisPersistence()
|
||
selfoptimize.SetGlobalGapDetector(gapDetector) // 注册全局实例供 recall 回调使用
|
||
|
||
// 因果追踪持久化
|
||
routes.CascadeR.Tracker().EnableCausalRedisPersistence()
|
||
gapAPI := routes.NewGapAPI(gapDetector)
|
||
routes.InitGapRepair(gapDetector, ldb) // 共享同一个 GapDetector + LanceDB
|
||
// G1a: 挂缺隙记录器（0 结果 → 自动记录 miss）
|
||
api.Pipeline.SetMissRecorder(func(query, namespace string) {
|
||
gapDetector.RecordMiss(query + " [" + namespace + "]")
|
||
})
|
||
|
||
feedbackAPI := routes.NewFeedbackAPI(ldb)
|
||
// 遗忘器：支持按 Agent 类型设置衰减率
|
||
forgetter := governance.NewForgetter()
|
||
adminAPI := routes.NewAdminAPI(ldb, forgetter, graphStore)
|
||
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")
|
||
|
||
// Obsidian
|
||
obsidian := routes.NewObsidianSyncer("", ldb)
|
||
|
||
// ─── 蒸馏引擎 ───────────────────────────────
|
||
llmEndpoint := os.Getenv("LLM_ENDPOINT")
|
||
llmModel := os.Getenv("LLM_MODEL")
|
||
if llmModel == "" { llmModel = "deepseek/deepseek-v4-pro" }
|
||
llmAPIKey := os.Getenv("LLM_API_KEY")
|
||
if llmAPIKey == "" { llmAPIKey = os.Getenv("API_KEY") }
|
||
distillEngine := distill.NewEngine(llmEndpoint, llmModel, llmAPIKey)
|
||
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,
|
||
EpisodeID: input.EpisodeID,
|
||
})
|
||
|
||
// 回写蒸馏产物到记忆库（LanceDB）
|
||
for _, fact := range result.Facts {
|
||
mem := models.MemoryRecord{
|
||
ID: fmt.Sprintf("mem_%d", time.Now().UnixNano()),
|
||
AgentID: input.AgentID,
|
||
Namespace: input.Namespace,
|
||
Content: fact,
|
||
Category: "distilled",
|
||
Tier: "normal",
|
||
QualityScore: result.Overall,
|
||
Version: 1,
|
||
CreatedAt: time.Now(),
|
||
UpdatedAt: time.Now(),
|
||
DerivedFrom: input.EpisodeID,
|
||
Freshness: "fresh",
|
||
}
|
||
if err := ldb.InsertMemory(mem); err != nil {
|
||
log.Printf("[distill] commit fact failed: %v", err)
|
||
}
|
||
}
|
||
// 冲突检测：加载同 namespace 已有记忆进行比较
|
||
zeroVec := make([]float32, 1024)
|
||
existingMems, _ := ldb.Search("memories", zeroVec, 100, input.Namespace)
|
||
conflictExisting := make([]map[string]interface{}, 0)
|
||
for _, m := range existingMems {
|
||
conflictExisting = append(conflictExisting, map[string]interface{}{
|
||
"content": m.Content,
|
||
"entities": entityNames, // 用当前内容提取的实体做交叉比较
|
||
})
|
||
}
|
||
for _, c := range conflictDetector.Scan(input.Content, entityNames, conflictExisting) {
|
||
if c.Strategy == "latest_wins" { conflictDetector.AutoResolve(c) }
|
||
}
|
||
// 被动验证：使用已加载的记忆进行 P1/P2/P3 匹配
|
||
validMems := make([]selfoptimize.MemoryForValidation, len(existingMems))
|
||
for i, m := range existingMems {
|
||
validMems[i] = selfoptimize.MemoryForValidation{
|
||
ID: m.ID, Content: m.Content, QualityScore: m.QualityScore,
|
||
}
|
||
}
|
||
selfoptimize.Validator.Validate(input.Content, validMems)
|
||
// 被动验证结果写回 quality_score
|
||
records := selfoptimize.Validator.GetRecords()
|
||
for _, r := range records {
|
||
if r.Confidence > 0 {
|
||
_ = ldb.Update("memories", r.MemoryID, map[string]any{
|
||
"quality_score": r.Confidence,
|
||
})
|
||
}
|
||
}
|
||
}
|
||
|
||
// ─── CO_OCCURS 追踪器（持久化到 Redis）───
|
||
storage.CoOccurTrackerInstance = storage.NewCoOccurTracker(storage.RedisCoOccurPersist())
|
||
storage.LoadCoOccurFromRedis()
|
||
|
||
// ─── Dashboard 持久化到 Redis（restart 不丢指标）───
|
||
selfoptimize.EnableRedisPersistence()
|
||
|
||
// ─── VProp 持久化到 Redis（restart 不丢）───
|
||
selfoptimize.VProp.EnableVPropRedisPersistence()
|
||
|
||
// ─── Skill 持久化到 Redis（G7: restart 不丢 skills）───
|
||
routes.BayesianSkills.EnableRedisPersistence()
|
||
|
||
// ─── V 值传播器（竞争性架构核心）─────────────
|
||
vPropagator := selfoptimize.VProp
|
||
|
||
// 注入 V 值查询函数到存储层（召回排序融合）
|
||
storage.VValueProvider = func(memoryID string) float64 {
|
||
return vPropagator.GetMemVValue(memoryID)
|
||
}
|
||
// 在蒸馏完成回调中记录 V 值
|
||
originalOnComplete := distill.OnDistillComplete
|
||
distill.OnDistillComplete = func(input distill.DistillInput, result distill.DistillResult) {
|
||
// 调用已有的回调
|
||
if originalOnComplete != nil {
|
||
originalOnComplete(input, result)
|
||
}
|
||
// 记录 V 值决策：蒸馏成功 = 有用
|
||
if len(result.Facts) > 0 {
|
||
vPropagator.RecordDecision(
|
||
"distill_"+input.EpisodeID,
|
||
[]string{input.EpisodeID},
|
||
"distill", "success", "",
|
||
)
|
||
}
|
||
// 更新仪表盘指标
|
||
selfoptimize.Dash.RecordUseful()
|
||
// 记录蒸馏质量损失（avg_distill_loss = 1 - Overall）
|
||
// 仅记录真实蒸馏：Overall > 0 表示 LLM 成功返回，过滤 fallback (Overall=0)
|
||
if result.Overall > 0 {
|
||
loss := 1.0 - result.Overall
|
||
if loss < 0 {
|
||
loss = 0
|
||
}
|
||
if loss > 1 {
|
||
loss = 1
|
||
}
|
||
selfoptimize.Dash.RecordDistillLoss(loss)
|
||
}
|
||
}
|
||
|
||
// ─── 跨 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)
|
||
})
|
||
}()
|
||
|
||
// ─── 路由注册 ──────────────────────────────
|
||
|
||
// ─── 静态文件服务（Web UI）─────────────────────
|
||
// webUIRoot 在函数顶部声明
|
||
var webUIRoot string
|
||
// 后续在 catch-all 中处理（见文件末尾）
|
||
|
||
mux.HandleFunc("/health", routes.HandleHealth)
|
||
mux.HandleFunc("/api/v1/health", routes.HandleHealth)
|
||
// 图谱缓存统计
|
||
mux.HandleFunc("/api/v1/cache/stats", func(w http.ResponseWriter, r *http.Request) {
|
||
if cachedGraphStoreRef != nil {
|
||
respondJSON(w, 200, map[string]interface{}{
|
||
"graph_cache": cachedGraphStoreRef.Stats(),
|
||
})
|
||
} else {
|
||
respondJSON(w, 200, map[string]interface{}{"graph_cache": nil})
|
||
}
|
||
})
|
||
// ─── 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/recall/debug", api.RecallDebug)
|
||
mux.HandleFunc("/api/v1/feedback", api.Feedback)
|
||
mux.HandleFunc("/api/v1/bootstrap", api.Bootstrap)
|
||
mux.HandleFunc("/api/v1/stats", api.Stats)
|
||
mux.HandleFunc("/api/v1/batch-commit", api.BatchCommit)
|
||
// 自优化指标（Dashboard，含 avg_distill_loss）
|
||
mux.HandleFunc("/api/v1/metrics", func(w http.ResponseWriter, r *http.Request) {
|
||
m := selfoptimize.Dash.Metrics()
|
||
if stats, err := ldb.Stats(); err == nil {
|
||
if total, ok := stats["total_memories"].(float64); ok {
|
||
m["total_memories"] = total
|
||
}
|
||
if total, ok := stats["total_episodes"].(float64); ok {
|
||
m["total_episodes"] = total
|
||
}
|
||
}
|
||
respondJSON(w, 200, m)
|
||
})
|
||
// ─── 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)
|
||
// 新增：添加关系边（POST JSON body: {"from":"实体A","to":"实体B","relation":"关系类型","namespace":""}）
|
||
mux.HandleFunc("/api/v1/graph/edge", func(w http.ResponseWriter, r *http.Request) {
|
||
if r.Method != "POST" {
|
||
http.Error(w, "POST only", 405)
|
||
return
|
||
}
|
||
var req struct {
|
||
From string `json:"from"`
|
||
To string `json:"to"`
|
||
Relation string `json:"relation"`
|
||
Namespace string `json:"namespace"`
|
||
}
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||
http.Error(w, err.Error(), 400)
|
||
return
|
||
}
|
||
if req.From == "" || req.To == "" || req.Relation == "" {
|
||
http.Error(w, "from, to, relation required", 400)
|
||
return
|
||
}
|
||
ns := req.Namespace
|
||
if ns == "" {
|
||
ns = "default"
|
||
}
|
||
edgeID := fmt.Sprintf("e_%s_%s_%d", req.From, req.Relation, time.Now().UnixNano())
|
||
err := graphStore.AddEdge(edgeID, req.From, req.To, req.Relation, ns, 1.0)
|
||
if err != nil {
|
||
http.Error(w, err.Error(), 500)
|
||
return
|
||
}
|
||
// 图谱变更：失效 navigate 缓存
|
||
if cachedGraphStoreRef != nil {
|
||
cachedGraphStoreRef.InvalidateAll()
|
||
}
|
||
json.NewEncoder(w).Encode(map[string]interface{}{"status": "ok", "edge_id": edgeID})
|
||
})
|
||
// 新增：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/graph/export", func(w http.ResponseWriter, r *http.Request) {
|
||
ns := r.URL.Query().Get("namespace")
|
||
limitStr := r.URL.Query().Get("limit")
|
||
limit := 0
|
||
if limitStr != "" {
|
||
if l, err := strconv.Atoi(limitStr); err == nil && l > 0 {
|
||
limit = l
|
||
}
|
||
}
|
||
nodes, edges := graphStore.GetGraph(ns, limit)
|
||
respondJSON(w, 200, map[string]interface{}{
|
||
"nodes": nodes,
|
||
"edges": edges,
|
||
"count": map[string]int{"nodes": len(nodes), "edges": len(edges)},
|
||
})
|
||
})
|
||
|
||
// 图谱导航 + Obsidian 笔记关联
|
||
mux.HandleFunc("/api/v1/graph/notes", func(w http.ResponseWriter, r *http.Request) {
|
||
entity := r.URL.Query().Get("entity")
|
||
maxHops := 2
|
||
if h := r.URL.Query().Get("max_hops"); h != "" {
|
||
if parsed, err := strconv.Atoi(h); err == nil && parsed > 0 {
|
||
maxHops = parsed
|
||
}
|
||
}
|
||
maxNotes := 10
|
||
if n := r.URL.Query().Get("max_notes"); n != "" {
|
||
if parsed, err := strconv.Atoi(n); err == nil && parsed > 0 {
|
||
maxNotes = parsed
|
||
}
|
||
}
|
||
|
||
var noteResults []map[string]interface{}
|
||
var paths []map[string]interface{}
|
||
if entity != "" {
|
||
p, err := graphStore.Navigate(normalizeEntity(entity), maxHops, "", nil)
|
||
if err == nil {
|
||
paths = p
|
||
entities := []string{normalizeEntity(entity)}
|
||
for _, path := range p {
|
||
if s, ok := path["source"].(string); ok {
|
||
entities = append(entities, s)
|
||
}
|
||
if t, ok := path["target"].(string); ok {
|
||
entities = append(entities, t)
|
||
}
|
||
}
|
||
if obsidian != nil {
|
||
noteResults = obsidian.SearchNotesByEntities(entities, maxNotes)
|
||
}
|
||
}
|
||
}
|
||
respondJSON(w, 200, map[string]interface{}{
|
||
"entity": entity,
|
||
"graph_paths": paths,
|
||
"notes": noteResults,
|
||
"paths_count": len(paths),
|
||
"notes_count": len(noteResults),
|
||
})
|
||
})
|
||
|
||
// 自然语言图谱查询 /api/v1/graph/nl_query
|
||
// 输入: {"query": "织忆和牧尘是什么关系"}
|
||
// 返回: 自然语言关系描述
|
||
mux.HandleFunc("/api/v1/graph/nl_query", func(w http.ResponseWriter, r *http.Request) {
|
||
if r.Method != "POST" {
|
||
http.Error(w, "POST only", http.StatusMethodNotAllowed)
|
||
return
|
||
}
|
||
var req struct {
|
||
Query string `json:"query"`
|
||
}
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||
http.Error(w, "invalid body", http.StatusBadRequest)
|
||
return
|
||
}
|
||
query := strings.TrimSpace(req.Query)
|
||
if query == "" {
|
||
http.Error(w, "query is required", http.StatusBadRequest)
|
||
return
|
||
}
|
||
|
||
// 1. 从 query 中提取可能的实体名（去掉问号、语气词）
|
||
cleanQuery := strings.TrimSuffix(strings.TrimSuffix(query, "？"), "?")
|
||
cleanQuery = strings.TrimPrefix(cleanQuery, "请问")
|
||
cleanQuery = strings.TrimPrefix(cleanQuery, "请说")
|
||
cleanQuery = strings.TrimSpace(cleanQuery)
|
||
|
||
// 2. 尝试识别实体对（"A和B的关系"、"A与B"）
|
||
var entityA, entityB string
|
||
// 模式1: "A和B是什么关系" / "A和B有什么关系" / "A和B的关系"
|
||
hasAnd := strings.Contains(cleanQuery, "和") || strings.Contains(cleanQuery, "与")
|
||
if hasAnd && (strings.Contains(cleanQuery, "关系") || strings.Contains(cleanQuery, "关系")) {
|
||
delim := "和"
|
||
if strings.Contains(cleanQuery, "与") {
|
||
delim = "与"
|
||
}
|
||
parts := strings.Split(cleanQuery, delim)
|
||
if len(parts) == 2 {
|
||
entityA = strings.TrimSpace(parts[0])
|
||
rawB := strings.TrimSpace(parts[1])
|
||
// 去掉句末的问句模式得到 entityB
|
||
entityB = strings.TrimSuffix(rawB, "是什么关系")
|
||
entityB = strings.TrimSuffix(entityB, "有什么关系")
|
||
entityB = strings.TrimSuffix(entityB, "的")
|
||
entityB = strings.TrimSuffix(entityB, "是什么")
|
||
entityB = strings.TrimSpace(entityB)
|
||
// 如果去掉问句后为空（如"织忆和牧尘" -> entityB=""），则原始 query 可能是"A和B"
|
||
if entityB == "" {
|
||
entityB = rawB
|
||
}
|
||
}
|
||
}
|
||
|
||
var result map[string]interface{}
|
||
if entityA != "" && entityB != "" {
|
||
// 双向查询：A → B 的关系
|
||
paths, _ := graphStore.Navigate(normalizeEntity(entityA), 2, "", nil)
|
||
relB, _ := graphStore.Navigate(normalizeEntity(entityB), 2, "", nil)
|
||
// 找 A → B 的直接边
|
||
var directPath map[string]interface{}
|
||
for _, p := range paths {
|
||
toNorm := normalizeEntity(p["to"].(string))
|
||
targetNorm := normalizeEntity(entityB)
|
||
if toNorm == targetNorm {
|
||
directPath = p
|
||
break
|
||
}
|
||
}
|
||
result = map[string]interface{}{
|
||
"query": query,
|
||
"entity_a": entityA,
|
||
"entity_b": entityB,
|
||
"direct_path": directPath,
|
||
"paths_from_a": len(paths),
|
||
"paths_from_b": len(relB),
|
||
"summary": fmt.Sprintf("%s 与 %s 存在 %d 条关联路径", entityA, entityB, len(paths)),
|
||
}
|
||
} else {
|
||
// 单实体查询：找 entity 的关系网
|
||
entity := cleanQuery
|
||
paths, _ := graphStore.Navigate(normalizeEntity(entity), 2, "", nil)
|
||
// 按 relation 分组
|
||
relCounts := make(map[string]int)
|
||
for _, p := range paths {
|
||
rel := p["relation"].(string)
|
||
if rel == "" {
|
||
rel = "unknown"
|
||
}
|
||
relCounts[rel]++
|
||
}
|
||
// 生成自然语言描述
|
||
var rels []string
|
||
for rel, cnt := range relCounts {
|
||
rels = append(rels, fmt.Sprintf("%s(%d条)", rel, cnt))
|
||
}
|
||
sort.Strings(rels)
|
||
result = map[string]interface{}{
|
||
"query": query,
|
||
"entity": entity,
|
||
"total_paths": len(paths),
|
||
"relation_summary": rels,
|
||
"summary": fmt.Sprintf("%s 共有 %d 条关联，分布在 %d 种关系类型", entity, len(paths), len(relCounts)),
|
||
}
|
||
}
|
||
respondJSON(w, 200, result)
|
||
})
|
||
|
||
// 图谱脏数据清理 /api/v1/graph/cleanup
|
||
mux.HandleFunc("/api/v1/graph/cleanup", func(w http.ResponseWriter, r *http.Request) {
|
||
if r.Method != "POST" && r.Method != "GET" {
|
||
http.Error(w, "GET or POST only", http.StatusMethodNotAllowed)
|
||
return
|
||
}
|
||
dryRun := r.URL.Query().Get("dry_run") != "false" // 默认为 dryRun
|
||
count, ids, err := graphStore.CleanupNoiseNodes(dryRun)
|
||
if err != nil {
|
||
http.Error(w, "cleanup failed: "+err.Error(), http.StatusInternalServerError)
|
||
return
|
||
}
|
||
// 实际执行后失效缓存（dry_run=true 时不执行，无需失效）
|
||
if !dryRun && cachedGraphStoreRef != nil && count > 0 {
|
||
cachedGraphStoreRef.InvalidateAll()
|
||
}
|
||
respondJSON(w, 200, map[string]interface{}{
|
||
"dry_run": dryRun,
|
||
"removed": count,
|
||
"node_ids": ids,
|
||
"message": fmt.Sprintf("Found %d noise nodes", count),
|
||
})
|
||
})
|
||
|
||
// 静态文件服务（知识图谱可视化 HTML）
|
||
staticDir := os.Getenv("STATIC_DIR")
|
||
if staticDir == "" {
|
||
staticDir = "/home/muc/projects/memoryweave/go/static"
|
||
}
|
||
fileServer := http.FileServer(http.Dir(staticDir))
|
||
mux.Handle("/static/", http.StripPrefix("/static/", fileServer))
|
||
|
||
// 冲突
|
||
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()
|
||
// V 值：useful = success
|
||
selfoptimize.VProp.RecordDecision(
|
||
"useful_"+req.MemoryID,
|
||
[]string{req.MemoryID},
|
||
"user_feedback", "success", "",
|
||
)
|
||
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()
|
||
// V 值：not-useful = failure
|
||
selfoptimize.VProp.RecordDecision(
|
||
"notuseful_"+req.MemoryID,
|
||
[]string{req.MemoryID},
|
||
"user_feedback", "failure", "",
|
||
)
|
||
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"})
|
||
})
|
||
|
||
// 管理
|
||
// ?task=full 触发完整整合（DBSCAN + 衰减校准 + 质量回溯），默认 cluster_only
|
||
mux.HandleFunc("/api/v1/admin/consolidate", func(w http.ResponseWriter, r *http.Request) {
|
||
task := r.URL.Query().Get("task")
|
||
if task == "" {
|
||
task = "cluster_only"
|
||
}
|
||
// full 模式给 5 分钟；cluster_only 保持 30 秒
|
||
timeout := 30 * time.Second
|
||
if task == "full" {
|
||
timeout = 5 * time.Minute
|
||
}
|
||
ctx, cancel := context.WithTimeout(r.Context(), timeout)
|
||
defer cancel()
|
||
done := make(chan *routes.ConsolidationReport, 1)
|
||
go func() {
|
||
report, _ := consolPipe.RunWithMode(task)
|
||
done <- report
|
||
}()
|
||
select {
|
||
case <-ctx.Done():
|
||
respondJSON(w, 408, map[string]string{"error": "consolidation timeout", "task": task})
|
||
return
|
||
case report := <-done:
|
||
if report == nil {
|
||
respondJSON(w, 500, map[string]string{"error": "consolidation failed", "task": task})
|
||
return
|
||
}
|
||
respondJSON(w, 200, report)
|
||
go governance.PushDistillationComplete(report.Merged, report.ConflictsFound)
|
||
}
|
||
})
|
||
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) {
|
||
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/admin/distill/force", func(w http.ResponseWriter, r *http.Request) {
|
||
if routes.DistillEngineRef == nil {
|
||
respondJSON(w, 400, map[string]string{"error": "distill engine not initialized"})
|
||
return
|
||
}
|
||
var req struct {
|
||
Content string `json:"content"`
|
||
Category string `json:"category"`
|
||
Namespace string `json:"namespace"`
|
||
AgentID string `json:"agent_id"`
|
||
}
|
||
json.NewDecoder(r.Body).Decode(&req)
|
||
if req.Content == "" {
|
||
respondJSON(w, 400, map[string]string{"error": "content required"})
|
||
return
|
||
}
|
||
if req.Namespace == "" { req.Namespace = "hermes-main" }
|
||
if req.AgentID == "" { req.AgentID = "hermes-a06" }
|
||
epID := fmt.Sprintf("ep_manual_%d", time.Now().UnixNano())
|
||
routes.AutoDistillTrigger(epID, req.Content, req.Category, req.Namespace, req.AgentID)
|
||
respondJSON(w, 200, map[string]string{"status": "queued", "episode_id": epID})
|
||
})
|
||
// 蒸馏队列状态端点（G8-G9 配套）
|
||
mux.HandleFunc("/api/v1/distill/status", func(w http.ResponseWriter, r *http.Request) {
|
||
if routes.DistillEngineRef == nil {
|
||
respondJSON(w, 200, map[string]interface{}{"status": "not_initialized", "queue_len": 0})
|
||
return
|
||
}
|
||
respondJSON(w, 200, routes.DistillEngineRef.GetStatus())
|
||
})
|
||
mux.HandleFunc("/api/v1/distill/queue", func(w http.ResponseWriter, r *http.Request) {
|
||
if routes.DistillEngineRef == nil {
|
||
respondJSON(w, 200, map[string]interface{}{"queue": []interface{}{}, "count": 0})
|
||
return
|
||
}
|
||
items := routes.DistillEngineRef.QueueItems()
|
||
respondJSON(w, 200, map[string]interface{}{"queue": items, "count": len(items)})
|
||
})
|
||
mux.HandleFunc("/api/v1/distill/quota", func(w http.ResponseWriter, r *http.Request) {
|
||
if routes.DistillEngineRef == nil {
|
||
respondJSON(w, 200, map[string]interface{}{"remaining": 0, "used": 0, "limit": 0, "status": "not_initialized"})
|
||
return
|
||
}
|
||
respondJSON(w, 200, routes.DistillEngineRef.GetQuota())
|
||
})
|
||
mux.HandleFunc("/api/v1/distilled/{id}", func(w http.ResponseWriter, r *http.Request) {
|
||
if r.Method == "DELETE" {
|
||
adminAPI.DeleteDistilled(w, r)
|
||
} else {
|
||
http.NotFound(w, r)
|
||
}
|
||
})
|
||
// GET /api/v1/memories — list all memories (zero-vector search, for plugin compat)
|
||
// mux.HandleFunc("/api/v1/memories", api.ListMemories) // removed: ListMemories not in routes
|
||
// /api/v1/memory/{id}/versions — existing version history endpoint
|
||
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（G7）
|
||
mux.HandleFunc("/api/v1/skills", routes.Skills.List)
|
||
mux.HandleFunc("POST /api/v1/skills", routes.Skills.Register) // 注册新 skill
|
||
mux.HandleFunc("/api/v1/skills/stats", routes.Skills.Stats) // skill 统计
|
||
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/{name}", skillByNameHandler) // GET/DELETE /api/v1/skills/{name}
|
||
mux.HandleFunc("/api/v1/skills/{name}/trial", routes.Skills.Trial)
|
||
mux.HandleFunc("/api/v1/skills/{name}/execute", routes.ExecuteSkill) // G7.3 skill执行
|
||
// G7.2: 结晶路由
|
||
mux.HandleFunc("/api/v1/crystallize/candidates", routes.GetSkillCandidates) // 获取候选
|
||
mux.HandleFunc("/api/v1/crystallize/memory/{id}", routes.CrystallizeSkill) // 对记忆执行结晶
|
||
|
||
// 评估 + 自调参
|
||
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)
|
||
})
|
||
// G5.3: V 值查询 + 触发反向传播
|
||
mux.HandleFunc("/api/v1/vvalue/memory/", func(w http.ResponseWriter, r *http.Request) {
|
||
memID := strings.TrimPrefix(r.URL.Path, "/api/v1/vvalue/memory/")
|
||
if memID == "" {
|
||
respondJSON(w, 400, map[string]string{"error": "memory_id required"})
|
||
return
|
||
}
|
||
v := selfoptimize.VProp.GetMemVValue(memID)
|
||
respondJSON(w, 200, map[string]interface{}{"memory_id": memID, "v_value": v})
|
||
})
|
||
mux.HandleFunc("/api/v1/trace/vprop", func(w http.ResponseWriter, r *http.Request) {
|
||
var req struct {
|
||
MemoryIDs []string `json:"memory_ids"`
|
||
Action string `json:"action"`
|
||
Outcome string `json:"outcome"`
|
||
ParentID string `json:"parent_id"`
|
||
}
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err == nil && len(req.MemoryIDs) > 0 {
|
||
id := fmt.Sprintf("trace_%d", time.Now().UnixNano())
|
||
selfoptimize.VProp.RecordDecision(id, req.MemoryIDs, req.Action, req.Outcome, req.ParentID)
|
||
// 更新相关记忆的 quality_score
|
||
for _, mid := range req.MemoryIDs {
|
||
v := selfoptimize.VProp.GetMemVValue(mid)
|
||
lamb := 0.1
|
||
if ldb, ok := ldb.(interface{ UpdateQualityScore(id string, delta float64) error }); ok {
|
||
_ = ldb.UpdateQualityScore(mid, lamb*v)
|
||
}
|
||
}
|
||
respondJSON(w, 200, map[string]string{"status": "propagated"})
|
||
return
|
||
}
|
||
respondJSON(w, 400, map[string]string{"error": "memory_ids and outcome required"})
|
||
})
|
||
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
|
||
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)
|
||
// 注册真正的 consolidate 处理器：调用 ConsolidationPipeline
|
||
selfoptimize.Flow.Register("consolidate", func(task *selfoptimize.PipelineTask) error {
|
||
report, err := consolPipe.RunWithMode("full")
|
||
if err == nil && report != nil {
|
||
// 注入 consolidate 报告 → Dashboard
|
||
if report.Merged > 0 {
|
||
// 合并启发式: 按合并数比例缩放, 但限定单次贡献 ≤ 0.05, 防止污染 avg_distill_loss
|
||
loss := float64(report.Merged) * 0.002
|
||
if loss > 0.05 {
|
||
loss = 0.05
|
||
}
|
||
selfoptimize.Dash.RecordDistillLoss(loss)
|
||
}
|
||
if report.ConflictsFound > 0 {
|
||
selfoptimize.Dash.RecordConflictResolved(true)
|
||
}
|
||
for _, p := range report.Patterns {
|
||
if len(p) > 13 && p[:13] == "quality_score" {
|
||
// 解析 quality_score=0.50 → 注入 avg_distill_loss
|
||
var qs float64
|
||
if _, e := fmt.Sscanf(p, "quality_score=%f", &qs); e == nil {
|
||
selfoptimize.Dash.RecordDistillLoss(1.0 - qs)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
return err
|
||
})
|
||
// 注册 conflict_muchen 处理器：自动裁决低信任差异冲突
|
||
selfoptimize.Flow.Register("conflict_muchen", func(task *selfoptimize.PipelineTask) error {
|
||
conflicts := conflictAPI.Detector.ListActive()
|
||
for _, c := range conflicts {
|
||
if c.Status != "pending" {
|
||
continue
|
||
}
|
||
// 按策略自动裁决：latest_wins / primary_wins / dismiss
|
||
resolution := conflictAPI.Detector.AutoResolve(c)
|
||
if resolution == "pending" {
|
||
// 低信任差异（两记忆 trust 差 < 0.2）→ dismiss
|
||
resolution = "dismiss"
|
||
}
|
||
if err := conflictAPI.Detector.Resolve(c.ID, resolution, ""); err == nil {
|
||
selfoptimize.Dash.RecordConflictResolved(true)
|
||
}
|
||
}
|
||
return nil
|
||
})
|
||
go selfoptimize.Flow.Start()
|
||
selfoptimize.Executor.Start(selfoptimize.Flow)
|
||
|
||
// G6: ephemeral 定时清理（每 5 分钟扫描一次，清理断开超过 5 分钟的 agent 的 ephemeral 记忆）
|
||
go func() {
|
||
for {
|
||
time.Sleep(5 * time.Minute)
|
||
now := time.Now().UnixMilli()
|
||
agents := agentRegistry.ListAgents()
|
||
for _, agent := range agents {
|
||
if now-agent.LastSeen > 5*60*1000 {
|
||
ns := agent.AgentID + "-ephemeral"
|
||
zeroVec := make([]float32, 1024)
|
||
memories, _ := ldb.Search("memories", zeroVec, 1000, ns)
|
||
for _, mem := range memories {
|
||
ldb.SoftDelete(mem.ID, "ephemeral_expired")
|
||
}
|
||
if len(memories) > 0 {
|
||
log.Printf("[ephemeral] cleaned %d memories from %s (inactive for %ds)",
|
||
len(memories), ns, (now-agent.LastSeen)/1000)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}()
|
||
|
||
// 质量下降监控：每 30 分钟扫描所有低质量记忆
|
||
go func() {
|
||
for {
|
||
time.Sleep(30 * time.Minute)
|
||
zeroVec := make([]float32, 1024)
|
||
memories, err := ldb.Search("memories", zeroVec, 2000, "")
|
||
if err != nil || len(memories) == 0 {
|
||
continue
|
||
}
|
||
alerts := 0
|
||
for _, m := range memories {
|
||
if m.QualityScore > 0 && m.QualityScore < 0.3 {
|
||
if record := selfoptimize.QualityMonitor.Check(m.ID, m.QualityScore, 2); record != nil {
|
||
alerts++
|
||
if record.Status == "deprecating" {
|
||
log.Printf("[quality] memory %s quality=%.2f deprecating", m.ID[:20], m.QualityScore)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
if alerts > 0 {
|
||
log.Printf("[quality] scan: %d low-quality memories flagged", alerts)
|
||
}
|
||
}
|
||
}()
|
||
|
||
// 图持久化 — 确保启动后写入
|
||
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":
|
||
// distill 保持 cluster_only（高频触发，只做快速聚类）
|
||
_, err = consolPipe.Run()
|
||
case "consolidation", "backtrack":
|
||
// full 模式：含 prune + decay + 质量回溯（Step 4）
|
||
// 48h cooldown 保护，不会真正每分钟跑 full（cooldown 内 CanFire 仍返回 false）
|
||
_, err = consolPipe.RunWithMode("full")
|
||
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
|
||
}
|
||
// E4.3: 获取实体图谱度（namespace 作为实体）
|
||
entity := ""
|
||
if ns, ok := m["namespace"].(string); ok {
|
||
entity = ns
|
||
}
|
||
degree := 0
|
||
if entity != "" {
|
||
degree = graphStore.GetEntityDegree(entity)
|
||
}
|
||
if forgetter.ShouldForget(lastAccessed, recallCount, tier, degree) {
|
||
if id, ok := m["id"].(string); ok {
|
||
forgetter.ScanAndForget(id)
|
||
// 实际执行软删除（持久化到 LanceDB）
|
||
_ = ldb.SoftDelete(id, "auto_forget")
|
||
}
|
||
}
|
||
}
|
||
}
|
||
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)
|
||
}
|
||
}
|
||
// 自动修复开放的缺口
|
||
for _, g := range gaps {
|
||
if !g.Closed && routes.GapRepair != nil {
|
||
if routes.GapRepair.AutoRepair(g) {
|
||
selfoptimize.Dash.RecordGapClosed()
|
||
}
|
||
}
|
||
}
|
||
}
|
||
if err != nil {
|
||
routes.Triggers.RecordFail(triggerID)
|
||
}
|
||
// Skill 结晶：每次蒸馏成功后记录
|
||
if action == "distill" || action == "consolidation" {
|
||
routes.Skills.RecordTrial("auto_distill", true)
|
||
}
|
||
routes.WSBus.Broadcast("trigger.fired", map[string]string{
|
||
"trigger_id": triggerID,
|
||
"action": action,
|
||
})
|
||
}(t.id, t.action)
|
||
}
|
||
}
|
||
}()
|
||
|
||
// 自优化指标定时采集（每 30 分钟）
|
||
go func() {
|
||
for {
|
||
time.Sleep(30 * time.Minute)
|
||
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 数据库文件大小（Linux only）
|
||
if runtime.GOOS != "windows" {
|
||
gPath := os.Getenv("GRAPH_PATH")
|
||
if gPath == "" {
|
||
gPath = "/var/lib/memoryweave/graph.db"
|
||
}
|
||
if fi, err := os.Stat(gPath); 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 — 已启动")
|
||
|
||
// Web UI 静态文件根目录
|
||
webUIRoot = os.Getenv("ZHIYI_WEB_UI_ROOT")
|
||
if webUIRoot == "" {
|
||
webUIRoot = "../web-ui"
|
||
}
|
||
|
||
// Catch-all: 优先检查静态文件，再 JSON
|
||
mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
|
||
p := r.URL.Path
|
||
|
||
// 静态文件服务（安全检查，不允许 .. 遍历）
|
||
if p != "/" && !strings.Contains(p, "..") {
|
||
filePath := webUIRoot + p
|
||
if _, err := os.Stat(filePath); err == nil {
|
||
http.ServeFile(w, r, filePath)
|
||
return
|
||
}
|
||
}
|
||
|
||
// 根路径 → index.html
|
||
if p == "/" || p == "" {
|
||
if _, err := os.Stat(webUIRoot + "/index.html"); err == nil {
|
||
http.ServeFile(w, r, webUIRoot+"/index.html")
|
||
return
|
||
}
|
||
respondJSON(w, 200, map[string]interface{}{
|
||
"service": "zhiyid", "status": "ok", "version": "0.1.0",
|
||
})
|
||
return
|
||
}
|
||
respondJSON(w, 404, map[string]interface{}{
|
||
"error": "endpoint not found: " + p,
|
||
})
|
||
})
|
||
|
||
return middleware.CORS()(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
|
||
}
|
||
|
||
// runStartupChecks 启动时数据目录一致性检查
|
||
// 检测废弃路径（如 /home/muc/data）并警告，防止数据源混乱
|
||
func runStartupChecks(backend string) {
|
||
canonicalDataDir := "/var/lib/memoryweave"
|
||
deprecatedPaths := []string{
|
||
"/home/muc/data",
|
||
"/home/muc/.local/share/memoryweave",
|
||
}
|
||
|
||
for _, dep := range deprecatedPaths {
|
||
if _, err := os.Stat(dep); err == nil {
|
||
log.Printf("[WARN] 检测到废弃数据目录 %s，仍有数据残留（当前 canonical: %s）", dep, canonicalDataDir)
|
||
}
|
||
}
|
||
|
||
// LanceDB 后端：检查 Rust sidecar socket 是否可达
|
||
if backend == "lancedb" {
|
||
sockPath := os.Getenv("LANCEDB_SOCKET")
|
||
if sockPath == "" {
|
||
sockPath = "/tmp/zhiyi-ipc.sock"
|
||
}
|
||
if _, err := os.Stat(sockPath); os.IsNotExist(err) {
|
||
log.Printf("[WARN] LanceDB socket 不存在 (%s)，Rust sidecar 可能未运行", sockPath)
|
||
} else {
|
||
log.Printf("[startup] LanceDB socket 就绪: %s", sockPath)
|
||
}
|
||
}
|
||
|
||
// canonical 目录存在性检查
|
||
if _, err := os.Stat(canonicalDataDir); os.IsNotExist(err) {
|
||
log.Printf("[WARN] canonical 数据目录不存在: %s（首次部署？）", canonicalDataDir)
|
||
} else {
|
||
log.Printf("[startup] 数据目录正常: %s", canonicalDataDir)
|
||
}
|
||
}
|