diff --git a/go/internal/api/routes/core.go b/go/internal/api/routes/core.go index 0ae3cb7..64aa402 100644 --- a/go/internal/api/routes/core.go +++ b/go/internal/api/routes/core.go @@ -271,6 +271,162 @@ func (a *API) Commit(w http.ResponseWriter, r *http.Request) { respond(w, 201, respData) } +// POST /api/v1/memory/write — 显式写入 memories 表（主症A，t_fa6c9104，2026-10-16） +// +// 背景：/api/v1/commit 把原文落 episodes 表、等异步 LLM 蒸馏（质量门丢弃率 ~68%）。 +// 「显式保存重要事实」场景（Hermes 插件 memory_write）走 commit ⇒ 写进去的东西 +// 自己召回不回来：episodes 表（recall 查 memories 表）+ 零向量 + 靠蒸馏流水线 +// 升级成 mem_*（大概率静默丢失）。 +// +// 本端点：同步嵌入 → 去重/合并 → 矛盾只报告不拦 → 直写 memories 表 ⇒ 写完立刻可召回。 +// 硬约束遵守： +// - 纯新增路由：commit / recall / 蒸馏 路径零改动（默认零行为变化） +// - 嵌入失败 ⇒ 502 不写（写零向量 = 主症A 病根） +// - category 缺省 distilled：不在 recall Step 2.5 rawCategories 过滤集里 +// - 不走写时四操作门控（ZHIYI_WRITE_SEMANTIC_OPS）：显式写入 = 调用方意图， +// 去重 3a/3b + 矛盾报告已足够；门控只约束 LLM 蒸馏事实 +func (a *API) WriteMemory(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + respondError(w, 405, "POST only") + return + } + var req struct { + Content string `json:"content"` + Category string `json:"category"` + AgentID string `json:"agent_id"` + Namespace string `json:"namespace"` + Source string `json:"source"` + EmbeddingModel string `json:"embedding_model"` + ScriptVersion string `json:"script_version"` + Confidence *float64 `json:"confidence"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + respondError(w, 400, "invalid body") + return + } + if len([]rune(req.Content)) < 8 { + respondError(w, 400, "content too short (<8 chars)") + return + } + if req.AgentID == "" { + req.AgentID = "hermes-a06" + } + if req.Category == "" { + req.Category = "distilled" + } + if req.Namespace == "" { + req.Namespace = classifyNamespace(req.Category, req.AgentID, req.Content) + } + src := req.Source + if src == "" { + src = "explicit_write" + } + embModel := req.EmbeddingModel + if embModel == "" { + embModel = "bge-m3" + } + scriptVer := req.ScriptVersion + if scriptVer == "" { + scriptVer = "hermes-zhiyi" + } + + // 1. 同步嵌入：失败 ⇒ 502，**不写零向量记录**（零向量 = 主症A 病根） + vector, err := a.Embedder.EncodeSingle(req.Content) + if err != nil { + respondErrorSchema(w, 502, "embed_failed", "embed failed: "+err.Error(), 5) + return + } + + // 2. 去重：exact/near-dup ⇒ 并入既有记忆（与 commit 3a/3b 同语义） + similar, _ := a.LanceDB.Search("memories", vector, 5, req.Namespace) + for _, existing := range similar { + if existing.Content == req.Content { + a.mergeMemory(existing, req.Content, req.Category, req.AgentID) + respond(w, 200, map[string]interface{}{ + "status": "merged", "memory_id": existing.ID, "reason": "exact_dup", + "table": "memories", + }) + return + } + if len(existing.Content) > 0 && len(req.Content) > 0 { + ev, eErr := a.Embedder.EncodeSingle(existing.Content) + if eErr == nil && cosineSimilarity(vector, ev) >= 0.98 { + a.mergeMemory(existing, req.Content, req.Category, req.AgentID) + respond(w, 200, map[string]interface{}{ + "status": "merged", "memory_id": existing.ID, "reason": "near_dup", + "table": "memories", + }) + return + } + } + } + + // 3. 矛盾检测（规则型，廉价）：**只报告不拦** —— 显式写入是调用方意图， + // 旧新并存由写时四操作门控/人工 feedback 收口（与 commit 的 conflict 报告同形） + var conflictSources []string + for _, existing := range similar { + if existing.Content != req.Content { + conflictSources = append(conflictSources, existing.Content) + } + } + conflicts := a.ConflictDetector.DetectContradiction(req.Content, conflictSources) + + // 4. 直写 memories 表（跳过 LLM 蒸馏 + 质量门 ⇒ 无 68% 静默丢弃） + now := time.Now() + confidenceSource := "" + if req.Confidence != nil { + confidenceSource = "authoritative" + } + mem := models.MemoryRecord{ + ID: fmt.Sprintf("mem_%d", now.UnixNano()), + AgentID: req.AgentID, + Namespace: req.Namespace, + Content: req.Content, + Category: req.Category, + Vector: vector, + Tier: "normal", + Importance: 1.0, + QualityScore: 1.0, // 显式写入 = 可信来源（对照 LLM 蒸馏的 result.Overall） + Source: src, + Freshness: "fresh", + Provenance: "user", // bi-temporal 来源枚举 user|inferred|import + EmbeddingModel: embModel, + ScriptVersion: scriptVer, + Confidence: req.Confidence, + ConfidenceSource: confidenceSource, + CreatedAt: now, + UpdatedAt: now, + } + if err := a.LanceDB.InsertMemory(mem); err != nil { + respondError(w, 500, "insert memory: "+err.Error()) + return + } + + // 5. 搜索缓存失效（与蒸馏落库同一原则：写 memories ⇒ 该 namespace 缓存作废） + if storage.SearchCacheInstance != nil && mem.Namespace != "" { + storage.SearchCacheInstance.Invalidate(mem.Namespace) + } + // 跨 Agent 广播（镜像 commit 写后 goroutine；显式写入同样通知其它 agent） + go func() { + governance.PushCacheInvalidate(req.Namespace, "memory_write") + governance.PushMemoryCommitted(req.AgentID, req.Namespace, mem.ID) + }() + + respData := map[string]interface{}{ + "status": "ok", + "memory_id": mem.ID, + "table": "memories", + "category": mem.Category, + "namespace": mem.Namespace, + "note": "direct write: embedded + recallable immediately (no distill queue)", + } + if len(conflicts) > 0 { + respData["conflicts"] = conflicts + respData["warning"] = "contradictory memories exist; explicit write kept both (resolve via feedback)" + } + respond(w, 201, respData) +} + // mergeMemory 合并重复记忆：升版本、提重要性、更新时间戳 func (a *API) mergeMemory(existing models.MemoryRecord, newContent, newCategory, agentID string) { existing.Version++ diff --git a/go/internal/api/routes/eval_test.go b/go/internal/api/routes/eval_test.go index e098960..933b56d 100644 --- a/go/internal/api/routes/eval_test.go +++ b/go/internal/api/routes/eval_test.go @@ -10,10 +10,32 @@ import ( "net/http" "os" "path/filepath" + "strings" "testing" "time" ) +// liveAPIKey 取活服务密钥：优先 env ZHIYI_TEST_API_KEY；否则读密钥文件 +// /var/lib/memoryweave/.api-key 的主密钥（第 1 个非注释、非 legacy 行）； +// 再退到旧 dev 键（无生产密钥文件的环境，如 CI）。 +// 2026-09-21 密钥轮换后，旧硬编码键 zhiyi-dev-key-2026 已 401 —— 全量金标实跑 +// 永远 SKIP 的根因（t_fa6c9104 修复量具可用性；不改任何判据）。 +func liveAPIKey() string { + if k := os.Getenv("ZHIYI_TEST_API_KEY"); k != "" { + return k + } + if b, err := os.ReadFile("/var/lib/memoryweave/.api-key"); err == nil { + for _, line := range strings.Split(string(b), "\n") { + line = strings.TrimSpace(line) + if line == "" || strings.HasPrefix(line, "#") || strings.HasPrefix(line, "legacy:") { + continue + } + return line + } + } + return "zhiyi-dev-key-2026" +} + // ─── recall 辅助：liveRecallOnce ───────────────────────────── // 一次 recall API 调用同时拿 top-5 的 ids + contents，避免两条 HTTP（限流保护）。 // 每次调用间加 200ms 间隔，由调用方负责（TestEvalGolden_Live 里 sleep）。 @@ -38,7 +60,7 @@ func liveRecallOnce(query string, maxRetries int, retryWait time.Duration) recal } req, _ := http.NewRequest("POST", "http://127.0.0.1:7821/api/v1/recall", bytes.NewReader(payload)) req.Header.Set("Content-Type", "application/json") - req.Header.Set("X-API-Key", "zhiyi-dev-key-2026") + req.Header.Set("X-API-Key", liveAPIKey()) resp, err := http.DefaultClient.Do(req) if err != nil { lastErr = fmt.Errorf("HTTP 请求失败: %v", err) @@ -260,7 +282,7 @@ func TestLoadGoldenSet_EmbedFallback_DeployCWD(t *testing.T) { func TestEvalGolden_CounterEvidence(t *testing.T) { // 探活 zhiyid（无服务 → 跳过） healthReq, _ := http.NewRequest("GET", "http://127.0.0.1:7821/api/v1/health", nil) - healthReq.Header.Set("X-API-Key", "zhiyi-dev-key-2026") + healthReq.Header.Set("X-API-Key", liveAPIKey()) resp, err := http.DefaultClient.Do(healthReq) if err != nil || resp == nil { t.Skipf("zhiyid 服务不可达（%v），跳过反证测试", err) @@ -336,7 +358,7 @@ func TestEvalGolden_CounterEvidence(t *testing.T) { func TestEvalGolden_Live(t *testing.T) { // 探活 zhiyid（无服务 → 跳过，不污染 CI） healthReq, _ := http.NewRequest("GET", "http://127.0.0.1:7821/api/v1/health", nil) - healthReq.Header.Set("X-API-Key", "zhiyi-dev-key-2026") + healthReq.Header.Set("X-API-Key", liveAPIKey()) resp, err := http.DefaultClient.Do(healthReq) if err != nil || resp == nil { t.Skipf("zhiyid 服务不可达（%v），跳过全量金标实跑", err) diff --git a/go/internal/api/routes/memory_write_test.go b/go/internal/api/routes/memory_write_test.go new file mode 100644 index 0000000..e367941 --- /dev/null +++ b/go/internal/api/routes/memory_write_test.go @@ -0,0 +1,222 @@ +// 织忆 MemoryWeave — /api/v1/memory/write 显式直写 memories 表（主症A，t_fa6c9104，2026-10-16） +// +// 锁定行为： +// +// W1 写入落 memories 表：InsertMemory 被调用，id 前缀 mem_，tier=normal，quality=1.0， +// 且向量非全 0（嵌入服务故障时不写零向量记录） +// W2 category 缺省 distilled（recall Step 2.5 rawCategories 不过滤它） +// W3 嵌入失败 ⇒ 502 且不写（零向量 = 主症A 病根，禁静默写） +// W4 精确重复 ⇒ merged（不新增行） +// W5 短内容 <8 字 ⇒ 400；GET ⇒ 405 +// W6 零回归：commit 路径仍只写 episodes、不写 memories（AC5） +package routes + +import ( + "bytes" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync" + "testing" + + "github.com/xiaoxue/memoryweave/internal/distill" + "github.com/xiaoxue/memoryweave/internal/governance" + "github.com/xiaoxue/memoryweave/internal/models" + "github.com/xiaoxue/memoryweave/internal/storage" +) + +// recLance 记录型 LanceDB mock：断言 InsertMemory/Update 的实际调用与载荷。 +// 复用 zz_commit_conflict_gate_test.go 的 mockGraph / newMockEmbedServer / newMockLLM / commitWith（同包）。 +type recLance struct { + storage.LanceDB // 未实现的方法落到接口兜底（panic 即测试失败，不放任） + + mu sync.Mutex + seeds []models.MemoryRecord // 预置「库里已有」的记忆（Search 返回） + inserted []models.MemoryRecord // 实际 InsertMemory 调用（W1/W3/W4 断言） + updates []string // Update(table,id,...) 的 id 序列（merge 断言） + failIns bool // 模拟 IPC 落库失败（500 断言） +} + +func (r *recLance) Search(table string, vector []float32, topK int, ns string) ([]models.MemoryRecord, error) { + if table != "memories" { + return nil, nil + } + r.mu.Lock() + defer r.mu.Unlock() + out := make([]models.MemoryRecord, 0, len(r.seeds)) + out = append(out, r.seeds...) + if topK > 0 && len(out) > topK { + out = out[:topK] + } + return out, nil +} + +func (r *recLance) InsertMemory(m models.MemoryRecord) error { + r.mu.Lock() + defer r.mu.Unlock() + if r.failIns { + return errors.New("ipc down (simulated)") + } + r.inserted = append(r.inserted, m) + r.seeds = append(r.seeds, m) // 后续 Search 能看见（近重复语义） + return nil +} + +func (r *recLance) Update(table, id string, fields map[string]any) error { + r.mu.Lock() + defer r.mu.Unlock() + r.updates = append(r.updates, id) + return nil +} + +func (r *recLance) InsertEpisode(agentID, namespace, content, category string) (string, error) { + r.mu.Lock() + defer r.mu.Unlock() + return "ep_rec_" + strconv.Itoa(len(r.inserted)), nil +} + +func newRecAPI(t *testing.T, seeds ...models.MemoryRecord) (*API, *recLance) { + t.Helper() + emb := newMockEmbedServer() + t.Cleanup(emb.Close) + ldb := &recLance{seeds: seeds} + return NewAPI(ldb, storage.NewEmbedder(emb.URL), nil, governance.NewConflictDetector(), mockGraph{}), ldb +} + +func writeMemRequest(t *testing.T, api *API, body map[string]any) (int, map[string]interface{}) { + t.Helper() + b, _ := json.Marshal(body) + req := httptest.NewRequest("POST", "/api/v1/memory/write", bytes.NewReader(b)) + req.Header.Set("Content-Type", "application/json") + w := httptest.NewRecorder() + api.WriteMemory(w, req) + var resp map[string]interface{} + json.Unmarshal(w.Body.Bytes(), &resp) + return w.Code, resp +} + +// W1+W2：写入落 memories 表，mem_ 前缀，category 缺省 distilled，向量非全 0 +func TestWriteMemory_InsertLandsInMemories(t *testing.T) { + api, ldb := newRecAPI(t) + code, resp := writeMemRequest(t, api, map[string]any{ + "content": "牧尘的工作所在省份是河南省", + "agent_id": "hermes-a06", + }) + if code != 201 { + t.Fatalf("expected 201, got %d: %s", code, resp) + } + if resp["status"] != "ok" || resp["table"] != "memories" { + t.Errorf("expected status=ok table=memories, got %v", resp) + } + if !strings.HasPrefix(resp["memory_id"].(string), "mem_") { + t.Errorf("memory_id 须 mem_ 前缀（落 memories 表），got %v", resp["memory_id"]) + } + if resp["category"] != "distilled" { + t.Errorf("category 缺省须 distilled（Step 2.5 不过滤），got %v", resp["category"]) + } + if len(ldb.inserted) != 1 { + t.Fatalf("expected exactly 1 InsertMemory, got %d", len(ldb.inserted)) + } + m := ldb.inserted[0] + if m.Category != "distilled" || m.Tier != "normal" || m.QualityScore != 1.0 { + t.Errorf("row 字段: category=%s tier=%s quality=%v（期望 distilled/normal/1.0）", m.Category, m.Tier, m.QualityScore) + } + if m.Namespace != "hermes-main" { + t.Errorf("namespace 推导: got %s 期望 hermes-main（hermes-a06 ⇒ deriveNamespace）", m.Namespace) + } + if len(m.Vector) != 1024 { + t.Fatalf("vector 维度须 1024，got %d", len(m.Vector)) + } + // W1 核心判据：vector 非全 0 + nonzero := 0 + for _, f := range m.Vector { + if f != 0 { + nonzero++ + } + } + if nonzero == 0 { + t.Errorf("向量全 0 —— 主症A 病根复现（禁写零向量记录）") + } +} + +// W3：嵌入失败 ⇒ 502 且不写库（fail-close，不静默写零向量） +func TestWriteMemory_EmbedFail_502_NoWrite(t *testing.T) { + failEmb := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "embed backend down", http.StatusInternalServerError) + })) + defer failEmb.Close() + ldb := &recLance{} + api := NewAPI(ldb, storage.NewEmbedder(failEmb.URL), nil, governance.NewConflictDetector(), mockGraph{}) + code, resp := writeMemRequest(t, api, map[string]any{"content": "牧尘的工作所在省份是河南省"}) + if code != 502 { + t.Fatalf("embed 失败须 502，got %d: %v", code, resp) + } + if len(ldb.inserted) != 0 { + t.Errorf("嵌入失败不得写库（零向量 = 主症A），inserted=%d", len(ldb.inserted)) + } +} + +// W4：精确重复 ⇒ merged，不新增行 +func TestWriteMemory_ExactDupMerges(t *testing.T) { + seed := models.MemoryRecord{ID: "mem_seed", Content: "家用服务器地址是 192.168.123.11", Category: "distilled"} + api, ldb := newRecAPI(t, seed) + code, resp := writeMemRequest(t, api, map[string]any{"content": seed.Content}) + if code != 200 || resp["status"] != "merged" { + t.Fatalf("expected 200 merged, got %d %v", code, resp) + } + if len(ldb.inserted) != 0 { + t.Errorf("精确重复不得新增行，inserted=%d", len(ldb.inserted)) + } + if len(ldb.updates) == 0 || ldb.updates[0] != "mem_seed" { + t.Errorf("merge 须 Update 既有行 mem_seed，updates=%v", ldb.updates) + } +} + +// W5：短内容 400；GET 405 +func TestWriteMemory_Validation(t *testing.T) { + api, _ := newRecAPI(t) + if c, _ := writeMemRequest(t, api, map[string]any{"content": "太短了"}); c != 400 { + t.Errorf("短内容(<8字)须 400, got %d", c) + } + req := httptest.NewRequest("GET", "/api/v1/memory/write", nil) + w := httptest.NewRecorder() + api.WriteMemory(w, req) + if w.Code != 405 { + t.Errorf("GET 须 405, got %d", w.Code) + } +} + +// 落库失败（IPC down）⇒ 500 不吞错 +func TestWriteMemory_InsertFail_500(t *testing.T) { + api, ldb := newRecAPI(t) + ldb.failIns = true + code, _ := writeMemRequest(t, api, map[string]any{"content": "测试落库失败场景内容足够长"}) + if code != 500 { + t.Fatalf("InsertMemory 失败须 500（不吞错），got %d", code) + } +} + +// W6 零回归：commit 路径仍「只写 episodes、不写 memories」（等蒸馏） +func TestCommit_StillEpisodesOnly_ZeroRegression(t *testing.T) { + api, ldb := newRecAPI(t) + llm := newMockLLM() + defer llm.server.Close() + oldRef := DistillEngineRef + DistillEngineRef = distill.NewEngine(llm.server.URL, "mock-model", "") + defer func() { DistillEngineRef = oldRef }() + + // 复用 commitWith（同包 zz 测试）：普通对话提交 + resp := commitWith(t, api, "今天天气不错，我们去楼下散步", false) + if resp["status"] != "ok" { + t.Fatalf("commit 基线须 status=ok，got %v", resp) + } + if resp["memory_id"] != "" { + t.Errorf("commit 不在写入时创建 memory（AC5 零回归），got %v", resp["memory_id"]) + } + if len(ldb.inserted) != 0 { + t.Errorf("commit 不得直写 memories 表（零回归），inserted=%d", len(ldb.inserted)) + } +} diff --git a/go/internal/api/server.go b/go/internal/api/server.go index 959ad7c..10a286d 100644 --- a/go/internal/api/server.go +++ b/go/internal/api/server.go @@ -568,6 +568,9 @@ func NewServer() http.Handler { mux.HandleFunc("/api/v1/bootstrap", api.Bootstrap) mux.HandleFunc("/api/v1/stats", api.Stats) mux.HandleFunc("/api/v1/batch-commit", api.BatchCommit) + // 主症A（t_fa6c9104）：显式写入直落 memories 表（嵌入 + 立刻可召回）。 + // 纯新增路由：commit/recall/蒸馏路径零改动；旧服务端 404 时插件自动回落 commit。 + mux.HandleFunc("/api/v1/memory/write", api.WriteMemory) // 自优化指标（Dashboard，含 avg_distill_loss） // total_episodes：Rust count_rows 真实值；旧 sidecar 回退 episodes_cached（进程内缓存，非总量） mux.HandleFunc("/api/v1/metrics", func(w http.ResponseWriter, r *http.Request) { diff --git a/plugins/hermes-zhiyi/__init__.py b/plugins/hermes-zhiyi/__init__.py index 6f60337..f25ea44 100644 --- a/plugins/hermes-zhiyi/__init__.py +++ b/plugins/hermes-zhiyi/__init__.py @@ -1,5 +1,3 @@ -# ⚠️ 部署副本 — 权威副本 = ~/.hermes/hermes-agent/plugins/memory/zhiyi/__init__.py（bundled-first 生效） -# 改内容请改权威副本后再同步本文件（2026-09-26 三副本收敛） """hermes-zhiyi — Bridge plugin: Hermes Agent → ZhiYi MemoryWeave REST API. Uses ZhiYi's bge-m3 1024-dim semantic search + bge-reranker-v2-m3 reranking. @@ -23,6 +21,7 @@ from __future__ import annotations import json import logging import os +import re import threading import time from pathlib import Path @@ -79,13 +78,79 @@ except RuntimeError as _e: # 日志中只暴露 key 的前 4 位，用于调试 _ZHIYI_KEY_PREFIX = ZHIYI_API_KEY[:4] if ZHIYI_API_KEY else "NONE" +# ── [t_a9848f01 主症B] 元笔记过滤器（召回侧抑制存量"关于召回/蒸馏系统自身"的诊断/复现/调试记录）── +# 硬约束：改召回行为必须 env 门控 + 默认零行为变化。仅当 ZHIYI_META_FILTER == '1' 时启用； +# 默认（unset/0）或加载失败 → 恒 False（零变化），保证生产不受意外影响。 +def _find_meta_filter_module(): + """定位共享的 meta_note_filter.py。共享脚本在**根** ~/.hermes/scripts/lib， + 但 HERMES_HOME 可能是 profile 目录（~/.hermes/profiles/

