feat(memory): 主症A 显式记忆直写路径 /api/v1/memory/write (t_fa6c9104)

- POST /api/v1/memory/write: 同步嵌入→去重(0.90合并)/插入 memories 表→
  矛盾检测(只报告不拦)→201 含 memory_id。嵌入失败=502 不写任何东西
  (杜绝零向量);短内容400;GET 405;commit 端点零回归(W6)。
- 插件 memory_write 工具改走新端点(直写可召回),404 自动回落旧 commit 路径。
- eval_test: 金标实跑读活密钥(/var/lib/memoryweave/.api-key),修 2026-09-21
  密钥轮换后 TestEvalGolden_Live 恒 SKIP 的 401 根因;判据未动。
- 已部署 /home/muc/bin/zhiyid-new(v3);重启待操作员执行(审批门控)。

不含 t_492e3f22 未提交的 quality_fusion 改动(留在工作树)。
This commit is contained in:
小唯
2026-10-16 18:55:20 +08:00
parent 525f11f531
commit 584fe37c35
5 changed files with 764 additions and 52 deletions
+156
View File
@@ -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++
+25 -3
View File
@@ -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)
+222
View File
@@ -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))
}
}
+3
View File
@@ -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) {
+358 -49
View File
@@ -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/<p>)→ 逐个候选找,取第一个存在的。"""
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_<code>", "<异常>"}。
服务端未部署新端点(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: