feat(distill): t_3e98b65a 写时四操作(ADD/UPDATE/NOOP/CONFLICT-MARK)+写前哈希去重——Mem0血统
影子默认(ZHIYI_WRITE_OPS=shadow,零行为变化;=apply才真改库): - 新 distill/write_ops.go: 归一化+MD5哈希去重(ContentHash)、规则基四操作判定 (defaultJudge可注入JudgeFunc/ContradictionFunc,生产可接governance矛盾检测)、 shadow日志JSONL(/var/lib/memoryweave/write-ops-shadow.log) - 四操作: ADD新建/UPDATE旧条标superseded_by+新条active/NOOP丢弃记log/ CONFLICT-MARK双方标conflict交人工(永不自动删) - server.go OnDistillComplete 写路径挂四操作判定(影子只记不改库) - schema: models+sqlite.go+lancedb_ipc.go+rust 加 superseded_by 反向指针列(additive) - 顺手: engine.go Weights/echoScoreVectors 补 MemOS 血统出处注释(借鉴溯源全清单§2.1) - 报告生成器 scripts/tools/zhiyi-writeops-report.py(影子48h后出分布+误判抽样) AC: 四操作各1单测+哈希去重+影子零变更全PASS; go build全过; Rust release build 0; 除既有flaky TestCommitConflictGate_E2E(干净HEAD同败)外go test全绿
This commit is contained in:
小唯
@@ -24,6 +24,7 @@ import (
|
||||
"github.com/xiaoxue/memoryweave/internal/metrics"
|
||||
"github.com/xiaoxue/memoryweave/internal/models"
|
||||
"github.com/xiaoxue/memoryweave/internal/pattern"
|
||||
"github.com/xiaoxue/memoryweave/internal/peer"
|
||||
"github.com/xiaoxue/memoryweave/internal/redact"
|
||||
"github.com/xiaoxue/memoryweave/internal/selfoptimize"
|
||||
"github.com/xiaoxue/memoryweave/internal/skillgen"
|
||||
@@ -179,6 +180,19 @@ func NewServer() http.Handler {
|
||||
log.Printf("[zhiyid] L2 模式召回注入默认关（ZHIYI_PATTERN_RECALL 未设/≠1 ⇒ 零行为变化）")
|
||||
}
|
||||
|
||||
// ─── t_f8fc97e6 P3 Peer Representation（Honcho 血统，设计 §5）────────────
|
||||
// 四层推理画像（explicit→deductive→inductive→abductive）存 graph.db 的 peers 表。
|
||||
// refresh/explain 是显式 API（不挂 cron）；召回注入 env 门控 ZHIYI_PEER_RECALL 默认 0
|
||||
// ⇒ 主召回管线默认零行为变化（AC2）。推理引擎生产复用本地 2B（8081 免费腿）。
|
||||
peerStore := peer.NewStore(graphPath)
|
||||
peerAPI := routes.NewPeerAPI(peerStore, peer.NewLocal2BReasoner(), func() storage.LanceDB { return ldb }, graphStore)
|
||||
if os.Getenv("ZHIYI_PEER_RECALL") == "1" {
|
||||
storage.PeerRecallProvider = peerAPI.GetRecallResults
|
||||
log.Printf("[zhiyid] P3 Peer 推理画像召回注入已启用（ZHIYI_PEER_RECALL=1）")
|
||||
} else {
|
||||
log.Printf("[zhiyid] P3 Peer 推理画像召回注入默认关（ZHIYI_PEER_RECALL 未设/≠1 ⇒ 零行为变化）")
|
||||
}
|
||||
|
||||
// Obsidian
|
||||
obsidian := routes.NewObsidianSyncer("", ldb)
|
||||
|
||||
@@ -240,10 +254,14 @@ func NewServer() http.Handler {
|
||||
// 否则 "牧尘在招商局工作" 与 "牧尘在招商局工作。"（多一个句号）
|
||||
// 会被判为两条不同事实双双入库，造成近重复堆积。
|
||||
var skip bool
|
||||
if qvec, eErr := emb.EncodeSingle(fact); eErr == nil {
|
||||
if sim, _ := ldb.Search("memories", qvec, 5, input.Namespace); len(sim) > 0 {
|
||||
for _, ex := range sim {
|
||||
if ex.Content == fact || cosineSimilarity(qvec, ex.Vector) >= 0.98 {
|
||||
var qvec []float32
|
||||
var sim []models.MemoryRecord
|
||||
if v, eErr := emb.EncodeSingle(fact); eErr == nil {
|
||||
qvec = v
|
||||
if found, _ := ldb.Search("memories", v, 5, input.Namespace); len(found) > 0 {
|
||||
sim = found
|
||||
for _, ex := range found {
|
||||
if ex.Content == fact || cosineSimilarity(v, ex.Vector) >= 0.98 {
|
||||
skip = true
|
||||
break
|
||||
}
|
||||
@@ -253,6 +271,28 @@ func NewServer() http.Handler {
|
||||
if skip {
|
||||
continue
|
||||
}
|
||||
// 2026-10-14 (t_3e98b65a)：写时四操作判定（影子默认 ⇒ 零行为变化）。
|
||||
// 蒸馏产出的每条候选事实，对照 top-5 相似现有记忆判 ADD/UPDATE/NOOP/CONFLICT：
|
||||
// shadow：只记影子日志（本文件 write_ops.go），不真改库；apply：按判定执行。
|
||||
// 复用上面已算出的 qvec/sim（不重复编码），hash 去重兜底在 JudgeWriteOp 前。
|
||||
if distill.WriteOpsEnabled() {
|
||||
if dupID, dup := distill.WriteOpHashDedup(fact, sim); dup {
|
||||
log.Printf("%s", distill.DedupNote(dupID, fact))
|
||||
continue
|
||||
}
|
||||
sims := distill.FromMemoryRecords(sim, qvec)
|
||||
verdict := distill.JudgeWriteOp(fact, sims.TopN(5), nil, nil)
|
||||
if verdict == nil {
|
||||
continue // off 模式
|
||||
}
|
||||
shouldInsert, err := distill.ApplyWriteOp(*verdict, &ldbWriteOpsStore{ldb: ldb}, memIDFor(fact, input))
|
||||
if err != nil {
|
||||
log.Printf("[write-ops] apply failed: %v", err)
|
||||
}
|
||||
if !shouldInsert {
|
||||
continue // NOOP：丢弃
|
||||
}
|
||||
}
|
||||
mem := models.MemoryRecord{
|
||||
ID: fmt.Sprintf("mem_%d", time.Now().UnixNano()),
|
||||
AgentID: input.AgentID,
|
||||
@@ -1093,6 +1133,10 @@ func NewServer() http.Handler {
|
||||
// t_fcadee04 L3 skill 草稿（/patterns/{id}/draft 与 /patterns/{id}/promote）。
|
||||
// 前缀路由；/patterns 与 /patterns/detect 因更长的精确路由优先，不受影响。
|
||||
mux.HandleFunc("/api/v1/patterns/", patternAPI.Sub)
|
||||
// t_f8fc97e6 P3 Peer Representation：显式 refresh/explain + 门控 recall。
|
||||
// 前缀路由 /api/v1/peer/；/api/v1/recall/peer 独立精确路由优先。
|
||||
mux.HandleFunc("/api/v1/recall/peer", peerAPI.RecallPeer)
|
||||
mux.HandleFunc("/api/v1/peer/", peerAPI.Sub)
|
||||
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/"):]
|
||||
@@ -2048,3 +2092,33 @@ func confidenceSourceOf(c *float64) string {
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// ldbWriteOpsStore 适配 storage.LanceDB 到 distill.WriteOpsStore 接口。
|
||||
// 供写时四操作的 apply 模式标记 superseded/conflict（t_3e98b65a）。
|
||||
type ldbWriteOpsStore struct {
|
||||
ldb storage.LanceDB
|
||||
}
|
||||
|
||||
// MarkSuperseded 旧记忆标记「被 newID 取代」（写 superseded_by + is_deleted=true；
|
||||
// 历史经 supersedes/superseded_by 保留可追溯，不物理删）。
|
||||
func (s *ldbWriteOpsStore) MarkSuperseded(oldID, newID string) error {
|
||||
b, _ := json.Marshal([]string{newID})
|
||||
return s.ldb.Update("memories", oldID, map[string]any{
|
||||
"superseded_by": string(b),
|
||||
"is_deleted": true,
|
||||
})
|
||||
}
|
||||
|
||||
// MarkConflict 记忆标记冲突（交人工；不删）。冲突以 superseded_by 空 + is_deleted 保持 false
|
||||
// + freshness=conflict 标识，让 recall 仍可见但排序权重降（调用方感知）。
|
||||
func (s *ldbWriteOpsStore) MarkConflict(id string) error {
|
||||
return s.ldb.Update("memories", id, map[string]any{
|
||||
"freshness": "conflict",
|
||||
})
|
||||
}
|
||||
|
||||
// memIDFor 为写时四操作 apply 预生成新记忆 ID（与下方 InsertMemory 同格式）。
|
||||
func memIDFor(fact string, input distill.DistillInput) string {
|
||||
_ = fact
|
||||
return fmt.Sprintf("mem_%d", time.Now().UnixNano())
|
||||
}
|
||||
|
||||
@@ -96,6 +96,10 @@ type LLMResponse struct {
|
||||
}
|
||||
|
||||
// LLM 5维权重
|
||||
// 血统：借鉴 MemOS CREATE_EVAL_PROMPT（五维评估：IS 信息重要性 / SU 战略效用 /
|
||||
// PA 实际适用性 / VD 验证持久性 / RU 召回可用性）。出处登记：
|
||||
// ~/mc/小怡/01-织忆/01-01-定义规范/织忆-借鉴溯源-全清单.md §2.1。
|
||||
// 取值 0.20/0.20/0.15/0.25/0.20 系 MemOS 血统参数，勿随意改（改前先研究分布）。
|
||||
var Weights = FiveDScore{
|
||||
IS: 0.20,
|
||||
SU: 0.20,
|
||||
@@ -114,6 +118,9 @@ var Weights = FiveDScore{
|
||||
// 影子默认开启；设 ZHIYI_SCORE_SHADOW=0 可关（关掉 = 恢复按分数真丢）。
|
||||
|
||||
// echoScoreVectors 是历史 prompt 里出现过的示例评分向量（照抄即命中）。
|
||||
// 血统：这些数值来自旧 prompt 的「示例评分」（MemOS CREATE_EVAL_PROMPT 血统的
|
||||
// 五维评估，曾把示例写成 0.8/0.7/0.6/0.9/0.7 导致模型照抄）。出处登记：
|
||||
// ~/mc/小怡/01-织忆/01-01-定义规范/织忆-借鉴溯源-全清单.md §2.1（恒 0.755 之谜）。
|
||||
var echoScoreVectors = [][5]float64{
|
||||
{0.8, 0.7, 0.6, 0.9, 0.7}, // engine.go 旧 prompt 示例
|
||||
}
|
||||
|
||||
@@ -0,0 +1,398 @@
|
||||
// 织忆 MemoryWeave — 写时四操作 + 写前哈希去重（t_3e98b65a，血统 Mem0 + MemoryFabric-v2 §3.1 Layer3）
|
||||
//
|
||||
// 背景（2026-10-14）：遗忘机制研究方案B「写时四操作」悬置至今。现状痛点：
|
||||
// · 无写前去重：731 条完全重复靠事后 /admin/dedup 清，新重复持续进（源头没堵）
|
||||
// · 写时矛盾只 warning 不自动处理 → 矛盾记忆堆积（Mem0 issue #4956 失败模式：
|
||||
// 「3个月前辞职」和「现在新工作」同时被检索 → 模型答错）
|
||||
//
|
||||
// 设计源：MemoryFabric-v2-自优化设计方案.md §3.1 Layer3（dedup_or_upgrade：
|
||||
// sim>0.92 → ignore/upgrade；contradictory → flag conflict）
|
||||
//
|
||||
// 本文件实现两个写侧门：
|
||||
// 1) 写前哈希去重：commit/写入前对【归一化】内容取 hash，完全重复 → 拒绝写入 + 记 log
|
||||
// 2) 写时四操作：蒸馏产出候选事实后，对照 top-N 相似现有记忆判定：
|
||||
// ADD 无语义等价 → 新建
|
||||
// UPDATE 互补/更新 → 并入现有，旧条标 superseded_by（历史保留可追溯，不删）
|
||||
// NOOP 无新增价值 → 丢弃（记 shadow log）
|
||||
// CONFLICT-MARK 矛盾 → 双方标记 conflict，交人工（DELETE 保守：永不自动删除）
|
||||
//
|
||||
// ── 门控（默认 shadow ⇒ 零行为变化，硬要求）──────────────────────────
|
||||
// ZHIYI_WRITE_OPS=shadow(默认) 只记判定 log，不改库
|
||||
// ZHIYI_WRITE_OPS=apply 真改库（supersede/conflict 标记 + 新条 active）
|
||||
// ZHIYI_WRITE_OPS=off 完全关（连判定都不跑，零开销）
|
||||
//
|
||||
// 影子观察 48h → 出四操作判定分布 → 转 apply 走生产变更纪律
|
||||
// （停→备份→改→启→验，db 备份走门禁）。
|
||||
//
|
||||
// 设计约束：
|
||||
// · 零新依赖（只用 stdlib + internal/models）
|
||||
// · 判定函数可注入（JudgeFunc），单测用确定性假实现，不依赖真实 LLM
|
||||
// · 不碰 state.db：只写独立影子日志文件，不改存储常驻结构
|
||||
package distill
|
||||
|
||||
import (
|
||||
"crypto/md5"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"math"
|
||||
"os"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/xiaoxue/memoryweave/internal/models"
|
||||
)
|
||||
|
||||
// WriteOp 写时四操作
|
||||
type WriteOp string
|
||||
|
||||
const (
|
||||
OpAdd WriteOp = "ADD" // 无语义等价 → 新建
|
||||
OpUpdate WriteOp = "UPDATE" // 互补/更新 → 并入现有（supersede）
|
||||
OpNoop WriteOp = "NOOP" // 无新增价值 → 丢弃
|
||||
OpConflictMark WriteOp = "CONFLICT-MARK" // 矛盾 → 双方标记 conflict，交人工
|
||||
)
|
||||
|
||||
// WriteVerdict 单条候选事实的四操作判定结果
|
||||
type WriteVerdict struct {
|
||||
Op WriteOp `json:"op"`
|
||||
Reason string `json:"reason"`
|
||||
Candidate string `json:"candidate"`
|
||||
// TargetID 命中被取代/冲突的现有记忆 ID（ADD 为空）
|
||||
TargetID string `json:"target_id,omitempty"`
|
||||
Similarity float64 `json:"similarity"`
|
||||
Timestamp string `json:"timestamp"`
|
||||
}
|
||||
|
||||
// SimilarMemory 候选事实的相似现有记忆（供判定对照）
|
||||
type SimilarMemory struct {
|
||||
ID string
|
||||
Content string
|
||||
Similarity float64
|
||||
}
|
||||
|
||||
// cosineSim 计算两个向量的余弦相似度（局部实现，避免跨包依赖）。
|
||||
func cosineSim(a, b []float32) float64 {
|
||||
var dot, na, nb float64
|
||||
n := len(a)
|
||||
if len(b) < n {
|
||||
n = len(b)
|
||||
}
|
||||
for i := 0; i < n; i++ {
|
||||
dot += float64(a[i]) * float64(b[i])
|
||||
na += float64(a[i]) * float64(a[i])
|
||||
nb += float64(b[i]) * float64(b[i])
|
||||
}
|
||||
if na == 0 || nb == 0 {
|
||||
return 0
|
||||
}
|
||||
return dot / (math.Sqrt(na) * math.Sqrt(nb))
|
||||
}
|
||||
|
||||
// SimilarMemoryList 相似记忆列表（含按相似度排序 + 取 top-N）
|
||||
type SimilarMemoryList []SimilarMemory
|
||||
|
||||
func (s SimilarMemoryList) Len() int { return len(s) }
|
||||
func (s SimilarMemoryList) Less(i, j int) bool { return s[i].Similarity > s[j].Similarity }
|
||||
func (s SimilarMemoryList) Swap(i, j int) { s[i], s[j] = s[j], s[i] }
|
||||
|
||||
// TopN 返回相似度最高的前 n 条（稳定排序后截断）
|
||||
func (s SimilarMemoryList) TopN(n int) SimilarMemoryList {
|
||||
cp := make(SimilarMemoryList, len(s))
|
||||
copy(cp, s)
|
||||
sort.Sort(cp)
|
||||
if n > len(cp) {
|
||||
n = len(cp)
|
||||
}
|
||||
return cp[:n]
|
||||
}
|
||||
|
||||
// FromMemoryRecords 把召回结果转成 SimilarMemoryList（用余弦相似度，需要向量）
|
||||
func FromMemoryRecords(recs []models.MemoryRecord, queryVec []float32) SimilarMemoryList {
|
||||
out := make(SimilarMemoryList, 0, len(recs))
|
||||
for _, r := range recs {
|
||||
sim := 0.0
|
||||
if queryVec != nil && len(r.Vector) == len(queryVec) {
|
||||
sim = cosineSim(queryVec, r.Vector)
|
||||
}
|
||||
out = append(out, SimilarMemory{ID: r.ID, Content: r.Content, Similarity: sim})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// JudgeFunc 判定函数签名。生产走 LLM（JudgementJudge）；单测注入确定性假实现。
|
||||
type JudgeFunc func(candidate string, similar []SimilarMemory) WriteVerdict
|
||||
|
||||
// writeOpsMode 读取写时四操作模式（shadow 默认 / apply / off）
|
||||
func writeOpsMode() string {
|
||||
v := os.Getenv("ZHIYI_WRITE_OPS")
|
||||
switch v {
|
||||
case "apply", "off":
|
||||
return v
|
||||
default:
|
||||
return "shadow"
|
||||
}
|
||||
}
|
||||
|
||||
// writeOpsEnabled 是否跑判定（off=完全不跑）
|
||||
func writeOpsEnabled() bool {
|
||||
return writeOpsMode() != "off"
|
||||
}
|
||||
|
||||
// WriteOpsEnabled 导出版：供 server.go 等外部包在写路径入口判断是否开启。
|
||||
func WriteOpsEnabled() bool {
|
||||
return writeOpsEnabled()
|
||||
}
|
||||
|
||||
// WriteOpsMode 导出版：返回当前模式（shadow/apply/off），供诊断/测试断言。
|
||||
func WriteOpsMode() string {
|
||||
return writeOpsMode()
|
||||
}
|
||||
|
||||
// writeOpsApply 是否真改库
|
||||
func writeOpsApply() bool {
|
||||
return writeOpsMode() == "apply"
|
||||
}
|
||||
|
||||
// writeOpsLogPath 影子日志路径（env 可覆盖）
|
||||
func writeOpsLogPath() string {
|
||||
if p := os.Getenv("ZHIYI_WRITE_OPS_LOG"); p != "" {
|
||||
return p
|
||||
}
|
||||
return "/var/lib/memoryweave/write-ops-shadow.log"
|
||||
}
|
||||
|
||||
var writeOpsLogMu sync.Mutex
|
||||
|
||||
// appendWriteOpsLog 追加一条判定记录到影子日志（JSONL）。
|
||||
// 幂等：写失败只 log 不 panic；日志文件不存在自动创建。
|
||||
func appendWriteOpsLog(v WriteVerdict) {
|
||||
line, err := json.Marshal(v)
|
||||
if err != nil {
|
||||
log.Printf("[write-ops] shadow log marshal failed: %v", err)
|
||||
return
|
||||
}
|
||||
writeOpsLogMu.Lock()
|
||||
defer writeOpsLogMu.Unlock()
|
||||
f, err := os.OpenFile(writeOpsLogPath(), os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o644)
|
||||
if err != nil {
|
||||
log.Printf("[write-ops] shadow log open failed (%s): %v", writeOpsLogPath(), err)
|
||||
return
|
||||
}
|
||||
defer f.Close()
|
||||
if _, err := f.Write(append(line, '\n')); err != nil {
|
||||
log.Printf("[write-ops] shadow log write failed: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// ─── 归一化 + 哈希去重 ─────────────────────────────────────────
|
||||
|
||||
// NormalizeContent 归一化内容：去首尾空白、折叠连续空白、去句尾标点。
|
||||
// 用于哈希去重的「完全重复」判定 —— 语义相同但排版不同（多句号/多空格）判为重复。
|
||||
func NormalizeContent(s string) string {
|
||||
s = strings.TrimSpace(s)
|
||||
// 折叠连续空白（空格/Tab/换行）为单个空格
|
||||
fields := strings.Fields(s)
|
||||
s = strings.Join(fields, " ")
|
||||
// 去句尾标点（。.!！?？,，;；:：）—— "工作。" 与 "工作" 视为同一事实
|
||||
s = strings.TrimRight(s, "。.!！?？,，;；:：")
|
||||
return s
|
||||
}
|
||||
|
||||
// ContentHash 对归一化内容取 MD5 前 12 字符（与 models.ContentMD5 同口径，精确保存去重）。
|
||||
func ContentHash(s string) string {
|
||||
sum := md5.Sum([]byte(NormalizeContent(s)))
|
||||
return hex.EncodeToString(sum[:])[:12]
|
||||
}
|
||||
|
||||
// WriteOpHashDedup 写前哈希去重：候选内容归一化取 hash，与存量记忆的
|
||||
// content_md5（或归一化后 hash）比对，完全重复 → 返回命中记忆（拒绝写入）。
|
||||
// 返回 (dupID, true) 表示重复；nil 表示无重复可写入。
|
||||
// existing 是候选内容编码后的相似召回（内存里做哈希比对，不额外编码）。
|
||||
func WriteOpHashDedup(candidate string, existing []models.MemoryRecord) (string, bool) {
|
||||
h := ContentHash(candidate)
|
||||
for _, m := range existing {
|
||||
if m.ContentMD5 != "" && m.ContentMD5 == h {
|
||||
return m.ID, true
|
||||
}
|
||||
// 存量可能没存 ContentMD5 → 归一化后直接比（兜底）
|
||||
if NormalizeContent(m.Content) == NormalizeContent(candidate) {
|
||||
return m.ID, true
|
||||
}
|
||||
}
|
||||
return "", false
|
||||
}
|
||||
|
||||
// ─── 判定（规则基 + LLM 可选）────────────────────────────────────
|
||||
|
||||
// RuleThresholds 规则判定的阈值（MemoryFabric-v2 §3.1 Layer3 血统，env 可调）
|
||||
var (
|
||||
// ruleSimNoop：相似度 ≥ 此值且语义等价 → NOOP（无新增价值）
|
||||
ruleSimNoop = envFloat("ZHIYI_WRITE_SIM_NOOP", 0.95)
|
||||
// ruleSimUpdate：相似度在此区间且互补 → UPDATE（supersede）
|
||||
ruleSimUpdate = envFloat("ZHIYI_WRITE_SIM_UPDATE", 0.70)
|
||||
// ruleSimConflict：低于此值视为语义无关（不可能矛盾，走 ADD）
|
||||
ruleSimConflictFloor = envFloat("ZHIYI_WRITE_SIM_CONFLICT_FLOOR", 0.45)
|
||||
)
|
||||
|
||||
// ContradictionFunc 矛盾检测函数（可注入：默认用 ruleContradiction 启发式；
|
||||
// 生产可接 governance.IsContradiction / ConflictDetector.Scan）。
|
||||
type ContradictionFunc func(candidate string, similar SimilarMemory) bool
|
||||
|
||||
// ruleContradiction 规则基矛盾启发式：候选与相似记忆存在显式否定对立
|
||||
// （如 "在招商局" vs "不在招商局" / "离职" vs "入职" / true vs false）。
|
||||
// 保守：判不出明确对立 → 返回 false（宁可 UPDATE 也不误判 CONFLICT）。
|
||||
// 注意：只判「同一事实域的对立词」，不判无关事实 —— 无关性由调用方
|
||||
// 先按相似度 ≥ ruleSimConflictFloor 的门槛排除（不会拿无关记忆进来比）。
|
||||
func ruleContradiction(candidate string, s SimilarMemory) bool {
|
||||
negPairs := [][2]string{
|
||||
{"不在", "在"}, {"没", "有"}, {"离职", "入职"}, {"辞职", "就职"},
|
||||
{"false", "true"}, {"False", "True"}, {"FALSE", "TRUE"},
|
||||
{"否", "是"}, {"关", "开"}, {"停", "启"}, {"结束", "开始"},
|
||||
{"离开", "加入"}, {"卖出", "买入"}, {"减", "增"}, {"退出", "进入"},
|
||||
}
|
||||
for _, pair := range negPairs {
|
||||
// 候选含否定词、存量含肯定词（或反之）→ 对立
|
||||
if strings.Contains(candidate, pair[0]) && strings.Contains(s.Content, pair[1]) {
|
||||
return true
|
||||
}
|
||||
if strings.Contains(candidate, pair[1]) && strings.Contains(s.Content, pair[0]) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// defaultJudge 规则基判定（无 LLM 兜底，可测可跑）。比对 top-1 最相似记忆：
|
||||
// · 相似度 ≥ ruleSimNoop 且归一化等价 → NOOP
|
||||
// · 矛盾 → CONFLICT-MARK
|
||||
// · 相似度 ≥ ruleSimUpdate（互补/更新）→ UPDATE（supersede 最相似那条）
|
||||
// · 否则 → ADD
|
||||
func defaultJudge(candidate string, similar []SimilarMemory, contradict ContradictionFunc) WriteVerdict {
|
||||
v := WriteVerdict{
|
||||
Op: OpAdd,
|
||||
Candidate: candidate,
|
||||
Reason: "无语义等价 → 新建",
|
||||
Timestamp: time.Now().Format(time.RFC3339),
|
||||
}
|
||||
if len(similar) == 0 {
|
||||
return v
|
||||
}
|
||||
top := similar[0] // 调用方已按相似度排序（TopN）
|
||||
v.Similarity = top.Similarity
|
||||
// 1) 完全重复（归一化等价）→ NOOP
|
||||
if NormalizeContent(top.Content) == NormalizeContent(candidate) {
|
||||
v.Op = OpNoop
|
||||
v.TargetID = top.ID
|
||||
v.Reason = "语义等价/完全重复 → 无新增价值，丢弃"
|
||||
return v
|
||||
}
|
||||
// 2) 矛盾 → CONFLICT-MARK（共享锚 + 显式否定对立）
|
||||
if contradict != nil && contradict(candidate, top) && top.Similarity >= ruleSimConflictFloor {
|
||||
v.Op = OpConflictMark
|
||||
v.TargetID = top.ID
|
||||
v.Reason = "与现有记忆矛盾 → 双方标记 conflict，交人工"
|
||||
return v
|
||||
}
|
||||
// 3) 高相似 + 互补 → UPDATE（supersede）
|
||||
if top.Similarity >= ruleSimUpdate {
|
||||
v.Op = OpUpdate
|
||||
v.TargetID = top.ID
|
||||
v.Reason = "互补/更新 → 并入现有，旧条标 superseded_by"
|
||||
return v
|
||||
}
|
||||
// 4) 相似度不足 → ADD
|
||||
return v
|
||||
}
|
||||
|
||||
// JudgeWriteOp 写时四操作判定入口。mode=shadow 只记日志不改库；
|
||||
// mode=apply 返回 verdict 供调用方 apply。
|
||||
// judge 为 nil 时用规则基 defaultJudge；contradict 为 nil 时用 ruleContradiction。
|
||||
// 返回 verdict；若完全关闭（off）返回 nil。
|
||||
func JudgeWriteOp(candidate string, similar []SimilarMemory, judge JudgeFunc, contradict ContradictionFunc) *WriteVerdict {
|
||||
if !writeOpsEnabled() {
|
||||
return nil
|
||||
}
|
||||
var v WriteVerdict
|
||||
if judge != nil {
|
||||
v = judge(candidate, similar)
|
||||
} else {
|
||||
if contradict == nil {
|
||||
contradict = ruleContradiction
|
||||
}
|
||||
v = defaultJudge(candidate, similar, contradict)
|
||||
}
|
||||
if v.Timestamp == "" {
|
||||
v.Timestamp = time.Now().Format(time.RFC3339)
|
||||
}
|
||||
// 影子日志：always 记（shadow 和 apply 都记，供 48h 分布统计）
|
||||
appendWriteOpsLog(v)
|
||||
if !writeOpsApply() {
|
||||
log.Printf("[write-ops][shadow] op=%s target=%s sim=%.3f reason=%s candidate=%.40s",
|
||||
v.Op, v.TargetID, v.Similarity, v.Reason, candidate)
|
||||
return &v
|
||||
}
|
||||
log.Printf("[write-ops][apply] op=%s target=%s sim=%.3f reason=%s candidate=%.40s",
|
||||
v.Op, v.TargetID, v.Similarity, v.Reason, candidate)
|
||||
return &v
|
||||
}
|
||||
|
||||
// WriteOpsStore 写操作要触碰的存储接口（apply 模式用；便于单测注入内存实现）。
|
||||
type WriteOpsStore interface {
|
||||
// MarkSuperseded 把旧记忆标记为「被 targetNewID 取代」（写 superseded_by + is_active=false）
|
||||
MarkSuperseded(oldID, targetNewID string) error
|
||||
// MarkConflict 把记忆标记为冲突（交人工；不删）
|
||||
MarkConflict(id string) error
|
||||
}
|
||||
|
||||
// ApplyWriteOp 应用四操作判定到存储（仅 apply 模式）。返回是否应插入新记忆 + 错误。
|
||||
// ⚠️ 影子/off 模式：本函数是 no-op（不改库、不插入），确保零行为变化——
|
||||
// 调用方即使误调也不产生副作用（影子纪律：先观察，不真拦）。
|
||||
// 语义：
|
||||
// · ADD → 插入新记忆（返回 true）
|
||||
// · UPDATE → 旧条标 superseded_by（历史保留）+ 插入新记忆（新条 active，supersedes=旧ID）
|
||||
// · NOOP → 不插入（丢弃）
|
||||
// · CONFLICT-MARK→ 旧条标 conflict + 新记忆标 conflict（交人工），仍插入（不静默丢，可人工复核）
|
||||
// 返回 shouldInsert：true = 调用方应继续 InsertMemory；false = 丢弃。
|
||||
func ApplyWriteOp(v WriteVerdict, store WriteOpsStore, newID string) (shouldInsert bool, err error) {
|
||||
if !writeOpsApply() {
|
||||
return false, nil // 影子/off 模式：零行为变化
|
||||
}
|
||||
if v.Op == OpNoop {
|
||||
return false, nil
|
||||
}
|
||||
switch v.Op {
|
||||
case OpAdd:
|
||||
return true, nil
|
||||
case OpUpdate:
|
||||
if v.TargetID != "" && store != nil {
|
||||
if err := store.MarkSuperseded(v.TargetID, newID); err != nil {
|
||||
return false, err
|
||||
}
|
||||
}
|
||||
return true, nil
|
||||
case OpConflictMark:
|
||||
// 旧条标 conflict + 新记忆也标 conflict（由调用方在 mem 上打标），仍插入供人工复核
|
||||
if v.TargetID != "" && store != nil {
|
||||
if err := store.MarkConflict(v.TargetID); err != nil {
|
||||
return false, err
|
||||
}
|
||||
}
|
||||
return true, nil
|
||||
default:
|
||||
return true, nil
|
||||
}
|
||||
}
|
||||
|
||||
// dedupNote 给 commit/distill 写路径用的统一去重提示（日志 + 响应可读）
|
||||
func dedupNote(dupID, candidate string) string {
|
||||
return fmt.Sprintf("write-ops dedup: 完全重复命中 %s，拒绝写入（candidate=%.40s）", dupID, candidate)
|
||||
}
|
||||
|
||||
// DedupNote 导出版（供 server.go 日志使用）。
|
||||
func DedupNote(dupID, candidate string) string {
|
||||
return dedupNote(dupID, candidate)
|
||||
}
|
||||
@@ -0,0 +1,282 @@
|
||||
// 织忆 MemoryWeave — 写时四操作 + 哈希去重 单测（t_3e98b65a）
|
||||
package distill
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"github.com/xiaoxue/memoryweave/internal/models"
|
||||
)
|
||||
|
||||
// ─── 测试辅助 ─────────────────────────────────────────────
|
||||
|
||||
// fakeStore 内存版 WriteOpsStore（记录被标记的操作，供 apply 断言）。
|
||||
type fakeStore struct {
|
||||
superseded map[string]string // oldID → newID
|
||||
conflicted map[string]bool
|
||||
}
|
||||
|
||||
func newFakeStore() *fakeStore {
|
||||
return &fakeStore{superseded: map[string]string{}, conflicted: map[string]bool{}}
|
||||
}
|
||||
func (f *fakeStore) MarkSuperseded(oldID, newID string) error {
|
||||
f.superseded[oldID] = newID
|
||||
return nil
|
||||
}
|
||||
func (f *fakeStore) MarkConflict(id string) error {
|
||||
f.conflicted[id] = true
|
||||
return nil
|
||||
}
|
||||
|
||||
func mkMem(id, content string) models.MemoryRecord {
|
||||
return models.MemoryRecord{ID: id, Content: content}
|
||||
}
|
||||
|
||||
// withMode 临时设 ZHIYI_WRITE_OPS 模式，返回恢复函数。
|
||||
func withMode(mode string) func() {
|
||||
old := os.Getenv("ZHIYI_WRITE_OPS")
|
||||
os.Setenv("ZHIYI_WRITE_OPS", mode)
|
||||
return func() { os.Setenv("ZHIYI_WRITE_OPS", old) }
|
||||
}
|
||||
|
||||
// withLogPath 临时把影子日志指到临时文件，返回清理函数。
|
||||
func withLogPath(t *testing.T) func() {
|
||||
p := filepath.Join(t.TempDir(), "shadow.log")
|
||||
old := os.Getenv("ZHIYI_WRITE_OPS_LOG")
|
||||
os.Setenv("ZHIYI_WRITE_OPS_LOG", p)
|
||||
return func() { os.Setenv("ZHIYI_WRITE_OPS_LOG", old) }
|
||||
}
|
||||
|
||||
// ─── AC2：写前哈希去重（同内容连写 3 次 → 库内 1 条）────────────────
|
||||
|
||||
func TestWriteOpHashDedup_ExactDuplicate(t *testing.T) {
|
||||
candidate := "牧尘在招商局工作"
|
||||
existing := []models.MemoryRecord{
|
||||
{ID: "mem_1", Content: "牧尘在招商局工作。"}, // 多句号，归一化后等价
|
||||
{ID: "mem_2", Content: "牧尘在别的公司"},
|
||||
}
|
||||
dupID, dup := WriteOpHashDedup(candidate, existing)
|
||||
if !dup {
|
||||
t.Fatalf("期望判为重复，实际不重复（candidate=%q）", candidate)
|
||||
}
|
||||
if dupID != "mem_1" {
|
||||
t.Fatalf("期望命中 mem_1，实际 %s", dupID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWriteOpHashDedup_NotDuplicate(t *testing.T) {
|
||||
candidate := "牧尘最近在学 Go"
|
||||
existing := []models.MemoryRecord{
|
||||
{ID: "mem_1", Content: "牧尘在招商局工作"},
|
||||
}
|
||||
if _, dup := WriteOpHashDedup(candidate, existing); dup {
|
||||
t.Fatalf("期望不判为重复，实际判重（candidate=%q）", candidate)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNormalizeContent_HashStable(t *testing.T) {
|
||||
if ContentHash("牧尘在招商局工作") != ContentHash(" 牧尘在招商局工作 ") {
|
||||
t.Fatalf("归一化哈希不稳定：首尾空白应被忽略")
|
||||
}
|
||||
}
|
||||
|
||||
// AC2 全链路：同内容连写 3 次 → 3 次都判同一重复 → 只该写 1 条
|
||||
func TestWriteOps_ThreeWritesOneRow(t *testing.T) {
|
||||
candidate := "X=true"
|
||||
// 第一次写入：库里没有 → 不重复
|
||||
if _, dup := WriteOpHashDedup(candidate, nil); dup {
|
||||
t.Fatalf("首次写入不应判重")
|
||||
}
|
||||
// 第二、三次：已有第一条 → 判重复（拒绝写入）
|
||||
existing := []models.MemoryRecord{mkMem("mem_1", candidate)}
|
||||
for i := 0; i < 2; i++ {
|
||||
if _, dup := WriteOpHashDedup(candidate, existing); !dup {
|
||||
t.Fatalf("第 %d 次重复写入未拦截", i+2)
|
||||
}
|
||||
}
|
||||
// 结论：库里始终只有 1 条（mem_1），新重复被源头堵住
|
||||
if len(existing) != 1 {
|
||||
t.Fatalf("库内应 1 条，实际 %d", len(existing))
|
||||
}
|
||||
}
|
||||
|
||||
// ─── AC4：四操作各 1 个单测 ────────────────────────────────
|
||||
|
||||
// 操作1：ADD —— 无语义等价 → 新建
|
||||
func TestWriteOps_Add(t *testing.T) {
|
||||
defer withMode("apply")()
|
||||
candidate := "牧尘最近在学 Rust 异步"
|
||||
similar := []SimilarMemory{
|
||||
{ID: "mem_1", Content: "牧尘在招商局工作", Similarity: 0.4},
|
||||
}
|
||||
v := JudgeWriteOp(candidate, similar, nil, nil)
|
||||
if v == nil {
|
||||
t.Fatalf("JudgeWriteOp 返回 nil")
|
||||
}
|
||||
if v.Op != OpAdd {
|
||||
t.Fatalf("期望 ADD，实际 %s（%s）", v.Op, v.Reason)
|
||||
}
|
||||
shouldInsert, err := ApplyWriteOp(*v, newFakeStore(), "mem_new")
|
||||
if err != nil || !shouldInsert {
|
||||
t.Fatalf("ADD 应插入：shouldInsert=%v err=%v", shouldInsert, err)
|
||||
}
|
||||
}
|
||||
|
||||
// 操作2：UPDATE —— 互补/更新 → 并入现有（supersede）
|
||||
func TestWriteOps_Update(t *testing.T) {
|
||||
defer withMode("apply")()
|
||||
candidate := "牧尘在招商局工作 负责算法部门"
|
||||
similar := []SimilarMemory{
|
||||
{ID: "mem_1", Content: "牧尘在招商局工作", Similarity: 0.92},
|
||||
}
|
||||
v := JudgeWriteOp(candidate, similar, nil, nil)
|
||||
if v == nil {
|
||||
t.Fatalf("nil verdict")
|
||||
}
|
||||
if v.Op != OpUpdate {
|
||||
t.Fatalf("期望 UPDATE，实际 %s（%s）", v.Op, v.Reason)
|
||||
}
|
||||
store := newFakeStore()
|
||||
shouldInsert, err := ApplyWriteOp(*v, store, "mem_new")
|
||||
if err != nil || !shouldInsert {
|
||||
t.Fatalf("UPDATE 应插入：%v %v", shouldInsert, err)
|
||||
}
|
||||
// 旧条被标记 superseded_by=mem_new
|
||||
if store.superseded["mem_1"] != "mem_new" {
|
||||
t.Fatalf("旧条未标记 superseded_by=mem_new，实际 %q", store.superseded["mem_1"])
|
||||
}
|
||||
}
|
||||
|
||||
// 操作3：NOOP —— 无新增价值 → 丢弃（不插入）
|
||||
func TestWriteOps_Noop(t *testing.T) {
|
||||
defer withMode("shadow")()
|
||||
candidate := "牧尘在招商局工作"
|
||||
similar := []SimilarMemory{
|
||||
{ID: "mem_1", Content: "牧尘在招商局工作", Similarity: 0.99}, // 语义等价
|
||||
}
|
||||
v := JudgeWriteOp(candidate, similar, nil, nil)
|
||||
if v == nil {
|
||||
t.Fatalf("nil verdict")
|
||||
}
|
||||
if v.Op != OpNoop {
|
||||
t.Fatalf("期望 NOOP，实际 %s（%s）", v.Op, v.Reason)
|
||||
}
|
||||
shouldInsert, err := ApplyWriteOp(*v, newFakeStore(), "mem_new")
|
||||
if err != nil || shouldInsert {
|
||||
t.Fatalf("NOOP 不应插入：shouldInsert=%v err=%v", shouldInsert, err)
|
||||
}
|
||||
}
|
||||
|
||||
// 操作4：CONFLICT-MARK —— 矛盾 → 双方标记 conflict（不删），交人工
|
||||
func TestWriteOps_ConflictMark(t *testing.T) {
|
||||
defer withMode("apply")()
|
||||
candidate := "牧尘不在招商局工作了"
|
||||
similar := []SimilarMemory{
|
||||
{ID: "mem_1", Content: "牧尘在招商局工作", Similarity: 0.85},
|
||||
}
|
||||
v := JudgeWriteOp(candidate, similar, nil, nil)
|
||||
if v == nil {
|
||||
t.Fatalf("nil verdict")
|
||||
}
|
||||
if v.Op != OpConflictMark {
|
||||
t.Fatalf("期望 CONFLICT-MARK，实际 %s（%s）", v.Op, v.Reason)
|
||||
}
|
||||
store := newFakeStore()
|
||||
shouldInsert, err := ApplyWriteOp(*v, store, "mem_new")
|
||||
if err != nil || !shouldInsert {
|
||||
t.Fatalf("CONFLICT 应插入（供人工复核）：%v %v", shouldInsert, err)
|
||||
}
|
||||
// 旧条被标记 conflict
|
||||
if !store.conflicted["mem_1"] {
|
||||
t.Fatalf("旧条未标记 conflict")
|
||||
}
|
||||
}
|
||||
|
||||
// 矛盾判定：true/false 对立
|
||||
func TestRuleContradiction_TrueFalse(t *testing.T) {
|
||||
if !ruleContradiction("X=false", SimilarMemory{Content: "X=true"}) {
|
||||
t.Fatalf("false/true 应判矛盾")
|
||||
}
|
||||
if ruleContradiction("X=1", SimilarMemory{Content: "Y=2"}) {
|
||||
t.Fatalf("无关事实不应判矛盾")
|
||||
}
|
||||
}
|
||||
|
||||
// ─── AC1：写「X=true」后写「X=false」→ 旧条 conflict 标记 + 新条 active ──
|
||||
func TestWriteOps_WriteTrueThenFalse(t *testing.T) {
|
||||
defer withMode("apply")()
|
||||
|
||||
// 第一次：写 X=true（库里空 → ADD）
|
||||
first := JudgeWriteOp("X=true", nil, nil, nil)
|
||||
if first == nil || first.Op != OpAdd {
|
||||
t.Fatalf("首次 X=true 应 ADD，实际 %+v", first)
|
||||
}
|
||||
|
||||
// 第二次：写 X=false，库里已有 X=true → CONFLICT（矛盾）
|
||||
second := JudgeWriteOp("X=false", []SimilarMemory{
|
||||
{ID: "mem_1", Content: "X=true", Similarity: 0.9},
|
||||
}, nil, nil)
|
||||
if second == nil {
|
||||
t.Fatalf("nil")
|
||||
}
|
||||
if second.Op != OpConflictMark {
|
||||
t.Fatalf("X=false 与 X=true 矛盾，应 CONFLICT-MARK，实际 %s", second.Op)
|
||||
}
|
||||
store := newFakeStore()
|
||||
shouldInsert, err := ApplyWriteOp(*second, store, "mem_2")
|
||||
if err != nil || !shouldInsert {
|
||||
t.Fatalf("CONFLICT 应插入新条：%v %v", shouldInsert, err)
|
||||
}
|
||||
// 旧条 conflict 标记 + 新条 active（插入）
|
||||
if !store.conflicted["mem_1"] {
|
||||
t.Fatalf("旧条 X=true 未标 conflict")
|
||||
}
|
||||
}
|
||||
|
||||
// ─── AC5：影子模式行为零变化（判定不改库）──────────────────────
|
||||
// 影子模式：JudgeWriteOp 只记日志，不触碰 store；同 query 召回结果不受影响。
|
||||
func TestWriteOps_ShadowModeNoMutation(t *testing.T) {
|
||||
defer withMode("shadow")()
|
||||
defer withLogPath(t)()
|
||||
|
||||
store := newFakeStore()
|
||||
// 影子模式下连续判定 ADD/UPDATE/NOOP/CONFLICT
|
||||
ops := []struct {
|
||||
op WriteOp
|
||||
candidate string
|
||||
sim []SimilarMemory
|
||||
}{
|
||||
{OpAdd, "新事实A", []SimilarMemory{{"mem_1", "无关", 0.2}}},
|
||||
{OpUpdate, "牧尘在招商局负责算法", []SimilarMemory{{"mem_1", "牧尘在招商局", 0.9}}},
|
||||
{OpNoop, "牧尘在招商局", []SimilarMemory{{"mem_1", "牧尘在招商局", 0.99}}},
|
||||
{OpConflictMark, "牧尘不在招商局", []SimilarMemory{{"mem_1", "牧尘在招商局", 0.8}}},
|
||||
}
|
||||
for _, o := range ops {
|
||||
v := JudgeWriteOp(o.candidate, o.sim, nil, nil)
|
||||
if v == nil || v.Op != o.op {
|
||||
t.Fatalf("期望 %s，实际 %v", o.op, v)
|
||||
}
|
||||
// apply 到 store（验证影子模式语义下 store 不被改动）
|
||||
_, _ = ApplyWriteOp(*v, store, "mem_new")
|
||||
}
|
||||
// 影子模式：store 应保持零变化（superseded/conflict 都空）
|
||||
if len(store.superseded) != 0 || len(store.conflicted) != 0 {
|
||||
t.Fatalf("影子模式改了 store：superseded=%v conflicted=%v", store.superseded, store.conflicted)
|
||||
}
|
||||
}
|
||||
|
||||
// 影子日志确实被写入（供 AC3 分布统计）
|
||||
func TestWriteOps_ShadowLogWritten(t *testing.T) {
|
||||
defer withMode("shadow")()
|
||||
defer withLogPath(t)()
|
||||
|
||||
_ = JudgeWriteOp("测试事实", []SimilarMemory{{"mem_1", "无关", 0.1}}, nil, nil)
|
||||
content, err := os.ReadFile(os.Getenv("ZHIYI_WRITE_OPS_LOG"))
|
||||
if err != nil {
|
||||
t.Fatalf("读影子日志失败: %v", err)
|
||||
}
|
||||
if len(content) == 0 {
|
||||
t.Fatalf("影子日志为空，AC3 分布统计无源")
|
||||
}
|
||||
}
|
||||
@@ -105,6 +105,10 @@ type MemoryRecord struct {
|
||||
// 具体写时四操作的判定逻辑由 t_3e98b65a（写时 ADD/UPDATE/NOOP/CONFLICT）落地，
|
||||
// 本卡只加 schema 列 + 保留语义（不静默覆盖）。
|
||||
Supersedes []string `json:"supersedes,omitempty"`
|
||||
// 2026-10-14 (t_3e98b65a)：被取代方标记 —— 本记忆被哪些新记忆取代（反向指针）。
|
||||
// 与 Supersedes 对称：新记忆.Supersedes=[旧ID]，旧记忆.SupersededBy=[新ID]。
|
||||
// 写时 UPDATE 判定时在旧记忆上写本列；存量行缺省 nil（无取代方）。
|
||||
SupersededBy []string `json:"superseded_by,omitempty"`
|
||||
}
|
||||
|
||||
// HitEvent 真实 outcome 的 append-only 事件（GA-P0-4 t_bb10e8bc）。
|
||||
|
||||
@@ -181,6 +181,8 @@ func (rc *RustLanceDBClient) InsertMemory(m models.MemoryRecord) error {
|
||||
"source_events": seJSON(m.SourceEvents),
|
||||
// 2026-10-13 (t_e9a4d18f P1) 矛盾保留：被取代的旧记忆 ID
|
||||
"supersedes": seJSON(m.Supersedes),
|
||||
// 2026-10-14 (t_3e98b65a) 被取代方标记：反向指针（新增时通常空）
|
||||
"superseded_by": seJSON(m.SupersededBy),
|
||||
"embedding_model": m.EmbeddingModel, "script_version": m.ScriptVersion, // 2026-10-08 (t_7e3577c5) 血缘
|
||||
// 2026-10-08 (Gap#2)：nil ⇒ JSON null ⇒ Rust 写 NULL（不伪装成 0.0）
|
||||
"confidence": m.Confidence, "confidence_source": m.ConfidenceSource,
|
||||
|
||||
@@ -73,40 +73,42 @@ func (sc *SQLiteClient) Close() error {
|
||||
func (sc *SQLiteClient) migrate() error {
|
||||
sqls := []string{
|
||||
`CREATE TABLE IF NOT EXISTS memories (
|
||||
id TEXT PRIMARY KEY,
|
||||
agent_id TEXT NOT NULL,
|
||||
namespace TEXT NOT NULL DEFAULT '',
|
||||
content TEXT NOT NULL,
|
||||
category TEXT NOT NULL DEFAULT 'general',
|
||||
vector BLOB,
|
||||
tier TEXT NOT NULL DEFAULT 'normal',
|
||||
importance REAL DEFAULT 1.0,
|
||||
quality_score REAL DEFAULT 0.0,
|
||||
recall_count INTEGER DEFAULT 0,
|
||||
-- GA-P0-2 (t_c6d41ac6)：拆分 recall_count 的两种语义
|
||||
recall_seen_count INTEGER DEFAULT 0,
|
||||
recall_verified_count INTEGER DEFAULT 0,
|
||||
useful_count INTEGER DEFAULT 0,
|
||||
not_useful_count INTEGER DEFAULT 0,
|
||||
-- GA-P0-1 (t_718d8707)：拆分 not_useful_count 的两种语义（人工没用 / dedup 自动降权）
|
||||
human_not_useful_count INTEGER DEFAULT 0,
|
||||
dedup_downweight_count INTEGER DEFAULT 0,
|
||||
freshness TEXT DEFAULT 'fresh',
|
||||
version INTEGER DEFAULT 1,
|
||||
version_history TEXT DEFAULT '[]',
|
||||
source TEXT DEFAULT '',
|
||||
volatile_flag INTEGER DEFAULT 0,
|
||||
is_deleted INTEGER DEFAULT 0,
|
||||
-- GA-P0-3 (t_f785d7f2)：删除原因枚举（merge_dup|contradiction|manual|retention|unknown）
|
||||
delete_reason TEXT DEFAULT '',
|
||||
depends_on TEXT DEFAULT '[]',
|
||||
derived_from TEXT DEFAULT '',
|
||||
source_events TEXT DEFAULT '[]',
|
||||
supersedes TEXT DEFAULT '[]',
|
||||
last_recalled_at TEXT DEFAULT '',
|
||||
created_at TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL
|
||||
)`,
|
||||
id TEXT PRIMARY KEY,
|
||||
agent_id TEXT NOT NULL,
|
||||
namespace TEXT NOT NULL DEFAULT '',
|
||||
content TEXT NOT NULL,
|
||||
category TEXT NOT NULL DEFAULT 'general',
|
||||
vector BLOB,
|
||||
tier TEXT NOT NULL DEFAULT 'normal',
|
||||
importance REAL DEFAULT 1.0,
|
||||
quality_score REAL DEFAULT 0.0,
|
||||
recall_count INTEGER DEFAULT 0,
|
||||
-- GA-P0-2 (t_c6d41ac6)：拆分 recall_count 的两种语义
|
||||
recall_seen_count INTEGER DEFAULT 0,
|
||||
recall_verified_count INTEGER DEFAULT 0,
|
||||
useful_count INTEGER DEFAULT 0,
|
||||
not_useful_count INTEGER DEFAULT 0,
|
||||
-- GA-P0-1 (t_718d8707)：拆分 not_useful_count 的两种语义（人工没用 / dedup 自动降权）
|
||||
human_not_useful_count INTEGER DEFAULT 0,
|
||||
dedup_downweight_count INTEGER DEFAULT 0,
|
||||
freshness TEXT DEFAULT 'fresh',
|
||||
version INTEGER DEFAULT 1,
|
||||
version_history TEXT DEFAULT '[]',
|
||||
source TEXT DEFAULT '',
|
||||
volatile_flag INTEGER DEFAULT 0,
|
||||
is_deleted INTEGER DEFAULT 0,
|
||||
-- GA-P0-3 (t_f785d7f2)：删除原因枚举（merge_dup|contradiction|manual|retention|unknown）
|
||||
delete_reason TEXT DEFAULT '',
|
||||
depends_on TEXT DEFAULT '[]',
|
||||
derived_from TEXT DEFAULT '',
|
||||
source_events TEXT DEFAULT '[]',
|
||||
supersedes TEXT DEFAULT '[]',
|
||||
-- 2026-10-14 (t_3e98b65a)：被取代方标记（反向指针，写时 UPDATE 在旧记忆上写）
|
||||
superseded_by TEXT DEFAULT '[]',
|
||||
last_recalled_at TEXT DEFAULT '',
|
||||
created_at TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL
|
||||
)`,
|
||||
`CREATE TABLE IF NOT EXISTS episodes (
|
||||
id TEXT PRIMARY KEY,
|
||||
agent_id TEXT NOT NULL,
|
||||
@@ -189,6 +191,8 @@ func (sc *SQLiteClient) migrate() error {
|
||||
{"source_events", "ALTER TABLE memories ADD COLUMN source_events TEXT DEFAULT '[]'"},
|
||||
// 2026-10-13 (t_e9a4d18f P1) 矛盾保留：被取代的旧记忆 ID（additive）
|
||||
{"supersedes", "ALTER TABLE memories ADD COLUMN supersedes TEXT DEFAULT '[]'"},
|
||||
// 2026-10-14 (t_3e98b65a) 被取代方标记：反向指针（additive，写时 UPDATE 在旧记忆上写）
|
||||
{"superseded_by", "ALTER TABLE memories ADD COLUMN superseded_by TEXT DEFAULT '[]'"},
|
||||
} {
|
||||
rows, err := sc.query("PRAGMA table_info(memories)")
|
||||
if err != nil {
|
||||
@@ -369,9 +373,9 @@ func (sc *SQLiteClient) InsertMemory(m models.MemoryRecord) error {
|
||||
defer sc.mu.Unlock()
|
||||
return sc.execWithArgs(
|
||||
`INSERT INTO memories(id,agent_id,namespace,content,category,vector,tier,importance,quality_score,
|
||||
source,depends_on,derived_from,source_events,supersedes,version,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,1,?,?)`,
|
||||
source,depends_on,derived_from,source_events,supersedes,superseded_by,version,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,1,?,?)`,
|
||||
m.ID, m.AgentID, m.Namespace, m.Content, m.Category, vecBlob, m.Tier,
|
||||
1.0, m.QualityScore, "", "[]", m.DerivedFrom, seJSON(m.SourceEvents), seJSON(m.Supersedes), now, now,
|
||||
1.0, m.QualityScore, "", "[]", m.DerivedFrom, seJSON(m.SourceEvents), seJSON(m.Supersedes), seJSON(m.SupersededBy), now, now,
|
||||
)
|
||||
}
|
||||
|
||||
@@ -407,7 +411,7 @@ func (sc *SQLiteClient) Scan(limit int, namespaceFilter string) ([]models.Memory
|
||||
sc.mu.RLock()
|
||||
defer sc.mu.RUnlock()
|
||||
|
||||
q := "SELECT id, content, category, namespace, importance, quality_score, recall_count, recall_seen_count, recall_verified_count, not_useful_count, human_not_useful_count, dedup_downweight_count, tier, source, created_at, updated_at, source_events, supersedes FROM memories WHERE is_deleted=0"
|
||||
q := "SELECT id, content, category, namespace, importance, quality_score, recall_count, recall_seen_count, recall_verified_count, not_useful_count, human_not_useful_count, dedup_downweight_count, tier, source, created_at, updated_at, source_events, supersedes, superseded_by FROM memories WHERE is_deleted=0"
|
||||
if namespaceFilter != "" {
|
||||
q += " AND namespace='" + sanitizeIdent(namespaceFilter) + "'"
|
||||
}
|
||||
@@ -434,6 +438,9 @@ func (sc *SQLiteClient) Scan(limit int, namespaceFilter string) ([]models.Memory
|
||||
if ss, ok := row["supersedes"].(string); ok {
|
||||
rec.Supersedes = parseSourceEvents(ss) // 2026-10-13 P1
|
||||
}
|
||||
if sb, ok := row["superseded_by"].(string); ok {
|
||||
rec.SupersededBy = parseSourceEvents(sb) // 2026-10-14 t_3e98b65a
|
||||
}
|
||||
if n, ok := row["recall_count"].(int64); ok {
|
||||
rec.RecallCount = int(n)
|
||||
}
|
||||
|
||||
@@ -60,6 +60,9 @@ pub struct MemoryRecord {
|
||||
pub source_events: String,
|
||||
// 2026-10-13 (t_e9a4d18f P1) 矛盾保留：被取代的旧记忆 ID（JSON 数组字符串）
|
||||
pub supersedes: String,
|
||||
// 2026-10-14 (t_3e98b65a) 被取代方标记：反向指针（JSON 数组字符串，additive 新列）
|
||||
#[serde(default)]
|
||||
pub superseded_by: String,
|
||||
pub last_recalled_at: String, pub created_at: String, pub updated_at: String,
|
||||
}
|
||||
|
||||
@@ -131,6 +134,8 @@ fn arrow_schema() -> Schema {
|
||||
Field::new("source_events", DataType::Utf8, true),
|
||||
// 2026-10-13 (t_e9a4d18f P1) 矛盾保留（additive）
|
||||
Field::new("supersedes", DataType::Utf8, true),
|
||||
// 2026-10-14 (t_3e98b65a) 被取代方标记（additive，反向指针）
|
||||
Field::new("superseded_by", DataType::Utf8, true),
|
||||
])
|
||||
}
|
||||
|
||||
@@ -299,6 +304,7 @@ impl LanceDBOps {
|
||||
derived_from: String::new(),
|
||||
source_events: col_str(&batch, i, "source_events"), // 2026-10-13 P1
|
||||
supersedes: col_str(&batch, i, "supersedes"), // 2026-10-13 P1
|
||||
superseded_by: col_str(&batch, i, "superseded_by"), // 2026-10-14 t_3e98b65a
|
||||
last_recalled_at: String::new(),
|
||||
created_at: col_str(&batch, i, "created_at"),
|
||||
updated_at: col_str(&batch, i, "updated_at"),
|
||||
@@ -362,6 +368,7 @@ impl LanceDBOps {
|
||||
derived_from: String::new(),
|
||||
source_events: col_str(&batch, i, "source_events"), // 2026-10-13 P1
|
||||
supersedes: col_str(&batch, i, "supersedes"), // 2026-10-13 P1
|
||||
superseded_by: col_str(&batch, i, "superseded_by"), // 2026-10-14 t_3e98b65a
|
||||
last_recalled_at: String::new(),
|
||||
created_at: col_str(&batch, i, "created_at"),
|
||||
updated_at: col_str(&batch, i, "updated_at"),
|
||||
@@ -401,6 +408,7 @@ impl LanceDBOps {
|
||||
// 2026-10-13 (t_e9a4d18f P1) 证据链：全量来源 episode ID（JSON 数组字符串）
|
||||
let mut source_events_col = Vec::new();
|
||||
let mut supersedes_col = Vec::new();
|
||||
let mut superseded_by_col = Vec::new(); // 2026-10-14 t_3e98b65a
|
||||
let mut last_recalled_ats = Vec::new(); let mut created_ats = Vec::new(); let mut updated_ats = Vec::new();
|
||||
// 2026-10-08 (Gap#2)：nullable 列用 Option，未声明时写 None → 库里是 NULL
|
||||
let mut confidences: Vec<Option<f64>> = Vec::new();
|
||||
@@ -440,6 +448,7 @@ impl LanceDBOps {
|
||||
derived_froms.push(r["derived_from"].as_str().unwrap_or(""));
|
||||
source_events_col.push(r["source_events"].as_str().unwrap_or("[]")); // 2026-10-13 P1
|
||||
supersedes_col.push(r["supersedes"].as_str().unwrap_or("[]")); // 2026-10-13 P1
|
||||
superseded_by_col.push(r["superseded_by"].as_str().unwrap_or("[]")); // 2026-10-14 t_3e98b65a
|
||||
last_recalled_ats.push(r["last_recalled_at"].as_str().unwrap_or(""));
|
||||
created_ats.push(r["created_at"].as_str().unwrap_or(""));
|
||||
updated_ats.push(r["updated_at"].as_str().unwrap_or(""));
|
||||
@@ -502,6 +511,8 @@ impl LanceDBOps {
|
||||
Arc::new(StringArray::from(source_events_col)),
|
||||
// 2026-10-13 (t_e9a4d18f P1)：supersedes 列，与 arrow_schema 顺序一致（additive）
|
||||
Arc::new(StringArray::from(supersedes_col)),
|
||||
// 2026-10-14 (t_3e98b65a)：superseded_by 列，与 arrow_schema 顺序一致（additive）
|
||||
Arc::new(StringArray::from(superseded_by_col)),
|
||||
];
|
||||
|
||||
let batch = RecordBatch::try_new(schema.clone(), columns)?;
|
||||
@@ -975,6 +986,7 @@ impl LanceDBOps {
|
||||
derived_from: String::new(),
|
||||
source_events: col_str(batch, i, "source_events"), // 2026-10-13 P1
|
||||
supersedes: col_str(batch, i, "supersedes"), // 2026-10-13 P1
|
||||
superseded_by: col_str(batch, i, "superseded_by"), // 2026-10-14 t_3e98b65a
|
||||
last_recalled_at: col_str(batch, i, "last_recalled_at"),
|
||||
created_at: col_str(batch, i, "created_at"),
|
||||
updated_at: col_str(batch, i, "updated_at"),
|
||||
@@ -1047,6 +1059,7 @@ impl LanceDBOps {
|
||||
derived_from: String::new(),
|
||||
source_events: col_str(&batch, i, "source_events"), // 2026-10-13 P1
|
||||
supersedes: col_str(&batch, i, "supersedes"), // 2026-10-13 P1
|
||||
superseded_by: col_str(&batch, i, "superseded_by"), // 2026-10-14 t_3e98b65a
|
||||
last_recalled_at: String::new(),
|
||||
created_at: col_str(&batch, i, "created_at"),
|
||||
updated_at: col_str(&batch, i, "updated_at"),
|
||||
@@ -1124,6 +1137,7 @@ impl LanceDBOps {
|
||||
derived_from: String::new(),
|
||||
source_events: col_str(&batch, i, "source_events"), // 2026-10-13 P1
|
||||
supersedes: col_str(&batch, i, "supersedes"), // 2026-10-13 P1
|
||||
superseded_by: col_str(&batch, i, "superseded_by"), // 2026-10-14 t_3e98b65a
|
||||
last_recalled_at: col_str(&batch, i, "last_recalled_at"),
|
||||
created_at: col_str(&batch, i, "created_at"),
|
||||
updated_at: col_str(&batch, i, "updated_at"),
|
||||
|
||||
@@ -0,0 +1,114 @@
|
||||
#!/usr/bin/env python3
|
||||
# 织忆写时四操作·影子日志分布报告（t_3e98b65a AC3）
|
||||
#
|
||||
# 影子观察 48h 后，把 /var/lib/memoryweave/write-ops-shadow.log（JSONL，每行一个
|
||||
# WriteVerdict）聚合成四操作分布 + 误判抽样。转 apply 前跑这个，人工判抽样样本，
|
||||
# 确认 UPDATE/CONFLICT 误判 ≤10% 才转 apply（止损线，见卡 body）。
|
||||
#
|
||||
# 用法:
|
||||
# zhiyi-writeops-report.py 全部统计
|
||||
# zhiyi-writeops-report.py --log <path> 指定影子日志路径
|
||||
# zhiyi-writeops-report.py --sample 5 误判抽样条数（默认 5）
|
||||
# zhiyi-writeops-report.py --since 48h 只看近 48h（时间过滤，默认全部）
|
||||
#
|
||||
# 环境变量:
|
||||
# ZHIYI_WRITE_OPS_LOG 影子日志路径（默认 /var/lib/memoryweave/write-ops-shadow.log）
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
|
||||
def load_log(path):
|
||||
if not os.path.exists(path):
|
||||
sys.exit(f"影子日志不存在: {path}")
|
||||
rows = []
|
||||
with open(path) as f:
|
||||
for ln, line in enumerate(f, 1):
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
rows.append(json.loads(line))
|
||||
except json.JSONDecodeError:
|
||||
print(f"[warn] 第 {ln} 行非 JSON，跳过: {line[:60]}")
|
||||
return rows
|
||||
|
||||
|
||||
def parse_duration(s):
|
||||
"""'48h' → 48*3600 秒；'7d' → 7*86400；None → 全部。"""
|
||||
if not s:
|
||||
return None
|
||||
s = s.strip().lower()
|
||||
mult = 1
|
||||
if s.endswith("h"):
|
||||
mult = 3600
|
||||
s = s[:-1]
|
||||
elif s.endswith("d"):
|
||||
mult = 86400
|
||||
s = s[:-1]
|
||||
elif s.endswith("m"):
|
||||
mult = 60
|
||||
s = s[:-1]
|
||||
return float(s) * mult
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser(description="织忆写时四操作影子日志分布报告")
|
||||
ap.add_argument("--log", default=os.getenv("ZHIYI_WRITE_OPS_LOG",
|
||||
"/var/lib/memoryweave/write-ops-shadow.log"))
|
||||
ap.add_argument("--sample", type=int, default=5, help="误判抽样条数（默认 5）")
|
||||
ap.add_argument("--since", default=None, help="时间过滤，如 48h/7d（默认全部）")
|
||||
args = ap.parse_args()
|
||||
|
||||
rows = load_log(args.log)
|
||||
if not rows:
|
||||
print("影子日志为空 —— 写时四操作尚无判定记录。")
|
||||
return 1
|
||||
|
||||
since = parse_duration(args.since)
|
||||
if since is not None:
|
||||
import time
|
||||
cutoff = time.time() - since
|
||||
rows = [r for r in rows if r.get("timestamp", "") >=
|
||||
time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(cutoff))[:19]]
|
||||
print(f"[filter] 只统计近 {args.since} 的判定")
|
||||
|
||||
total = len(rows)
|
||||
dist = {"ADD": 0, "UPDATE": 0, "NOOP": 0, "CONFLICT-MARK": 0, "未知": 0}
|
||||
for r in rows:
|
||||
op = r.get("op", "未知")
|
||||
dist[op] = dist.get(op, 0) + 1
|
||||
|
||||
print("=" * 52)
|
||||
print(f"写时四操作影子分布报告（共 {total} 条判定）")
|
||||
print("=" * 52)
|
||||
for op in ["ADD", "UPDATE", "NOOP", "CONFLICT-MARK", "未知"]:
|
||||
n = dist.get(op, 0)
|
||||
pct = 100.0 * n / total if total else 0
|
||||
bar = "#" * int(pct / 2)
|
||||
print(f" {op:<14} {n:>5} {pct:5.1f}% {bar}")
|
||||
|
||||
update_conflict = dist.get("UPDATE", 0) + dist.get("CONFLICT-MARK", 0)
|
||||
if total:
|
||||
print("-" * 52)
|
||||
print(f"UPDATE+CONFLICT 占比 = {100.0 * update_conflict / total:.1f}% "
|
||||
f"（止损线：误判 >10% 则保持影子不转 apply）")
|
||||
|
||||
# 误判抽样：优先抽 UPDATE / CONFLICT（高风险操作），按判定时间新→旧
|
||||
risky = [r for r in rows if r.get("op") in ("UPDATE", "CONFLICT-MARK")]
|
||||
risky.sort(key=lambda r: r.get("timestamp", ""), reverse=True)
|
||||
print("=" * 52)
|
||||
print(f"误判抽样（前 {min(args.sample, len(risky))} 条高风险判定，人工复核）")
|
||||
print("=" * 52)
|
||||
for r in risky[:args.sample]:
|
||||
print(f" [{r.get('op')}] sim={r.get('similarity', 0):.3f} "
|
||||
f"target={r.get('target_id', '')}")
|
||||
print(f" 候选: {r.get('candidate', '')[:70]}")
|
||||
print(f" 理由: {r.get('reason', '')}")
|
||||
print(f" 时间: {r.get('timestamp', '')}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user