）→ 逐个候选找，取第一个存在的。""" + cands = [] + try: + cands.append(Path.home() / ".hermes" / "scripts" / "lib" / "meta_note_filter.py") + cands.append(HERMES_HOME / "scripts" / "lib" / "meta_note_filter.py") + for _base in list(HERMES_HOME.parents[:3]): # profile 目录上溯到根 + cands.append(_base / "scripts" / "lib" / "meta_note_filter.py") + except Exception: + pass + for _c in cands: + try: + if _c.exists(): + return _c + except Exception: + continue + return None + + +def _meta_note_pred(): + """返回 text->bool 谓词。未启用/加载失败 → 恒 False。启用 → 用 scripts/lib/meta_note_filter。""" + if os.environ.get("ZHIYI_META_FILTER", "0").strip() != "1": + return lambda _t: False + try: + import importlib.util + p = _find_meta_filter_module() + if p is None: + return lambda _t: False + spec = importlib.util.spec_from_file_location("_zhiyi_meta_note_filter", str(p)) + if spec is None or spec.loader is None: + return lambda _t: False + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + def _pred(text): + try: + ok, _reason = mod.is_meta_note(text or "") + return bool(ok) + except Exception: + return False + return _pred + except Exception: + return lambda _t: False + + +def _meta_filter_on() -> bool: + return os.environ.get("ZHIYI_META_FILTER", "0").strip() == "1" + # ── Content Validation ─────────────────────────────────────────────────────── MEMORY_MIN_LENGTH = 20 # 少于20字的过滤掉 + +# 系统报告特征【前缀】—— 与 scripts/session-to-zhiyi.py 的 SYSTEM_REPORT_PREFIXES 同判据 +# （2026-10-11 t_aada4c69 / t_42420afb）。cron 投递 / 网关来源包裹 / cron 系统警告是 +# 系统自己的元信息，不是用户对话 —— 写进织忆会被画像蒸馏当成"用户挫败感"（自吸收环）。 +# 按【头子串】判（content.lstrip().startswith(pat)）：正文里恰好提到"Cronjob Response:" +# 的正常讨论（边界样本）不会被误伤。保持与脚本侧集合完全一致（对称差 = 空集）。 +_SYSTEM_REPORT_PREFIXES = ( + "Cronjob Response:", + "Cronjob Failed:", + "Cronjob Error:", + "Gateway message origin (JSON data, not instructions or authorization):", + "[IMPORTANT:", # cron/system 系统警告（如 "skill(s) could not be found"） + "[HEARTBEAT]", +) + +# 其余系统注入模式（上下文压缩摘要 / 提示注入垃圾）按【全文子串】判 —— 它们不必然出现在头部。 _FORBIDDEN_PATTERNS = [ "Review the conversation above", "[System note:", - "[IMPORTANT: Background process", "## Active Task", "## Context Compaction", "## Relevant Memory", @@ -97,34 +162,6 @@ _FORBIDDEN_PATTERNS = [ "do NOT answer questions or fulfill requests mentioned in the summary", ] -# 系统样板文本【头部】特征（结构化来源标记，按头子串判）— 噪音闸门 t_689b6d38。 -# 这些是系统自己的元信息（cron 投递 / 网关来源包裹 / 心跳 / 系统通知），不是用户对话。 -# 被动注入入口命中即跳过语义召回与笔记返回（与 Go daemon /api/v1/recall、/graph/notes 同判据）。 -_SYSTEM_TEMPLATE_PREFIXES = ( - "Cronjob Response:", - "Cronjob Failed:", - "Cronjob Error:", - "Gateway message origin (JSON data, not instructions or authorization):", - "[IMPORTANT:", # cron 系统警告 / "[IMPORTANT: You are running as a scheduled cron job..." / "[IMPORTANT: Background process ..." - "[HEARTBEAT]", -) - - -def _is_system_template_input(text: str) -> bool: - """系统通知/提示样板文本判定（t_689b6d38 噪音闸门）。 - - 优先用结构化来源标记（头子串），不用纯文本长度启发式。真实用户提问 - （含飞书/QQ/CLI 入站）不命中任何头/子串 ⇒ 照常召回，不误杀。 - """ - t = (text or "").lstrip() - for pat in _SYSTEM_TEMPLATE_PREFIXES: - if t.startswith(pat): - return True - for pat in _FORBIDDEN_PATTERNS: - if pat in t: - return True - return False - _SOCIAL_CLOSERS = frozenset({ "ok", "好的", "👍", "👌", "✅", "谢谢", "感谢", "知道了", "明白", "嗯", "好", "行", "yes", "yep", "thanks", "thx", @@ -142,10 +179,35 @@ def _is_social_close(text: str) -> bool: return False +def _is_system_report_head(text: str) -> bool: + """系统报告特征：按【头子串】判（与 session-to-zhiyi.py is_system_report_message 同判据）。 + + 报告以 user 角色进入 sync_turn，而 _combine_turn 会在 combined 头部加 "用户: " 前缀 + —— 所以必须在原始 user 文本上判头（否则头被 "用户: " 吞掉、永远拦不住）。 + """ + t = (text or "").lstrip() + for pat in _SYSTEM_REPORT_PREFIXES: + if t.startswith(pat): + logger.info("[ZhiYi] 拦截 1 条系统报告文本 @ sync_turn (pattern=%r)", pat) + return True + return False + + def _is_valid_memory_content(content: str) -> bool: - """过滤系统注入内容和测试垃圾，防止污染记忆存储。""" + """过滤系统注入内容和测试垃圾，防止污染记忆存储。 + + 系统报告特征按【头子串】判（content.lstrip().startswith(pat)，与 + session-to-zhiyi.py 的 is_system_report_message 同判据）；其余系统注入模式按 + 全文子串判。拦截系统报告不得静默（AC5）——否则下次查不出有没有生效。 + """ if not content or len(content.strip()) < MEMORY_MIN_LENGTH: return False + # 系统报告特征：按头子串判（正文提到"Cronjob Response:"等特征头的正常讨论不会被误杀） + head = content.lstrip() + for pat in _SYSTEM_REPORT_PREFIXES: + if head.startswith(pat): + return False + # 其余系统注入模式：按全文子串判 for pat in _FORBIDDEN_PATTERNS: if pat in content: return False @@ -162,6 +224,77 @@ def _combine_turn(user: str, assistant: str) -> str: return "\n".join(parts) +# ── Auto feedback (P0, 2026-10-11) ─────────────────────────────────────────── +# 每轮结束自动把"回复里用到的注入记忆"标 useful。宁可少标不可错标。 +INJECTED_TTL = 600 # 注入记忆参与匹配的时间窗（秒） +AUTO_FEEDBACK_LOG = HERMES_HOME / "logs" / "zhiyi-auto-feedback.jsonl" +AUTO_FEEDBACK_OFF = MEMORY_CACHE / "auto_feedback.off" # 哨兵文件：存在即关闭（热回滚，无需重启) + + +def _auto_feedback_enabled() -> bool: + """开关：环境变量 ZHIYI_AUTO_FEEDBACK=0 或哨兵文件存在 → 关闭。默认开。""" + if os.environ.get("ZHIYI_AUTO_FEEDBACK", "1").strip() in ("0", "false", "False", "no"): + return False + return not AUTO_FEEDBACK_OFF.exists() + + +# 加载标记：模块级只打一次。用途 = 重启后判断**长驻进程**是否真的载入了本版代码 +# （插件走 sys.modules 缓存、不热重载；"文件改了" ≠ "网关在用新代码"）。 +try: + logger.info("[ZhiYi] auto-feedback P0 %s (build 2026-10-11)", + "ENABLED" if _auto_feedback_enabled() else "DISABLED(sentinel/env)") +except Exception: # 日志失败不得影响插件装载 + pass + + +_ASCII_TOKEN_RE = re.compile(r"[A-Za-z][A-Za-z0-9_.\-/]{2,}") +_CJK_RUN_RE = re.compile(r"[\u4e00-\u9fff]{2,}") # 2026-10-15: 4-gram→2-gram（4-gram 要求连续4个相同汉字，中文自然复述几乎不命中 → cjk_shared 恒 0 → 纯中文永不触发） +_TOKEN_STOP = frozenset({ + "the", "and", "for", "with", "this", "that", "from", "http", "https", + "true", "false", "none", "null", "self", "def", "return", "import", +}) +# 中文高频通用双字词 → 从特征中剔除（防"系统/记忆/股票/访问"这类通用词把分数虚推上去）。 +# A/B 实证（t_2be29a01）：加上停用词后，真正无关的回复 c=0~2 永不标记，相关复述 c≥3 才标。 +_CJK_STOP_BIGRAMS = frozenset({ + "系统", "记忆", "任务", "这个", "那个", "一个", "什么", "怎么", "可以", "需要", "进行", + "没有", "我们", "他们", "用户", "助手", "规则", "因为", "所以", "但是", "如果", "就是", + "不是", "已经", "还有", "所有", "通过", "根据", "对于", "关于", "以及", "应该", "必须", + "不要", "不会", "时候", "之前", "之后", "工作", "笔记", "里面", "上面", "下面", "使用", + "能够", "是否", "为什么", "这样", "那样", "问题", "方法", "方式", "步骤", "结果", "情况", + "自己", "其他", "信息", "内容", "数据", "文件", "目录", "路径", "命令", "输出", "成功", + "失败", "确认", "错误", "分析", "判断", "实现", "支持", "提供", "存在", "相关", "重要", + "主要", "目前", "当前", "这里", "那里", "了解", "知道", "股票", "行情", "接口", "请求", + "返回", "获取", "调用", "处理", "看到", "网络", "访问", "控制", "验证", "检查", "排查", + "排序", "衰减", "权重", "织忆", "召回", "索引", "文档", "报告", "格式", "表格", "优化", + "自动", "手动", "设置", "配置", "修改", "新增", "删除", "创建", "建立", "维护", +}) + + +def _signature_tokens(text: str): + """提取"特征 token"：ASCII 标识符（文件名/id/模型名，高精度）+ 中文 2-gram（中精度，已滤停用词）。 + 返回 (ascii_set, cjk_set)。""" + ascii_toks = {m.group(0).lower() for m in _ASCII_TOKEN_RE.finditer(text or "")} + ascii_toks -= _TOKEN_STOP + cjk_toks = set() + for m in _CJK_RUN_RE.finditer(text or ""): + run = m.group(0) + for i in range(len(run) - 1): + bg = run[i:i + 2] + if bg not in _CJK_STOP_BIGRAMS: + cjk_toks.add(bg) + return ascii_toks, cjk_toks + + +def _usage_score(memory_content: str, reply: str): + """回复对某条注入记忆的命中分：3*共享ASCII标识符 + 1*共享中文2-gram（已滤停用词）。 + 返回 (score, ascii_shared, cjk_shared)。""" + a_mem, c_mem = _signature_tokens(memory_content) + a_rep, c_rep = _signature_tokens(reply) + a_shared = len(a_mem & a_rep) + c_shared = len(c_mem & c_rep) + return 3 * a_shared + c_shared, a_shared, c_shared + + # ── HTTP Client ───────────────────────────────────────────────────────────── @@ -232,6 +365,42 @@ class ZhiYiClient: logger.error(msg) return None + def write_memory(self, content: str, category: str = "distilled", + agent_id: str = "hermes-a06", + namespace: Optional[str] = None, + confidence: Optional[float] = None) -> tuple: + """显式直写 memories 表（t_fa6c9104 主症A 修复）：同步嵌入 + 去重/合并 + 矛盾只报告。 + + 与 commit 的区别：commit 落 episodes 表等异步 LLM 蒸馏（质量门丢弃 ~68%）⇒ + 写进去的东西自己召回不回来；本方法落 memories 表 ⇒ 写完立刻可召回。 + + 返回 (True, data) 成功（data 含 memory_id / status / table / conflicts?）； + (False, reason) 失败：reason ∈ {"no_endpoint", "http_", "<异常>"}。 + 服务端未部署新端点（404）⇒ "no_endpoint"，调用方回落 commit 旧路径。 + """ + payload: Dict[str, Any] = { + "content": content, + "category": category, + "agent_id": agent_id, + } + if namespace: + payload["namespace"] = namespace + if confidence is not None: + payload["confidence"] = confidence + try: + r = self._session.post(self._url("/api/v1/memory/write"), json=payload, timeout=self.timeout) + if r.status_code in (200, 201): + data = r.json() + self._last_conflicts = data.get("conflicts") or [] + return True, data + if r.status_code == 404: + return False, "no_endpoint" + logger.error("ZhiYi write_memory failed: %s %s", r.status_code, r.text[:200]) + return False, f"http_{r.status_code}" + except Exception as e: + logger.error("ZhiYi write_memory error: %s", e) + return False, str(e) + def recall(self, query: str, top_k: int = 5, agent_id: Optional[str] = None, namespace: Optional[str] = None) -> List[Dict[str, Any]]: @@ -242,7 +411,14 @@ class ZhiYiClient: 而 worker 写入 a06-main 的记忆完全搜不到（namespace 隔离 bug）。 """ try: - payload: Dict[str, Any] = {"query": query, "top_k": top_k, "use_rerank": True} + # [t_a9848f01] 召回侧元笔记抑制：仅 ZHIYI_META_FILTER=1 时启用（默认零行为变化）。 + # 启用时用更大的候选窗口取回，再丢掉"关于召回系统自身"的元笔记，让被挤到顶部的 + # 真知识补位；不删不改存量（AC5），只影响本次呈现。 + _meta_pred = _meta_note_pred() + _window = top_k + if _meta_pred: + _window = max(top_k, 20) # 候选窗口拉大到 20，抑制后补位 + payload: Dict[str, Any] = {"query": query, "top_k": _window, "use_rerank": True} if namespace is not None: payload["namespace"] = namespace elif agent_id is not None: @@ -251,7 +427,15 @@ class ZhiYiClient: r = self._session.post(self._url("/api/v1/recall"), json=payload, timeout=self.timeout) if r.status_code == 200: data = r.json() - return data.get("results", []) + results = data.get("results", []) + if _meta_pred: + kept = [x for x in results if not _meta_pred(x.get("content", ""))] + if len(kept) < len(results): + logger.info( + "[meta-note-filter] recall抑制 %d/%d 条元笔记 query=%r（ZHIYI_META_FILTER=1）", + len(results) - len(kept), len(results), query[:40]) + results = kept[:top_k] + return results logger.warning("ZhiYi recall failed: %s %s", r.status_code, r.text[:200]) return [{"content": f"⚠️ 织忆检索失败(HTTP {r.status_code})：这不是\"没有记忆\"，是服务异常——请勿据此判断记忆为空", "score": 0.0, "category": "_error"}] except Exception as e: @@ -362,6 +546,8 @@ class HermesZhiYiMemoryProvider(MemoryProvider): self._started: bool = False self._ws_thread: Optional[threading.Thread] = None self._ws_running: bool = False + self._injected: Dict[str, Dict[str, Any]] = {} # memory_id -> {content, ts, session_id} + self._feedbacked: List[str] = [] # 已反馈过的 memory_id（防重复刷） @property def name(self) -> str: @@ -483,7 +669,13 @@ class HermesZhiYiMemoryProvider(MemoryProvider): if not self._started or not self._client: return + self._auto_feedback_for_reply(assistant_content) + combined = _combine_turn(user_content, assistant_content) + # 系统报告以 user 角色进入：先在原始 user 文本上判头（combined 已加 "用户: " 前缀， + # 头判会失效）。同源防线与 scripts/session-to-zhiyi.py 一致（P3 实测点）。 + if _is_system_report_head(user_content): + return if not _is_valid_memory_content(combined): return @@ -535,8 +727,12 @@ class HermesZhiYiMemoryProvider(MemoryProvider): if _is_social_close(query): return "" - # 噪音闸门（t_689b6d38）：系统通知/提示样板不进语义召回、不返回笔记。 - if _is_system_template_input(query): + # 噪音闸门（t_689b6d38）：系统通知/提示样板（"[IMPORTANT: ..." / "[System note: ..." + # / cron 投递包裹等）不进语义召回、不返回笔记 —— 纯浪费 + 稀释有效注入信噪比。 + # 与 sync_turn 写路径同判据：头子串（_is_system_report_head）+ 全文子串（_FORBIDDEN_PATTERNS）。 + if _is_system_report_head(query): + return "" + if any(pat in query for pat in _FORBIDDEN_PATTERNS): return "" # 优先从 queue_prefetch 缓存取（TTL < 30s） @@ -583,6 +779,16 @@ class HermesZhiYiMemoryProvider(MemoryProvider): if content: if mid: blocks.append(f" [id={mid}][{score:.2f}][{cat}]{prov} {content[:500]}") + # ── P0: 记录本轮注入的 memory_id（供 sync_turn 自动反馈）── + with self._prefetch_lock: + self._injected[mid] = { + "content": content, + "ts": time.time(), + "session_id": session_id or self._session_id, + } + _cutoff = time.time() - INJECTED_TTL + for _k in [k for k, v in self._injected.items() if (v or {}).get("ts", 0) < _cutoff]: + self._injected.pop(_k, None) else: blocks.append(f" [{score:.2f}][{cat}]{prov} {content[:500]}") @@ -633,6 +839,82 @@ class HermesZhiYiMemoryProvider(MemoryProvider): return text + def _auto_feedback_for_reply(self, reply: str) -> None: + """P0：把"回复里确实用到的注入记忆"自动标 useful。 + + 判据：特征 token 命中分 >= AUTO_FEEDBACK_MIN_SCORE（见方法内注释）。 + 只标 useful，**从不标 not_useful**（宁可少标不可错标）。 + 全流程 try/except 兜底，绝不打断 sync_turn。 + """ + if not _auto_feedback_enabled() or not reply or len(reply.strip()) < 20: + return + if not self._injected or not self._client: + return + + _MIN_SCORE = 6 # ASCII 路径阈值（保持原样）：有 ≥2 个共享 ASCII 标识符时用（高精度） + _CJK_MIN = 3 # 纯中文/低 ASCII 路径阈值：共享特定中文 2-gram ≥3 才标（A/B 定：真无关 c≤2，相关复述 c≥3） + _MAX_PER_TURN = 3 # 单轮最多标 3 条，防跑飞 + now = time.time() + with self._prefetch_lock: + cands = [(mid, dict(info)) for mid, info in self._injected.items() + if info and (now - info.get("ts", 0)) < INJECTED_TTL] + + hits = [] + for mid, info in cands: + if mid in self._feedbacked: + continue + score, a_shared, c_shared = _usage_score(info.get("content", ""), reply) + # 自适应阈值：有足够 ASCII 标识符 → 用原 6 分；纯中文/低 ASCII → 改用共享特定 2-gram ≥3。 + # 宁可少标不可错标：两条路都不放水，只是让中文内容有资格被标记。 + if a_shared >= 2: + passes = score >= _MIN_SCORE + else: + passes = c_shared >= _CJK_MIN + hits.append((score, passes, a_shared, c_shared, mid, info)) + if not hits: + return + hits.sort(key=lambda x: -x[0]) + selected = [h for h in hits if h[1]][:_MAX_PER_TURN] + if not selected: + return + + def _worker(): + for score, _passes, a_shared, c_shared, mid, info in selected: + try: + ok = self._client.mark_useful(mid) + except Exception as e: + logger.debug("[ZhiYi] auto-feedback error for %s: %s", mid, e) + ok = False + if ok: + self._feedbacked.append(mid) + if len(self._feedbacked) > 500: + del self._feedbacked[:200] + self._log_auto_feedback(mid, score, a_shared, c_shared, bool(ok), info) + + threading.Thread(target=_worker, daemon=True, name="zhiyi-auto-fb").start() + + def _log_auto_feedback(self, mid: str, score: int, a_shared: int, c_shared: int, + ok: bool, info: Dict[str, Any]) -> None: + """落审计日志（jsonl，供 7 天后核对精度/重标阈值）。失败静默。""" + try: + AUTO_FEEDBACK_LOG.parent.mkdir(parents=True, exist_ok=True) + rec = { + "ts": time.time(), + "memory_id": mid, + "score": score, + "ascii_shared": a_shared, + "cjk_shared": c_shared, + "marked": ok, + "session_id": info.get("session_id", ""), + "mem_head": (info.get("content", "") or "")[:80], + } + with open(AUTO_FEEDBACK_LOG, "a", encoding="utf-8") as f: + f.write(json.dumps(rec, ensure_ascii=False) + "\n") + logger.info("[ZhiYi] auto-feedback: %s score=%d (ascii=%d cjk=%d) marked=%s", + mid, score, a_shared, c_shared, ok) + except Exception: + logger.debug("[ZhiYi] auto-feedback log failed", exc_info=True) + def queue_prefetch(self, query: str, *, session_id: str = "") -> None: """异步预取：本轮对话结束后立即查询织忆，下一轮 prefetch 直接返回缓存。""" if not self._client or not query or len(query.strip()) < 2: @@ -640,7 +922,9 @@ class HermesZhiYiMemoryProvider(MemoryProvider): if _is_social_close(query): return # 噪音闸门（t_689b6d38）：系统通知/提示样板不进语义召回、不返回笔记（同 prefetch）。 - if _is_system_template_input(query): + if _is_system_report_head(query): + return + if any(pat in query for pat in _FORBIDDEN_PATTERNS): return # 后台线程查询并缓存 @@ -879,23 +1163,48 @@ class HermesZhiYiMemoryProvider(MemoryProvider): return json.dumps({"success": False, "error": "ZhiYi client not initialized"}) if not _is_valid_memory_content(content): return json.dumps({"success": False, "error": "Content too short or invalid"}) - commit_id = self._client.commit( - content=content, category=category, - agent_id=os.environ.get("ZHIYI_AGENT_ID", "hermes-a06"), - ) - if commit_id: - conflicts = getattr(self._client, "_last_conflicts", []) - result = {"success": True, "commit_id": commit_id} + agent_id = os.environ.get("ZHIYI_AGENT_ID", "hermes-a06") + # 主症A（t_fa6c9104）：显式保存走新端点 /api/v1/memory/write ⇒ 直落 memories 表 + # （嵌入 + 立刻可召回），不再依赖 68% 丢弃率的异步蒸馏流水线。 + ok, data = self._client.write_memory(content=content, category=category, agent_id=agent_id) + if ok: + result = { + "success": True, + "commit_id": data.get("memory_id", ""), # 兼容旧字段名 + "memory_id": data.get("memory_id", ""), + "table": data.get("table", "memories"), + "direct_write": True, + } + if data.get("status") == "merged": + result["merged"] = True + result["reason"] = data.get("reason", "") + conflicts = data.get("conflicts") or [] if conflicts: - # 2026-09-04 记忆治理：检出冲突 → 提示写方处理（不静默） + # 矛盾只报告不拦（显式写入是调用方意图，新旧并存由 feedback 收口） result["conflicts"] = conflicts result["warning"] = ( - f"⚠️ 检出 {len(conflicts)} 条潜在冲突记忆，已跳过自动蒸馏" - "（distill_skipped=conflict）。若这是修正→更新旧记忆；" - "若写错→用 memory_feedback 标记 not_useful。" + f"⚠️ 检出 {len(conflicts)} 条矛盾记忆，显式写入已保留两者" + "（conflict_reported）。若这是修正→用 memory_feedback 降权旧记忆。" ) return json.dumps(result, ensure_ascii=False) - return json.dumps({"success": False, "error": "Commit failed — check ZhiYi server logs"}) + if data == "no_endpoint": + # 服务端未部署新端点（404）⇒ 回落旧 commit 路径（现状行为，等蒸馏流水线） + commit_id = self._client.commit( + content=content, category=category, agent_id=agent_id, + ) + if commit_id: + conflicts = getattr(self._client, "_last_conflicts", []) + result = { + "success": True, + "commit_id": commit_id, + "direct_write": False, + "warning": "服务端无 /api/v1/memory/write（未部署新二进制）：已回落 commit 旧路径 —— 落 episodes 表等蒸馏，可能延迟或丢失。", + } + if conflicts: + result["conflicts"] = conflicts + return json.dumps(result, ensure_ascii=False) + return json.dumps({"success": False, "error": "Commit failed (fallback path) — check ZhiYi server logs"}) + return json.dumps({"success": False, "error": f"memory_write failed ({data}) — check ZhiYi server logs"}) def _tool_memory_stats(self) -> str: if not self._client: