Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f0cd747873 | ||
|
|
fe8126dd85 | ||
|
|
0634e4eb1d | ||
|
|
ba3ccd3e12 | ||
|
|
e98565f0c9 | ||
|
|
90af029c16 |
@@ -12,9 +12,21 @@ import (
|
||||
// DistillEngineRef 全局蒸馏引擎引用(server.go 注入)
|
||||
var DistillEngineRef *distill.Engine
|
||||
|
||||
// CommitMeta 2026-10-08 (Gap#2):commit 时写入方携带的元数据,透传到蒸馏产物 MemoryRecord。
|
||||
// 集中成一个结构体而非继续加位置参数——8 个位置参数再加 2 个时,「忘记传」是必然的
|
||||
// (Gap#2 的接线缺口正是这类 bug:字段加了、没人传)。
|
||||
type CommitMeta struct {
|
||||
Source string // 出处(笔记路径等)→ MemoryRecord.Source
|
||||
EmbeddingModel string // 血缘 → MemoryRecord.EmbeddingModel
|
||||
ScriptVersion string // 血缘 → MemoryRecord.ScriptVersion
|
||||
Confidence *float64 // 信任等级(写入方显式声明,不从 source 反推);nil=未声明
|
||||
ConfidenceSource string // authoritative | estimated
|
||||
}
|
||||
|
||||
// AutoDistillTrigger commit 后自动触发蒸馏流水线
|
||||
// 2026-10-08 (t_7e3577c5):新增 embedModel/scriptVer 血缘字段透传
|
||||
func AutoDistillTrigger(episodeID string, content string, category string, namespace string, agentID string, source string, embedModel string, scriptVer string) {
|
||||
// 2026-10-08 (Gap#2):改为 CommitMeta 结构体,新增 confidence 透传
|
||||
func AutoDistillTrigger(episodeID string, content string, category string, namespace string, agentID string, meta CommitMeta) {
|
||||
// 1. 硬规则过滤
|
||||
if !distill.HardRulesPass(content) {
|
||||
return
|
||||
@@ -28,9 +40,11 @@ func AutoDistillTrigger(episodeID string, content string, category string, names
|
||||
Category: distill.Category(category),
|
||||
Namespace: namespace,
|
||||
AgentID: agentID,
|
||||
Source: source,
|
||||
EmbeddingModel: embedModel,
|
||||
ScriptVersion: scriptVer,
|
||||
Source: meta.Source,
|
||||
EmbeddingModel: meta.EmbeddingModel,
|
||||
ScriptVersion: meta.ScriptVersion,
|
||||
Confidence: meta.Confidence,
|
||||
ConfidenceSource: meta.ConfidenceSource,
|
||||
})
|
||||
} else {
|
||||
// 降级: 关键词提取
|
||||
|
||||
@@ -115,6 +115,18 @@ func (a *API) Commit(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
log.Printf("[srcprobe] commit 收到 source=%q(顶层), metadata=%v → 解析后 src=%q", req.Source, req.Metadata, src)
|
||||
|
||||
// 2026-10-08 (Gap#2):信任等级——**写入方在 commit 时显式声明**,不从 source 反推。
|
||||
// 原因:source 已被挪用为「出处」(wiki:<路径>),而 confidence 是正交的「内容可不可信」。
|
||||
// 声明了才写(ConfidenceSource=authoritative);没声明保持 NULL → 读路径 COALESCE→1.0。
|
||||
var confidence *float64
|
||||
confidenceSource := ""
|
||||
if req.Metadata != nil {
|
||||
if f, ok := req.Metadata["confidence"].(float64); ok {
|
||||
confidence = &f
|
||||
confidenceSource = "authoritative"
|
||||
}
|
||||
}
|
||||
|
||||
// 2026-10-08 (t_7e3577c5):血缘字段——metadata 携带(vault-ingest 等脚本发 embedding_model/script_version)。
|
||||
// 缺省留空,server 蒸馏落库时兜底(embedding_model→bge-m3,script_version→unknown)。
|
||||
embedModel, scriptVer := "", ""
|
||||
@@ -123,6 +135,15 @@ func (a *API) Commit(w http.ResponseWriter, r *http.Request) {
|
||||
scriptVer, _ = req.Metadata["script_version"].(string)
|
||||
}
|
||||
|
||||
// 2026-10-08 (Gap#2):把 commit 元数据收成一个结构体传给蒸馏(防「字段加了没人传」)
|
||||
commitMeta := CommitMeta{
|
||||
Source: src,
|
||||
EmbeddingModel: embedModel,
|
||||
ScriptVersion: scriptVer,
|
||||
Confidence: confidence,
|
||||
ConfidenceSource: confidenceSource,
|
||||
}
|
||||
|
||||
// t_3fbb6cbd:引用可复算旁路观测(阶段1,开关 ZHIYI_COMMIT_PROVENANCE_OBSERVE 默认关)。
|
||||
// 异步 goroutine,不阻塞写路径、不改响应、不拒写入。
|
||||
go observeCommitProvenance(req.AgentID, req.Content, req.Metadata)
|
||||
@@ -152,7 +173,7 @@ func (a *API) Commit(w http.ResponseWriter, r *http.Request) {
|
||||
a.mergeMemory(existing, req.Content, req.Category, req.AgentID)
|
||||
// 合并后仍需触发蒸馏:让 t6 等冲突检测器有机会比对这条记忆,
|
||||
// 且 exact_dup 可能是 namespace 相同但实际语义冲突(如"在"vs"不在")。
|
||||
go AutoDistillTrigger(epID, req.Content, req.Category, req.Namespace, req.AgentID, src, embedModel, scriptVer)
|
||||
go AutoDistillTrigger(epID, req.Content, req.Category, req.Namespace, req.AgentID, commitMeta)
|
||||
respond(w, 200, map[string]string{
|
||||
"episode_id": epID, "memory_id": existing.ID,
|
||||
"status": "merged", "reason": "exact_dup",
|
||||
@@ -166,7 +187,7 @@ func (a *API) Commit(w http.ResponseWriter, r *http.Request) {
|
||||
if eErr == nil && cosineSimilarity(vector, ev) >= 0.98 {
|
||||
a.mergeMemory(existing, req.Content, req.Category, req.AgentID)
|
||||
// 与 exact_dup 同理:合并后仍需触发蒸馏,让冲突检测器有机会比对。
|
||||
go AutoDistillTrigger(epID, req.Content, req.Category, req.Namespace, req.AgentID, src, embedModel, scriptVer)
|
||||
go AutoDistillTrigger(epID, req.Content, req.Category, req.Namespace, req.AgentID, commitMeta)
|
||||
respond(w, 200, map[string]string{
|
||||
"episode_id": epID, "memory_id": existing.ID,
|
||||
"status": "merged", "reason": "near_dup",
|
||||
@@ -232,9 +253,9 @@ func (a *API) Commit(w http.ResponseWriter, r *http.Request) {
|
||||
// 2026-09-18 (t_a54aae79 #14): 客户端显式确认修正/新事实 → 仍触发蒸馏
|
||||
respData["conflicts"] = conflicts
|
||||
respData["distill_note"] = "conflict detected but resolve_conflict=true: proceeding with auto-distill"
|
||||
go AutoDistillTrigger(epID, req.Content, req.Category, req.Namespace, req.AgentID, src, embedModel, scriptVer)
|
||||
go AutoDistillTrigger(epID, req.Content, req.Category, req.Namespace, req.AgentID, commitMeta)
|
||||
} else {
|
||||
go AutoDistillTrigger(epID, req.Content, req.Category, req.Namespace, req.AgentID, src, embedModel, scriptVer)
|
||||
go AutoDistillTrigger(epID, req.Content, req.Category, req.Namespace, req.AgentID, commitMeta)
|
||||
}
|
||||
respond(w, 201, respData)
|
||||
}
|
||||
|
||||
@@ -220,6 +220,9 @@ func NewServer() http.Handler {
|
||||
// embedding_model→bge-m3(实际嵌入模型),script_version→unknown。
|
||||
EmbeddingModel: orDefault(input.EmbeddingModel, "bge-m3"),
|
||||
ScriptVersion: orDefault(input.ScriptVersion, "unknown"),
|
||||
// 2026-10-08 (Gap#2):信任等级透传到产物(nil 保持 NULL,不伪装 0.0)
|
||||
Confidence: input.Confidence,
|
||||
ConfidenceSource: input.ConfidenceSource,
|
||||
}
|
||||
log.Printf("[srcprobe] 蒸馏落库 episode=%s input.Source=%q → mem.Source=%q", input.EpisodeID, input.Source, mem.Source)
|
||||
if err := ldb.InsertMemory(mem); err != nil {
|
||||
@@ -242,8 +245,13 @@ func NewServer() http.Handler {
|
||||
conflictExisting := make([]map[string]interface{}, 0)
|
||||
for _, m := range existingMems {
|
||||
conflictExisting = append(conflictExisting, map[string]interface{}{
|
||||
"content": m.Content,
|
||||
"entities": entityNames, // 用当前内容提取的实体做交叉比较
|
||||
"content": m.Content,
|
||||
// 2026-10-09 修:此处原先填的是**新内容的实体**(同一个 entityNames 切片),
|
||||
// 导致 Scan 的实体预筛 e1==e2 恒真、彻底失效 —— 每条候选都跑全量比对,
|
||||
// 无关候选被纳入后 IsContradiction 误报(/var/lib/memoryweave/
|
||||
// conflicts-false-positive-*.json 现存 106 条已判假阳性即此产物)。
|
||||
// 改为**每条候选自己的实体**(从其正文抽取,与其它已存记忆同口径)。
|
||||
"entities": governance.ExtractEntities(m.Content),
|
||||
})
|
||||
}
|
||||
for _, c := range conflictDetector.Scan(input.Content, entityNames, conflictExisting) {
|
||||
@@ -1105,6 +1113,8 @@ func NewServer() http.Handler {
|
||||
AgentID string `json:"agent_id"`
|
||||
// 2026-09-26:手动蒸馏也支持带出处(便于补跑 wiki 批时保留笔记路径)
|
||||
Source string `json:"source"`
|
||||
// 2026-10-08 (Gap#2):手动蒸馏同样可显式声明信任等级(补跑存量时用)
|
||||
Confidence *float64 `json:"confidence"`
|
||||
}
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
http.Error(w, "invalid request body", http.StatusBadRequest)
|
||||
@@ -1121,7 +1131,11 @@ func NewServer() http.Handler {
|
||||
req.AgentID = "hermes-a06"
|
||||
}
|
||||
epID := fmt.Sprintf("ep_manual_%d", time.Now().UnixNano())
|
||||
routes.AutoDistillTrigger(epID, req.Content, req.Category, req.Namespace, req.AgentID, req.Source, "", "")
|
||||
routes.AutoDistillTrigger(epID, req.Content, req.Category, req.Namespace, req.AgentID, routes.CommitMeta{
|
||||
Source: req.Source,
|
||||
Confidence: req.Confidence,
|
||||
ConfidenceSource: confidenceSourceOf(req.Confidence),
|
||||
})
|
||||
respondJSON(w, 200, map[string]string{"status": "queued", "episode_id": epID})
|
||||
})
|
||||
// 蒸馏队列状态端点(G8-G9 配套)
|
||||
@@ -1917,3 +1931,12 @@ func orDefault(v, def string) string {
|
||||
}
|
||||
return v
|
||||
}
|
||||
|
||||
// confidenceSourceOf 2026-10-08 (Gap#2):声明了信任等级 ⇒ authoritative;未声明 ⇒ 空
|
||||
// (不写 "estimated"——estimated 专指「事后回填的猜测值」,语义上不该由写入路径产生)。
|
||||
func confidenceSourceOf(c *float64) string {
|
||||
if c != nil {
|
||||
return "authoritative"
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
@@ -16,6 +16,7 @@ import (
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
"unicode/utf8"
|
||||
|
||||
"github.com/xiaoxue/memoryweave/internal/metrics"
|
||||
"github.com/xiaoxue/memoryweave/internal/selfoptimize"
|
||||
@@ -47,6 +48,10 @@ type DistillInput struct {
|
||||
// 2026-10-08 (t_7e3577c5):血缘字段——透传到蒸馏产物 MemoryRecord 对应列
|
||||
EmbeddingModel string
|
||||
ScriptVersion string
|
||||
// 2026-10-08 (Gap#2):信任等级——写入方在 commit 时显式声明,跨异步蒸馏跳透传到
|
||||
// 产物 MemoryRecord.Confidence。不从 source 反推(source 是管道,confidence 是信任,正交)。
|
||||
Confidence *float64
|
||||
ConfidenceSource string
|
||||
}
|
||||
|
||||
// DistillResult 蒸馏输出
|
||||
@@ -847,8 +852,33 @@ func backoffDelay(attempt int, initial, max time.Duration, r *rand.Rand) time.Du
|
||||
//
|
||||
// 注意:补齐后得到的可能是「截断前已完整的那些条目」,而非完整 facts——这正是容错目标
|
||||
// (保留能救的部分,而不是把整条 episode 当空结果丢弃)。
|
||||
// trimPartialUTF8 去掉结尾处被按字节切断的半个 UTF-8 字符。
|
||||
//
|
||||
// 背景:LLM 输出被 max_tokens 截断时,切断点可能落在多字节字符中间,尾部残留
|
||||
// 0xE4-0xE9 这类「中文 UTF-8 首字节」。
|
||||
//
|
||||
// 实测边界(2026-10-09;216 个截断点 × 4 种形态:数组对象 / 行协议 / 单对象 /
|
||||
// 裸数组字串):解析结局不受影响,因为 encoding/json 容忍字符串内部的非法字节
|
||||
// (解码时替换为 U+FFFD),加不加本函数逐点 ok/valid 签名完全一致;受影响的是
|
||||
// 产物内容,不加本函数时字符中间截断救回的结果里仍带残字节(数组对象 73/73 点、
|
||||
// 行协议 42/73 点),它们随后被 Unmarshal 静默替换成 U+FFFD 写进记忆库,加上本
|
||||
// 函数后残余为 0。
|
||||
//
|
||||
// 即:本函数是产物净化(不让脏字节入库),不是解析可用性的修复。切记不要用
|
||||
// 「加了它 json.Valid 才通过」当判据 —— 实测该说法不成立。
|
||||
func trimPartialUTF8(s string) string {
|
||||
for len(s) > 0 {
|
||||
r, size := utf8.DecodeLastRuneInString(s)
|
||||
if r != utf8.RuneError || size > 1 {
|
||||
return s
|
||||
}
|
||||
s = s[:len(s)-1]
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
func recoverTruncatedJSON(s string) (string, bool) {
|
||||
s = strings.TrimSpace(s)
|
||||
s = trimPartialUTF8(strings.TrimSpace(s))
|
||||
if s == "" {
|
||||
return "", false
|
||||
}
|
||||
|
||||
@@ -0,0 +1,131 @@
|
||||
package distill
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
"unicode/utf8"
|
||||
)
|
||||
|
||||
// ─── trimPartialUTF8 单元判据 ────────────────────────────────────
|
||||
|
||||
// 完整的多字节串必须原样返回(不许误削)
|
||||
func TestTrimPartialUTF8_KeepsValidStrings(t *testing.T) {
|
||||
cases := []string{
|
||||
"",
|
||||
"abc",
|
||||
"中文",
|
||||
"中文abc",
|
||||
"事实:蒸馏模型日限计数器跨天重置",
|
||||
"emoji 😀 结尾",
|
||||
"半角括号]和}",
|
||||
}
|
||||
for _, in := range cases {
|
||||
if got := trimPartialUTF8(in); got != in {
|
||||
t.Errorf("完整串被改动:%q → %q", in, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 尾部半截多字节字符必须被削掉,且结果必须是合法 UTF-8
|
||||
func TestTrimPartialUTF8_StripsTrailingPartialBytes(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
in string
|
||||
want string
|
||||
}{
|
||||
{"中文尾+1字节残", "第一条记录中\xe4", "第一条记录中"},
|
||||
{"中文尾+2字节残", "另一条中文\xe4\xb8", "另一条中文"},
|
||||
{"纯残字节", "\xe4\xe4", ""},
|
||||
{"ascii后跟残字节", "abc\xe4\xb8", "abc"},
|
||||
}
|
||||
for _, c := range cases {
|
||||
got := trimPartialUTF8(c.in)
|
||||
if got != c.want {
|
||||
t.Errorf("%s: got %q, want %q", c.name, got, c.want)
|
||||
}
|
||||
if !utf8.ValidString(got) {
|
||||
t.Errorf("%s: 结果仍非合法 UTF-8: %q", c.name, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ─── 端到端:救回结果的产物必须是「干净 UTF-8」 ──────────────────────
|
||||
|
||||
func midCharSample() string {
|
||||
return `[{"fact":"第一条中文记录"},{"fact":"另一条中文` + "\xe4\xb8"
|
||||
}
|
||||
|
||||
// 真正被修的性质:救回的产物里**不残留**半个 UTF-8 字符。
|
||||
// (注意:判据不是「加了它 json.Valid 才通过」—— encoding/json 本来就容忍
|
||||
// 字符串内部的非法字节,实测加不加结局一致;受影响的是产物内容。)
|
||||
func TestRecoverTruncatedJSON_OutputIsCleanUTF8(t *testing.T) {
|
||||
in := midCharSample()
|
||||
if utf8.ValidString(in) {
|
||||
t.Fatal("样本本身应是非法 UTF-8(截断在字符中间),否则用例没在测目标场景")
|
||||
}
|
||||
|
||||
out, ok := recoverTruncatedJSON(in)
|
||||
if !ok {
|
||||
t.Fatalf("✗ 带残字节的截断输出未被救回")
|
||||
}
|
||||
if !json.Valid([]byte(out)) {
|
||||
t.Fatalf("✗ 救回结果不是合法 JSON: %q", out)
|
||||
}
|
||||
if !utf8.ValidString(out) {
|
||||
t.Fatalf("✗ 救回产物残留非法 UTF-8 字节(会被下游静默替换成 U+FFFD 入库): %q", out)
|
||||
}
|
||||
var arr []map[string]string
|
||||
if err := json.Unmarshal([]byte(out), &arr); err != nil {
|
||||
t.Fatalf("✗ 救回结果无法反序列化: %v", err)
|
||||
}
|
||||
for _, m := range arr {
|
||||
for k, v := range m {
|
||||
if strings.ContainsRune(v, utf8.RuneError) {
|
||||
t.Errorf("✗ 字段 %s 含替换字符 U+FFFD(说明残字节漏进了内容): %q", k, v)
|
||||
}
|
||||
}
|
||||
}
|
||||
t.Logf("✓ 救回 %d 条且产物为干净 UTF-8:%q", len(arr), out)
|
||||
}
|
||||
|
||||
// 反证:不削残字节时,同样补引号+补括号得到的产物**含非法字节**。
|
||||
// 若这条断言失败,说明本组用例对该修复没有鉴别力。
|
||||
func TestTrimPartialUTF8_IsNecessaryForCleanOutput(t *testing.T) {
|
||||
in := midCharSample()
|
||||
naive := strings.TrimSpace(in) + `"}]` // 模拟旧行为:不削残字节,直接补引号+补括号
|
||||
if utf8.ValidString(naive) {
|
||||
t.Fatalf("反证失败:不削残字节的产物竟是合法 UTF-8 ⇒ 本用例无鉴别力")
|
||||
}
|
||||
t.Logf("✓ 反证成立:不削残字节的产物含非法字节(%q)", naive)
|
||||
}
|
||||
|
||||
// 不漏点:在一段真实形态输出的**每个**截断点上,只要救回成功,产物就必须是干净 UTF-8
|
||||
func TestRecoverTruncatedJSON_AllCutPointsClean(t *testing.T) {
|
||||
samples := map[string]string{
|
||||
"数组对象": `[{"fact":"蒸馏模型日限计数器跨天重置后应重新计数","tags":["蒸馏"]},` +
|
||||
`{"fact":"来源加权λ默认关闭,存量confidence全为NULL","tags":["召回"]}]`,
|
||||
"行协议": `{"fact":"蒸馏模型日限计数器跨天重置后应重新计数","tags":["蒸馏"]}` + "\n" +
|
||||
`{"fact":"来源加权λ默认关闭,存量confidence全为NULL","tags":["召回"]}`,
|
||||
}
|
||||
for name, full := range samples {
|
||||
okCount, dirty := 0, 0
|
||||
for cut := 1; cut <= len(full); cut++ {
|
||||
out, ok := recoverTruncatedJSON(full[:cut])
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
okCount++
|
||||
if !utf8.ValidString(out) {
|
||||
dirty++
|
||||
if dirty <= 2 {
|
||||
t.Errorf("[%s] cut=%d 救回产物含非法 UTF-8: %q", name, cut, out)
|
||||
}
|
||||
}
|
||||
}
|
||||
if okCount == 0 {
|
||||
t.Errorf("[%s] 一个截断点都没救回,样本可能无效", name)
|
||||
}
|
||||
t.Logf("[%s] 救回 %d/%d 个截断点,含非法字节 %d 个", name, okCount, len(full), dirty)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,186 @@
|
||||
package governance
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"unicode"
|
||||
"unicode/utf8"
|
||||
)
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// 冲突预筛(2026-10-09 修 server.go:249 实体 bug)
|
||||
//
|
||||
// 背景:Scan 原本用「新内容实体 ∩ 候选实体」做廉价预筛,只有交集非空才跑
|
||||
// IsContradiction。但调用方(server.go)给**每个候选**塞的都是「新内容的实体」
|
||||
// (同一个切片),于是 e1==e2 恒真 ⇒ 预筛完全失效。实测后果:
|
||||
// - 预筛本应过滤话题无关的候选,实际一条都没过滤;
|
||||
// - 每条候选跑 len(实体) 次冗余 IsContradiction(线上实测每次 4~12 个实体),
|
||||
// 同一 (新内容, 候选) 对产出多条重复 Conflict(靠 suppressGate 的 SHA 兜住);
|
||||
// - `Entity` 字段恒记「新内容第一个实体」,与候选无关。
|
||||
//
|
||||
// ⚠️ 更正(勿沿用旧说法):曾把 /var/lib/memoryweave/conflicts-false-positive-*.json
|
||||
// 的 106 条已判假阳性归因于本 bug —— **实测不成立**。869 条真实样本(106 FP + 763
|
||||
// pending)在当前 IsContradiction 下判 true 的对数为 0,那些误报的根因是
|
||||
// IsContradiction 自身的旧启发式(阈值 0.3 + 单字否定),已于 2026-09-20/09-21 修复。
|
||||
// 本 bug 只是放大了比对面,不是误报的产生者。
|
||||
//
|
||||
// 为什么**不**恢复「实体相等」这条判据:候选侧只能从正文现提,与 LLM 为新内容抽的
|
||||
// 实体词表不同源,实测在 869 条真实冲突对上通过率仅 9% ⇒ 会拦掉 89% 的真实比对,
|
||||
// 把「预筛失效」换成「冲突检测整体静默失效」(更糟且无任何报错)。
|
||||
//
|
||||
// 改用**字符 bigram Jaccard**做话题交集判据(不依赖分词/词表,中英改写都稳)。
|
||||
// 阈值由实测读(2026-10-09,869 对真实样本 + 9 对锁定真矛盾):
|
||||
//
|
||||
// 真实样本 n=869:被判 true 的 0 对(判据已收敛,无从误杀)
|
||||
// 假阳性 n=106:bigram 上界 0.118
|
||||
// 锁定真矛盾 n=9 :bigram 下界 0.400
|
||||
// ⇒ 取 0.15:假阳性留 1.27× 余量,真矛盾留 2.67× 余量;实测误杀 0 对
|
||||
//
|
||||
// 两侧误差代价不对称,故刻意取偏宽容的一侧:**漏放**只多花廉价 CPU
|
||||
// (IsContradiction 有否定不对称等前置门,无关文本直接返回 false),
|
||||
// 而**误杀**是不可观测的静默漏检。故不把阈值推向真矛盾下界一侧。
|
||||
//
|
||||
// 兜底原则:**取不到信号时照常比较**(宁可多算,不可静默漏检)。
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// topicOverlapMin 话题交集最低 bigram Jaccard(实测标定见文件头)。
|
||||
const topicOverlapMin = 0.15
|
||||
|
||||
// sharesTopicWith 判定候选是否与新内容有真实话题交集。
|
||||
// 无交集 ⇒ 二者讲的是两件事,不可能互相矛盾 ⇒ 跳过昂贵比对。
|
||||
func sharesTopicWith(newContent string, newEntities []string, candidateContent string) bool {
|
||||
// 取不到信号 → 不预筛(宁可多算,不可漏检)
|
||||
if strings.TrimSpace(newContent) == "" || strings.TrimSpace(candidateContent) == "" {
|
||||
return true
|
||||
}
|
||||
// ① 字符 bigram 重叠(词表无关,主判据)
|
||||
if charBigramJaccard(newContent, candidateContent) >= topicOverlapMin {
|
||||
return true
|
||||
}
|
||||
// ② 新内容的实体出现在候选正文里(跨词表的补充信号;候选是改写句时的桥)
|
||||
low := strings.ToLower(candidateContent)
|
||||
for _, e := range newEntities {
|
||||
e = strings.TrimSpace(e)
|
||||
if utf8.RuneCountInString(e) < 2 {
|
||||
continue
|
||||
}
|
||||
if strings.Contains(low, strings.ToLower(e)) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// charBigramJaccard 计算两串的字符 bigram Jaccard 相似度(只计字母/数字,忽略大小写与标点)。
|
||||
func charBigramJaccard(a, b string) float64 {
|
||||
A, B := charBigrams(a), charBigrams(b)
|
||||
if len(A) == 0 || len(B) == 0 {
|
||||
return 0
|
||||
}
|
||||
inter := 0
|
||||
for g := range A {
|
||||
if B[g] {
|
||||
inter++
|
||||
}
|
||||
}
|
||||
union := len(A) + len(B) - inter
|
||||
if union == 0 {
|
||||
return 0
|
||||
}
|
||||
return float64(inter) / float64(union)
|
||||
}
|
||||
|
||||
// charBigrams 取字符串的字符 bigram 集合(归一化:仅字母/数字、转小写)。
|
||||
func charBigrams(s string) map[string]bool {
|
||||
var sb strings.Builder
|
||||
for _, r := range s {
|
||||
if unicode.IsLetter(r) || unicode.IsDigit(r) {
|
||||
sb.WriteRune(unicode.ToLower(r))
|
||||
}
|
||||
}
|
||||
t := []rune(sb.String())
|
||||
m := make(map[string]bool, len(t))
|
||||
for i := 0; i+1 < len(t); i++ {
|
||||
m[string(t[i:i+2])] = true
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
// entityTrimCutset 实体抽取时剥掉的首尾标点。
|
||||
const entityTrimCutset = " \t\r\n,.;:!?,。;:!?、\"'()()[]【】{}<>《》…—·-+*/\\|"
|
||||
|
||||
// maxExtractEntities 单次抽取上限(预筛只需话题信号,过多实体无益)。
|
||||
const maxExtractEntities = 8
|
||||
|
||||
// ExtractEntities 从一段文本抽取实体,**中英混排皆可**。
|
||||
//
|
||||
// 与 graph_auto.extractEntitiesFromText 的区别(重要):后者只认「ASCII 大写开头词」,
|
||||
// 对纯中文文本恒返回空 —— 直接当预筛输入会让冲突检测整体失效(见文件头教训),
|
||||
// 故此处自建中文支持版本。抽取口径:ASCII 大写开头词(≥2 字符)+ 连续汉字串(2~20 字),
|
||||
// 去重、上限 maxExtractEntities。
|
||||
func ExtractEntities(text string) []string {
|
||||
seen := make(map[string]bool, maxExtractEntities)
|
||||
out := make([]string, 0, maxExtractEntities)
|
||||
add := func(s string) {
|
||||
if s == "" || len(out) >= maxExtractEntities || seen[s] {
|
||||
return
|
||||
}
|
||||
seen[s] = true
|
||||
out = append(out, s)
|
||||
}
|
||||
for _, w := range strings.Fields(text) {
|
||||
w = strings.Trim(w, entityTrimCutset)
|
||||
if utf8.RuneCountInString(w) < 2 {
|
||||
continue
|
||||
}
|
||||
if r := []rune(w); r[0] >= 'A' && r[0] <= 'Z' { // ASCII 专名:Hermes / Docker / CVE
|
||||
add(w)
|
||||
}
|
||||
for _, seg := range chineseRuns(w) { // 连续汉字串(中文无空格分词,按非汉字切)
|
||||
if n := utf8.RuneCountInString(seg); n >= 2 && n <= 20 {
|
||||
add(seg)
|
||||
}
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// chineseRuns 切出字符串中的连续汉字串。
|
||||
func chineseRuns(s string) []string {
|
||||
var runs []string
|
||||
var b strings.Builder
|
||||
flush := func() {
|
||||
if b.Len() > 0 {
|
||||
runs = append(runs, b.String())
|
||||
b.Reset()
|
||||
}
|
||||
}
|
||||
for _, r := range s {
|
||||
if unicode.Is(unicode.Han, r) {
|
||||
b.WriteRune(r)
|
||||
} else {
|
||||
flush()
|
||||
}
|
||||
}
|
||||
flush()
|
||||
return runs
|
||||
}
|
||||
|
||||
// conflictEntity 为冲突记录挑一个「可读的触发实体」:优先取**在候选正文里真的出现过**的
|
||||
// 新内容实体(即两侧共同话题),取不到则退化为第一个新实体(可能为空串)。
|
||||
// 旧实现直接写 e1(恒为新内容的第一个实体,与候选无关),事后核对时无法据此定位共同话题。
|
||||
func conflictEntity(newContent string, newEntities []string, candidateContent string) string {
|
||||
low := strings.ToLower(candidateContent)
|
||||
for _, e := range newEntities {
|
||||
e = strings.TrimSpace(e)
|
||||
if utf8.RuneCountInString(e) < 2 {
|
||||
continue
|
||||
}
|
||||
if strings.Contains(low, strings.ToLower(e)) {
|
||||
return e
|
||||
}
|
||||
}
|
||||
if len(newEntities) > 0 {
|
||||
return strings.TrimSpace(newEntities[0])
|
||||
}
|
||||
return ""
|
||||
}
|
||||
@@ -0,0 +1,249 @@
|
||||
package governance
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// 2026-10-09 验收测试:修 server.go:249「实体预筛失效」bug
|
||||
//
|
||||
// 缺陷:调用方给每个候选塞的都是**新内容的实体**(同一切片)⇒ Scan 里 e1==e2 恒真
|
||||
// ⇒ 预筛完全失效。实害(按本条实测重述,勿沿用旧说法):
|
||||
// a) 预筛本应过滤「话题无关」的候选,实际一条都没过滤;
|
||||
// b) 每条候选跑 len(实体) 次冗余 IsContradiction(线上实测每次 4~12 个实体),
|
||||
// 并对同一 (新内容, 候选) 对产出多条重复 Conflict(靠 suppressGate 的 SHA 兜住);
|
||||
// c) `Entity` 字段恒记「新内容第一个实体」,与候选无关,事后无法据此定位共同话题。
|
||||
//
|
||||
// ⚠️ 更正:曾把 /var/lib/memoryweave/conflicts-false-positive-*.json 的 106 条假阳性
|
||||
// 归因于本 bug —— **实测不成立**:这 869 条真实样本(106 FP + 763 pending)在当前
|
||||
// IsContradiction 下判 true 的对数为 **0**,即那些误报的根因是 IsContradiction 自身的
|
||||
// 旧启发式(阈值 0.3 + 单字否定),已于 2026-09-20/09-21 修复。本 bug 只放大了比对面,
|
||||
// 不是误报的产生者。判据如下(本文件锁定):
|
||||
// ① 门不得拦下任何「判据判 true」的对(真实样本 + 锁定真矛盾双向验证)
|
||||
// ② 每条候选最多产出 1 条 Conflict(消除 len(实体) 倍冗余)
|
||||
// ③ 阈值 0.15 必须落在实测的 FP 上界与真矛盾下界之间
|
||||
// ④ 中文实体抽取必须非空(拿只认 ASCII 大写词的旧版当修法会静默杀死检出)
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
type fpRec struct {
|
||||
LeftContent string `json:"left_content"`
|
||||
RightContent string `json:"right_content"`
|
||||
Entity string `json:"entity"`
|
||||
Content string `json:"content"`
|
||||
}
|
||||
|
||||
func loadRealConflicts(t *testing.T, pat string) []fpRec {
|
||||
t.Helper()
|
||||
paths, _ := filepath.Glob(pat)
|
||||
var all []fpRec
|
||||
for _, p := range paths {
|
||||
b, err := os.ReadFile(p)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
var w struct {
|
||||
Conflicts []fpRec `json:"conflicts"`
|
||||
Pending []fpRec `json:"pending"`
|
||||
}
|
||||
if err := json.Unmarshal(b, &w); err == nil && (len(w.Conflicts) > 0 || len(w.Pending) > 0) {
|
||||
all = append(all, w.Conflicts...)
|
||||
all = append(all, w.Pending...)
|
||||
continue
|
||||
}
|
||||
var arr []fpRec
|
||||
if err := json.Unmarshal(b, &arr); err == nil {
|
||||
all = append(all, arr...)
|
||||
}
|
||||
}
|
||||
out := all[:0]
|
||||
for _, r := range all {
|
||||
if r.RightContent == "" && r.Content != "" {
|
||||
r.RightContent = r.Content
|
||||
}
|
||||
if r.LeftContent != "" && r.RightContent != "" {
|
||||
out = append(out, r)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// ① 真实样本上:门拦下的对里,「判据判 true」的必须为 0 —— 即门一次都不许误杀。
|
||||
func TestScanPrefilter_NoDetectionLossOnRealSamples(t *testing.T) {
|
||||
fp := loadRealConflicts(t, "/var/lib/memoryweave/conflicts-false-positive-*.json")
|
||||
pending := loadRealConflicts(t, "/var/lib/memoryweave/conflicts.json")
|
||||
all := append(append([]fpRec{}, fp...), pending...)
|
||||
if len(all) == 0 {
|
||||
t.Skip("无生产样本(/var/lib/memoryweave/conflicts*.json)")
|
||||
}
|
||||
|
||||
judgePos, gateBlocked, killed := 0, 0, 0
|
||||
for _, r := range all {
|
||||
j := IsContradiction(r.RightContent, r.LeftContent)
|
||||
g := !sharesTopicWith(r.RightContent, ExtractEntities(r.RightContent), r.LeftContent)
|
||||
if j {
|
||||
judgePos++
|
||||
}
|
||||
if g {
|
||||
gateBlocked++
|
||||
}
|
||||
if j && g {
|
||||
killed++
|
||||
if killed <= 5 {
|
||||
t.Logf(" ⚠️ 判据判 true 却被门拦下: new=%s | cand=%s | bigram=%.3f",
|
||||
trim(r.RightContent), trim(r.LeftContent),
|
||||
charBigramJaccard(r.RightContent, r.LeftContent))
|
||||
}
|
||||
}
|
||||
}
|
||||
t.Logf("真实样本 %d 对(FP %d + pending %d):判据判 true %d 对;门拦下 %d 对(省 %.0f%% 比对);★误杀 %d 对",
|
||||
len(all), len(fp), len(pending), judgePos, gateBlocked,
|
||||
100*float64(gateBlocked)/float64(len(all)), killed)
|
||||
if killed > 0 {
|
||||
t.Errorf("门拦下了 %d 对判据判 true 的样本 —— 预筛在误杀真检出", killed)
|
||||
}
|
||||
}
|
||||
|
||||
// ② 锁定真矛盾必须仍检出,且**不依赖实体词表**(entities=nil 也要检出)。
|
||||
// 这批用例来自 conflict_scope_test.go / zz_overlap_probe_test.go 已锁定的真矛盾。
|
||||
func TestScanPrefilter_KeepsTrueContradictions(t *testing.T) {
|
||||
pairs := []struct{ newC, cand string }{
|
||||
{"用户昨天没买苹果", "用户昨天买了苹果"},
|
||||
{"用户今天还没完成任务", "用户今天完成了任务"},
|
||||
{"Docker is NOT used by Hermes", "Docker is used by Hermes"},
|
||||
{"config not updated", "config updated"},
|
||||
{"MemoryFabric不是独立记忆层服务", "MemoryFabric是独立记忆层服务"},
|
||||
{"助手确认了织忆不是跨域关联与语义检索载体", "助手确认了织忆作为跨域关联与语义检索载体"},
|
||||
{"牧尘不在招商局工作", "牧尘在招商局工作"},
|
||||
{"牧尘没在招商局工作", "牧尘在招商局工作"},
|
||||
{"用户未在招商局工作", "用户在招商局工作"},
|
||||
}
|
||||
for _, p := range pairs {
|
||||
// 前置:判据本身必须认为它矛盾(否则本用例没有意义)
|
||||
if !IsContradiction(p.newC, p.cand) {
|
||||
t.Errorf("用例失效(判据不认为是矛盾): %q vs %q", p.newC, p.cand)
|
||||
continue
|
||||
}
|
||||
cd := NewConflictDetector()
|
||||
got := cd.Scan(p.newC, nil, []map[string]interface{}{{"content": p.cand}})
|
||||
if len(got) == 0 {
|
||||
t.Errorf("真矛盾漏检(entities=nil): new=%q cand=%q", p.newC, p.cand)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ③ 冗余消除的可观测证据。
|
||||
//
|
||||
// 注意:**不能用「返回条数」当判据** —— 旧实现里同一 (新内容, 候选) 对会 append
|
||||
// len(实体) 条 Conflict,但 suppressGate 按 SHA 合并 ⇒ 返回条数同样是 1(实测确认),
|
||||
// 无鉴别力。真正可观测的差异是**比对工作量**与 **Entity 字段来源**。本测试锁后者。
|
||||
func TestConflictEntity_ProvenanceIsShared(t *testing.T) {
|
||||
// ① 实体列表首项是「新内容独有、候选正文里根本没有」的词 ⇒
|
||||
// 旧实现 Entity 恒取 newEntities[0]="虚构实体X"(与候选无关);
|
||||
// 新实现必须挑**真的出现在候选正文里**的共同话题。
|
||||
cd := NewConflictDetector()
|
||||
got := cd.Scan("牧尘不在招商局工作", []string{"虚构实体X", "招商局", "牧尘"},
|
||||
[]map[string]interface{}{{"content": "牧尘在招商局工作"}})
|
||||
if len(got) != 1 {
|
||||
t.Fatalf("应检出 1 条冲突,实际 %d 条", len(got))
|
||||
}
|
||||
if got[0].Entity == "虚构实体X" {
|
||||
t.Errorf("Entity=%q 取到了不在候选正文里的实体(旧 bug 行为:盲取 newEntities[0])", got[0].Entity)
|
||||
}
|
||||
if got[0].Entity != "招商局" && got[0].Entity != "牧尘" {
|
||||
t.Errorf("Entity=%q 不是两侧共同话题(期望候选正文里出现的实体)", got[0].Entity)
|
||||
}
|
||||
t.Logf("✓ Entity 来源正确: %q(候选正文 = %q)", got[0].Entity, "牧尘在招商局工作")
|
||||
|
||||
// ② 实体全都不在候选正文里 ⇒ 退化为首项(有文档记录的兜底),不得为空串/崩溃
|
||||
cd2 := NewConflictDetector()
|
||||
got2 := cd2.Scan("牧尘不在招商局工作", []string{"甲", "乙"},
|
||||
[]map[string]interface{}{{"content": "牧尘在招商局工作"}})
|
||||
if len(got2) == 1 && got2[0].Entity == "" {
|
||||
t.Errorf("兜底路径产出了空 Entity")
|
||||
}
|
||||
}
|
||||
|
||||
// ③b 门确实在过滤(不是恒真):真实样本上跳过比例必须 > 0。
|
||||
func TestScanPrefilter_ActuallyFilters(t *testing.T) {
|
||||
all := append(
|
||||
loadRealConflicts(t, "/var/lib/memoryweave/conflicts-false-positive-*.json"),
|
||||
loadRealConflicts(t, "/var/lib/memoryweave/conflicts.json")...)
|
||||
if len(all) == 0 {
|
||||
t.Skip("无生产样本")
|
||||
}
|
||||
skipped := 0
|
||||
for _, r := range all {
|
||||
if !sharesTopicWith(r.RightContent, ExtractEntities(r.RightContent), r.LeftContent) {
|
||||
skipped++
|
||||
}
|
||||
}
|
||||
ratio := float64(skipped) / float64(len(all))
|
||||
t.Logf("真实样本 %d 对,门跳过 %d 对(%.0f%%)", len(all), skipped, ratio*100)
|
||||
if skipped == 0 {
|
||||
t.Errorf("门一次都没跳过 —— 与旧 bug(恒真)无区别")
|
||||
}
|
||||
}
|
||||
|
||||
// ④ 阈值必须落在实测区间内(阈值由数据标定,不许拍脑袋)。
|
||||
func TestTopicThreshold_LockedToMeasuredData(t *testing.T) {
|
||||
fp := loadRealConflicts(t, "/var/lib/memoryweave/conflicts-false-positive-*.json")
|
||||
maxFP := 0.0
|
||||
for _, r := range fp {
|
||||
if v := charBigramJaccard(r.RightContent, r.LeftContent); v > maxFP {
|
||||
maxFP = v
|
||||
}
|
||||
}
|
||||
trueMin := 1.0
|
||||
for _, p := range [][2]string{
|
||||
{"用户昨天没买苹果", "用户昨天买了苹果"},
|
||||
{"config not updated", "config updated"},
|
||||
{"牧尘不在招商局工作", "牧尘在招商局工作"},
|
||||
{"Docker is NOT used by Hermes", "Docker is used by Hermes"},
|
||||
{"用户未在招商局工作", "用户在招商局工作"},
|
||||
} {
|
||||
if v := charBigramJaccard(p[0], p[1]); v < trueMin {
|
||||
trueMin = v
|
||||
}
|
||||
}
|
||||
t.Logf("实测:假阳性 bigram 上界=%.3f | 阈值=%.2f | 真矛盾 bigram 下界=%.3f(两侧余量 %.2fx / %.2fx)",
|
||||
maxFP, topicOverlapMin, trueMin, topicOverlapMin/maxFP, trueMin/topicOverlapMin)
|
||||
if len(fp) > 0 && maxFP >= topicOverlapMin {
|
||||
t.Errorf("阈值 %.2f 未覆盖假阳性上界 %.3f —— 会漏放话题无关候选", topicOverlapMin, maxFP)
|
||||
}
|
||||
if trueMin < topicOverlapMin {
|
||||
t.Errorf("阈值 %.2f 高于真矛盾下界 %.3f —— 会误杀真冲突", topicOverlapMin, trueMin)
|
||||
}
|
||||
}
|
||||
|
||||
// ⑤ 中文实体抽取必须非空 —— 这正是「拿 graph_auto 版(只认 ASCII 大写词)当修法」
|
||||
// 会踩的坑:对纯中文恒返回空 ⇒ 预筛永不匹配 ⇒ 冲突检测整体静默失效。
|
||||
func TestExtractEntities_ChineseSupported(t *testing.T) {
|
||||
for _, s := range []string{
|
||||
"牧尘在招商局工作",
|
||||
"织忆服务部署在 192.168.5.106 上",
|
||||
"小怡使用 Hermes Agent 部署织忆",
|
||||
} {
|
||||
if got := ExtractEntities(s); len(got) == 0 {
|
||||
t.Errorf("ExtractEntities(%q) 返回空 —— 纯中文文本必须抽得出实体", s)
|
||||
} else {
|
||||
t.Logf("ExtractEntities(%q) = %v", s, got)
|
||||
}
|
||||
}
|
||||
if got := ExtractEntities("The Deployment uses Hermes Agent"); len(got) == 0 {
|
||||
t.Errorf("英文专名未抽到")
|
||||
}
|
||||
if got := ExtractEntities("Aaa Bbb Ccc Ddd Eee Fff Ggg Hhh Iii Jjj Kkk"); len(got) > maxExtractEntities {
|
||||
t.Errorf("实体数 %d 超过上限 %d", len(got), maxExtractEntities)
|
||||
}
|
||||
}
|
||||
|
||||
func trim(s string) string {
|
||||
r := []rune(s)
|
||||
if len(r) > 46 {
|
||||
return string(r[:46]) + "…"
|
||||
}
|
||||
return s
|
||||
}
|
||||
@@ -236,34 +236,44 @@ func (cd *ConflictDetector) persistLocked() {
|
||||
}
|
||||
|
||||
// Scan 扫描新记忆与已有记忆的冲突
|
||||
//
|
||||
// 2026-10-09 修(实体预筛失效):原实现用「新内容实体 ∩ 候选实体」做预筛,而调用方
|
||||
// 给每个候选塞的都是**新内容的实体**(同一个切片)⇒ e1==e2 恒真 ⇒ 预筛完全失效。
|
||||
// 实害:预筛一条都没过滤 + 每条候选跑 len(实体) 次冗余 IsContradiction(线上实测 4~12)
|
||||
// + 同一 (新内容, 候选) 对产出多条重复 Conflict + Entity 字段恒为「新内容第一个实体」。
|
||||
// (**不是** 106 条假阳性的成因——那批误报的根因是 IsContradiction 旧启发式,
|
||||
// 已于 2026-09-20/09-21 修复;869 条真实样本在当前判据下判 true 的为 0。)
|
||||
//
|
||||
// 修法:候选侧改为用**它自己的正文**做话题交集判定(sharesTopicWith,词表无关),
|
||||
// 每条候选最多比对一次、最多产出一条 Conflict;取不到信号时照常比较(宁可多算,不可漏检)。
|
||||
// 详见 conflict_prefilter.go 文件头的实测标定。
|
||||
func (cd *ConflictDetector) Scan(newContent string, newEntities []string, existing []map[string]interface{}) []*Conflict {
|
||||
var conflicts []*Conflict
|
||||
|
||||
for _, existing := range existing {
|
||||
existingContent := existing["content"].(string)
|
||||
existingEntities := toStringSlice(existing["entities"])
|
||||
for _, cand := range existing {
|
||||
existingContent, _ := cand["content"].(string)
|
||||
if existingContent == "" {
|
||||
continue
|
||||
}
|
||||
|
||||
// 检查相同实体但不同关系
|
||||
for _, e1 := range newEntities {
|
||||
for _, e2 := range existingEntities {
|
||||
if e1 == e2 {
|
||||
// 检测事实冲突:内容语义矛盾
|
||||
if IsContradiction(newContent, existingContent) {
|
||||
conflicts = append(conflicts, &Conflict{
|
||||
ID: "cft_" + strconv.FormatInt(time.Now().UnixNano(), 10) + "_" + strconv.Itoa(len(conflicts)),
|
||||
Type: ConflictFact,
|
||||
Entity: e1,
|
||||
Description: "事实冲突:新内容与已有记录矛盾",
|
||||
Strategy: "ask_user",
|
||||
Status: "pending",
|
||||
CreatedAt: time.Now(),
|
||||
// 2026-09-20:留档两侧原文,供事后核实真假矛盾
|
||||
LeftContent: existingContent,
|
||||
RightContent: newContent,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
// 预筛:无话题交集 ⇒ 二者讲的是两件事,不可能互相矛盾
|
||||
if !sharesTopicWith(newContent, newEntities, existingContent) {
|
||||
continue
|
||||
}
|
||||
|
||||
if IsContradiction(newContent, existingContent) {
|
||||
conflicts = append(conflicts, &Conflict{
|
||||
ID: "cft_" + strconv.FormatInt(time.Now().UnixNano(), 10) + "_" + strconv.Itoa(len(conflicts)),
|
||||
Type: ConflictFact,
|
||||
Entity: conflictEntity(newContent, newEntities, existingContent),
|
||||
Description: "事实冲突:新内容与已有记录矛盾",
|
||||
Strategy: "ask_user",
|
||||
Status: "pending",
|
||||
CreatedAt: time.Now(),
|
||||
// 2026-09-20:留档两侧原文,供事后核实真假矛盾
|
||||
LeftContent: existingContent,
|
||||
RightContent: newContent,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -25,25 +25,64 @@ func NewAutoGraphUpdater(g GraphStore) *AutoGraphUpdater {
|
||||
// 会话/笔记内容里抽出实体 n_小唯 并重建 DERIVED_FROM 边。
|
||||
// 根治:在**唯一写入点**(entityID / cleanEntityName,所有节点 id/name 与边 id
|
||||
// 都必经这两个转换)做旧名→新名映射,从源头杜绝旧名节点再入图。
|
||||
//
|
||||
// ⚠️ 2026-10-03 扩充:原表只有中文「小唯」,漏掉全部英文旧名。实查图谱确认:
|
||||
// n_xiaowei 仍有 7 条边(源:会话正文里的 Xiaowei / soulful-xiaowei-main 等),
|
||||
// 中文旧名已清干净(小唯 count=0),但英文旧名从未被归一。
|
||||
// 归一到「小怡」(n_小怡 count=880 = 真实主节点)——注意**不是** xiaoyi,
|
||||
// 实查 n_xiaoyi count=0,图谱里没有这个节点,映射过去会造出第二个孤儿。
|
||||
// 教训:归一目标必须是**实查存在的**主节点,不能从"拼音应该怎么写"推断。
|
||||
var entityAliases = map[string]string{
|
||||
"小唯": "小怡", // 旧名 → 新名(2026-09-12 全域改名)
|
||||
"小唯": "小怡", // 旧名 → 新名(2026-09-12 全域改名)
|
||||
// 英文/拼音旧名(2026-10-03 补)→ 统一归到实查存在的主节点「小怡」
|
||||
// ⚠️ 归一目标必须是实查存在的节点:n_小怡 count=880(真主节点);
|
||||
// n_xiaoyi count=0(不存在)——不能按拼音推断,否则造出第二个孤儿。
|
||||
"xiaowei": "小怡",
|
||||
"soulful-xiaowei-main": "soulful-小怡-main",
|
||||
}
|
||||
|
||||
// normalizeAlias 把实体名的旧名别名归一为新名(仅在写入点调用)。
|
||||
// 既做整体等名映射,也做子串替换 —— 蒸馏实体可能是单词「小唯」,
|
||||
// 事实节点 id 里的旧名则嵌在整句里(如 f_…_mc_小唯_07-wiki…)。
|
||||
// normalizeAlias 命中判断:对 map 做「精确 → 小写精确 → 子串(大小写不敏感)」
|
||||
// 三级匹配。中文 key 不受大小写影响,英文 key 用小写登记覆盖全部拼写。
|
||||
func normalizeAlias(name string) string {
|
||||
// 1. 全名精确匹配(最具体,如 soulful-xiaowei-main)
|
||||
if alias, ok := entityAliases[name]; ok {
|
||||
return alias
|
||||
}
|
||||
// 2. 小写精确匹配(覆盖 Xiaowei/XIAOWEI 等)—— 先做这个再进子串,
|
||||
// 避免"整名匹配"被"子串替换"抢先降级成半截结果
|
||||
lower := strings.ToLower(name)
|
||||
for old, new := range entityAliases {
|
||||
if strings.Contains(name, old) {
|
||||
name = strings.ReplaceAll(name, old, new)
|
||||
if strings.ToLower(old) == lower {
|
||||
return new
|
||||
}
|
||||
}
|
||||
// 3. 子串替换(旧名嵌在整句/复合标识符里,大小写不敏感)
|
||||
for old, new := range entityAliases {
|
||||
o := strings.ToLower(old)
|
||||
if o != old && !isASCIIOnly(old) {
|
||||
continue // 中文 key 不走大小写路径
|
||||
}
|
||||
for {
|
||||
i := strings.Index(strings.ToLower(name), o)
|
||||
if i < 0 {
|
||||
break
|
||||
}
|
||||
name = name[:i] + new + name[i+len(o):]
|
||||
}
|
||||
}
|
||||
return name
|
||||
}
|
||||
|
||||
// isASCIIOnly 判断字符串是否纯 ASCII(用于区分钟文 key 与英文 key)
|
||||
func isASCIIOnly(s string) bool {
|
||||
for _, r := range s {
|
||||
if r > unicode.MaxASCII {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// UpdateFromDistill 从蒸馏产物自动更新图谱(§3.4)
|
||||
func (agu *AutoGraphUpdater) UpdateFromDistill(distilled *DistillInput) {
|
||||
// 0. 密钥过滤(2026-09-10 P0,见 internal/redact):含凭证形状的实体不进图,
|
||||
|
||||
@@ -60,6 +60,12 @@ type MemoryRecord struct {
|
||||
// script_version=写入脚本版本标识。写入方经 metadata 携带,server 蒸馏落库。
|
||||
EmbeddingModel string `json:"embedding_model"`
|
||||
ScriptVersion string `json:"script_version"`
|
||||
// 2026-10-08 (Gap#2 来源加权):confidence=写入方显式声明的信任等级(0.4~1.0),
|
||||
// 不从 source 反推(source 是「哪条管道进来」,confidence 是「内容可不可信」,两者正交)。
|
||||
// 指针类型:nil ⇒ JSON null ⇒ Rust 写 NULL ⇒ 读路径 COALESCE→1.0(加权首日 no-op)。
|
||||
// ConfidenceSource=authoritative(写入时声明) | estimated(事后回填猜测),用于日后定向纠偏。
|
||||
Confidence *float64 `json:"confidence,omitempty"`
|
||||
ConfidenceSource string `json:"confidence_source,omitempty"`
|
||||
LastRecalledAt time.Time `json:"last_recalled_at"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
@@ -99,6 +105,11 @@ type RecallResult struct {
|
||||
Namespace string `json:"namespace,omitempty"`
|
||||
AgentID string `json:"agent_id,omitempty"`
|
||||
Source string `json:"source,omitempty"`
|
||||
// 2026-10-08 (Gap#2):把信任等级直通给调用方。此前加列但不动本结构 ⇒
|
||||
// confidence 对排序/调用方零影响(「字段加了没接线」的经典静默 no-op)。
|
||||
// Confidence=nil 表示该条未声明(存量),调用方按 1.0 处理。
|
||||
Confidence *float64 `json:"confidence,omitempty"`
|
||||
ConfidenceSource string `json:"confidence_source,omitempty"`
|
||||
}
|
||||
|
||||
// RerankResult rerank 重排后的单条结果。
|
||||
|
||||
@@ -0,0 +1,103 @@
|
||||
package storage_test
|
||||
|
||||
// conf_e2e_test.go — 2026-10-08 Gap#2 confidence 字段 Go 侧端到端回归测试
|
||||
//
|
||||
// 为什么必须存在:confidence 这一个字段要穿过 **6 层**才到得了调用方 ——
|
||||
// ① LanceDB schema ② Rust insert_batch ③ Rust scan/search 的 MemoryRecord 字面量
|
||||
// ④ Go 解包结构 ⑤ Go 输出构造(models.MemoryRecord)⑥ Go RecallResult
|
||||
// 任意一层漏了,**编译、单测、读代码全都看不出来** —— 表现是「字段恒为 nil」的静默 no-op。
|
||||
// 本次开发中第 ⑤ 层就是靠这个测试才抓到的(当时其余 5 层都已通)。
|
||||
//
|
||||
// 跑法(需先起 sidecar,未设环境变量则自动跳过,可安全留在 CI):
|
||||
// zhiyi-consolidate --mode socket --socket /tmp/mw-test3.sock --data-dir /tmp/mw-copy3
|
||||
// ZHIYI_CONF_SOCK=/tmp/mw-test3.sock go test -run TestConfidenceE2E -v ./internal/storage/
|
||||
|
||||
import (
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/xiaoxue/memoryweave/internal/models"
|
||||
"github.com/xiaoxue/memoryweave/internal/storage"
|
||||
)
|
||||
|
||||
func TestConfidenceE2E(t *testing.T) {
|
||||
sock := os.Getenv("ZHIYI_CONF_SOCK")
|
||||
if sock == "" {
|
||||
t.Skip("ZHIYI_CONF_SOCK 未设置,跳过(需要 sidecar 在跑)")
|
||||
}
|
||||
const ns = "conf-go-e2e"
|
||||
|
||||
// embedder 传 nil:记录自带 Vector,走不到编码分支
|
||||
rc := storage.NewRustLanceDBClient(sock, nil)
|
||||
|
||||
vec := make([]float32, 1024)
|
||||
for i := range vec {
|
||||
vec[i] = 0.02
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
f := 0.5
|
||||
|
||||
mk := func(id string, conf *float64, src string) models.MemoryRecord {
|
||||
return models.MemoryRecord{
|
||||
ID: id, AgentID: "hermes-a06", Namespace: ns,
|
||||
Content: "Go 侧 confidence 端到端样本 " + id,
|
||||
Category: "test", Vector: vec,
|
||||
Tier: "normal", Importance: 1.0, QualityScore: 0.8,
|
||||
Version: 1, Freshness: "fresh",
|
||||
Source: "wiki:/tmp/go-conf-e2e.md",
|
||||
EmbeddingModel: "bge-m3", ScriptVersion: "go-conf-e2e",
|
||||
Confidence: conf, ConfidenceSource: src,
|
||||
CreatedAt: now, UpdatedAt: now,
|
||||
}
|
||||
}
|
||||
|
||||
t.Run("写入A_显式声明0.5", func(t *testing.T) {
|
||||
if err := rc.InsertMemory(mk("go-conf-A", &f, "authoritative")); err != nil {
|
||||
t.Fatalf("InsertMemory(A) 失败: %v", err)
|
||||
}
|
||||
})
|
||||
t.Run("写入B_不声明", func(t *testing.T) {
|
||||
if err := rc.InsertMemory(mk("go-conf-B", nil, "")); err != nil {
|
||||
t.Fatalf("InsertMemory(B) 失败: %v", err)
|
||||
}
|
||||
})
|
||||
|
||||
rows, err := rc.Search("memories", vec, 20, ns)
|
||||
if err != nil {
|
||||
t.Fatalf("Search 失败: %v", err)
|
||||
}
|
||||
t.Logf("Search 命中 %d 条", len(rows))
|
||||
|
||||
var a, b *models.MemoryRecord
|
||||
for i := range rows {
|
||||
switch rows[i].ID {
|
||||
case "go-conf-A":
|
||||
a = &rows[i]
|
||||
case "go-conf-B":
|
||||
b = &rows[i]
|
||||
}
|
||||
}
|
||||
|
||||
if a == nil {
|
||||
t.Fatal("❌ A 读不回来")
|
||||
}
|
||||
if a.Confidence == nil {
|
||||
t.Fatal("❌ A: Confidence 为 nil —— Go→IPC→Rust 写入或读回断链")
|
||||
}
|
||||
if *a.Confidence != 0.5 {
|
||||
t.Fatalf("❌ A: Confidence=%v,期望 0.5", *a.Confidence)
|
||||
}
|
||||
if a.ConfidenceSource != "authoritative" {
|
||||
t.Fatalf("❌ A: ConfidenceSource=%q,期望 authoritative", a.ConfidenceSource)
|
||||
}
|
||||
t.Logf("✅ A: confidence=%.2f source=%q 完整往返", *a.Confidence, a.ConfidenceSource)
|
||||
|
||||
if b == nil {
|
||||
t.Fatal("❌ B 读不回来")
|
||||
}
|
||||
if b.Confidence != nil {
|
||||
t.Fatalf("❌ B: 未声明却读回 %v —— 应为 nil(NULL)", *b.Confidence)
|
||||
}
|
||||
t.Logf("✅ B: confidence=nil(NULL),未声明的语义被保住")
|
||||
}
|
||||
@@ -158,6 +158,8 @@ func (rc *RustLanceDBClient) InsertMemory(m models.MemoryRecord) error {
|
||||
"source": m.Source, "volatile_flag": false, "is_deleted": m.IsDeleted,
|
||||
"depends_on": "[]", "derived_from": "", "last_recalled_at": "",
|
||||
"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,
|
||||
"created_at": m.CreatedAt.Format(time.RFC3339),
|
||||
"updated_at": m.UpdatedAt.Format(time.RFC3339),
|
||||
}
|
||||
@@ -191,6 +193,10 @@ func (rc *RustLanceDBClient) Search(table string, vector []float32, topK int, na
|
||||
// 2026-09-26:出处(笔记路径)—— 此前**解包结构缺该字段** → IPC 主路径 Source 恒空
|
||||
// (本地兜底路径有 Source,主路径没有 → recall 永远拿不到出处,链路看似修好实则断在这一环)
|
||||
Source string `json:"source"`
|
||||
// 2026-10-08 (Gap#2):读侧投影 —— 不补这里,加列也读不出来(静默 no-op)。
|
||||
// 用 *float64 区分「NULL(存量未声明)」与「显式 0.0」。
|
||||
Confidence *float64 `json:"confidence"`
|
||||
ConfidenceSource string `json:"confidence_source"`
|
||||
// vector 必须解析:近重复去重(commit near_dup / 蒸馏写入 gate /
|
||||
// admin/dedup 端点)全部依赖 records 携带向量做 cosine 比较。
|
||||
// 此前缺失该字段 → Vector 恒为空 → 所有 len(Vector)>0 判定失败 →
|
||||
@@ -270,6 +276,11 @@ func (rc *RustLanceDBClient) Search(table string, vector []float32, topK int, na
|
||||
Tier: r.Tier,
|
||||
Source: r.Source, // 2026-09-26:出处透出(缺此行 → recall.source 恒空)
|
||||
Vector: r.Vector, // 去重判据依赖,勿删
|
||||
// 2026-10-08 (Gap#2):解包结构有 ≠ 输出有。缺这两行 ⇒ 加列、改 schema、
|
||||
// 通 IPC 全做完,recall 拿到的 Confidence 仍是 nil(第 5 层静默断链,
|
||||
// 端到端测试才抓得到)。
|
||||
Confidence: r.Confidence,
|
||||
ConfidenceSource: r.ConfidenceSource,
|
||||
})
|
||||
if len(out) >= topK {
|
||||
break
|
||||
@@ -310,6 +321,9 @@ func (rc *RustLanceDBClient) Scan(limit int, namespaceFilter string) ([]models.M
|
||||
Tier string `json:"tier"`
|
||||
Freshness string `json:"freshness"`
|
||||
Source string `json:"source"`
|
||||
// 2026-10-08 (Gap#2):scan 路径同样要投影,否则该路径返回的 confidence 恒 nil
|
||||
Confidence *float64 `json:"confidence"`
|
||||
ConfidenceSource string `json:"confidence_source"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(resp.ReportJSON), &raw); err != nil {
|
||||
return nil, fmt.Errorf("scan unmarshal: %w", err)
|
||||
@@ -334,6 +348,7 @@ func (rc *RustLanceDBClient) Scan(limit int, namespaceFilter string) ([]models.M
|
||||
CreatedAt: parseTime(r.CreatedAt), UpdatedAt: parseTime(r.UpdatedAt),
|
||||
Importance: r.Importance, QualityScore: r.QualityScore,
|
||||
RecallCount: int(r.RecallCount), Tier: r.Tier, Source: r.Source,
|
||||
Confidence: r.Confidence, ConfidenceSource: r.ConfidenceSource, // 2026-10-08 (Gap#2)
|
||||
})
|
||||
if len(out) >= limit {
|
||||
break
|
||||
@@ -749,6 +764,17 @@ func mapFloat(m map[string]interface{}, key string) float64 {
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
// mapFloatPtr 2026-10-08 (Gap#2):可空浮点。JSON null / 缺键 → nil(区别于显式 0.0)。
|
||||
// 用于 confidence:nil 表示「该条未声明信任等级」(存量行),调用方 COALESCE 到 1.0。
|
||||
func mapFloatPtr(m map[string]interface{}, key string) *float64 {
|
||||
if v, ok := m[key]; ok {
|
||||
if f, ok := v.(float64); ok {
|
||||
return &f
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func mapInt(m map[string]interface{}, key string) int {
|
||||
if v, ok := m[key]; ok {
|
||||
if f, ok := v.(float64); ok {
|
||||
@@ -806,6 +832,9 @@ func (rc *RustLanceDBClient) GetSkillCandidates(minRecalls, limit int) ([]models
|
||||
}
|
||||
m.QualityScore = mapFloat(r, "quality_score")
|
||||
m.Importance = mapFloat(r, "importance")
|
||||
// 2026-10-08 (Gap#2):读侧接线(不补 ⇒ 加列也读不出来)
|
||||
m.Confidence = mapFloatPtr(r, "confidence")
|
||||
m.ConfidenceSource = mapStr(r, "confidence_source")
|
||||
m.UsefulCount = mapInt(r, "useful_count")
|
||||
m.NotUsefulCount = mapInt(r, "not_useful_count")
|
||||
m.RecallCount = mapInt(r, "recall_count")
|
||||
|
||||
@@ -6,12 +6,51 @@ import (
|
||||
"fmt"
|
||||
"log"
|
||||
"math"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/xiaoxue/memoryweave/internal/models"
|
||||
)
|
||||
|
||||
// ── 2026-10-08 (Gap#2 来源加权):门控参数 ──────────────────────────────
|
||||
//
|
||||
// 设计取舍(MoA 评审采纳):
|
||||
// · 默认 λ=0 ⇒ **首日 no-op**。存量 15105 条 confidence 全为 NULL,
|
||||
// 此时任何加权都只是在噪声上做乘法。先接线、采分布,再定系数。
|
||||
// · 形式为**加性去中心偏置** base + λ(conf − mean),而非乘性 base*conf:
|
||||
// memories 表 100% 经 LLM 蒸馏(server.go 唯一 InsertMemory 调用点),
|
||||
// 乘性会把整批蒸馏产物一起压沉,并与 QualityScore 在同一根轴上二次惩罚。
|
||||
// 加性只平移均值、不缩方差,对尾部排序影响可控。
|
||||
// · 启用 = 改环境变量,不改代码(可回滚,无需重新构建)。
|
||||
//
|
||||
// 启用前必做(护栏,见设计文档):跨 namespace 公平性检查 + distill 召回配额下限。
|
||||
var (
|
||||
confidenceLambda = envFloat("ZHIYI_CONFIDENCE_LAMBDA", 0) // 0 = 关闭
|
||||
confidenceMean = envFloat("ZHIYI_CONFIDENCE_MEAN", 0.7) // 去中心基准
|
||||
)
|
||||
|
||||
func envFloat(key string, def float64) float64 {
|
||||
if v := os.Getenv(key); v != "" {
|
||||
if f, err := strconv.ParseFloat(v, 64); err == nil {
|
||||
return f
|
||||
}
|
||||
log.Printf("[confidence] %s=%q 无法解析为浮点,回落默认 %v", key, os.Getenv(key), def)
|
||||
}
|
||||
return def
|
||||
}
|
||||
|
||||
// coalesceConfidence 2026-10-08 (Gap#2):NULL(存量未声明)⇒ 1.0。
|
||||
// 这是「加权首日 no-op」的实现基础:存量行不因缺字段被系统性惩罚。
|
||||
// 注意这是**读时兜底**,不是回填——库里仍是 NULL,可随时区分「未声明」与「声明为 1.0」。
|
||||
func coalesceConfidence(c *float64) float64 {
|
||||
if c == nil {
|
||||
return 1.0
|
||||
}
|
||||
return *c
|
||||
}
|
||||
|
||||
// MissRecorder 缺口记录接口(由 routes 层注入 gapDetector.RecordMiss)
|
||||
type MissRecorder func(query, namespace string)
|
||||
|
||||
@@ -69,9 +108,14 @@ type RecallMeta struct {
|
||||
// RecallWithMeta 同 Recall,但额外返回召回元数据(候选池真实大小)。
|
||||
// 缓存命中时无法重建候选池 ⇒ CandidatesScanned = 0(保守:无法测量就报 0,不虚报)。
|
||||
func (p *RecallPipeline) RecallWithMeta(query, namespace string, topK int, diversity float64) ([]models.RecallResult, RecallMeta, error) {
|
||||
// topK 先归一:缓存键必须与写入侧用的 topK 一致。recallUncached 内部也会归一,
|
||||
// 若此处不先归一,topK<=0 的调用读到的是"归一前的键"、永远命中不了自己写的条目。
|
||||
if topK <= 0 {
|
||||
topK = 10
|
||||
}
|
||||
// 缓存命中分支:直接返回,候选池不可重建
|
||||
if SearchCacheInstance != nil {
|
||||
if cached, ok := SearchCacheInstance.Get(query, namespace); ok {
|
||||
if cached, ok := SearchCacheInstance.Get(query, namespace, topK, diversity); ok {
|
||||
var results []models.RecallResult
|
||||
if err := json.Unmarshal(cached, &results); err == nil {
|
||||
filtered := filterRecallRawCategories(results)
|
||||
@@ -224,6 +268,25 @@ func (p *RecallPipeline) recallUncached(query, namespace string, topK int, diver
|
||||
}
|
||||
}
|
||||
|
||||
// Step 3.7: 来源加权偏置(2026-10-09 P0 修正)
|
||||
//
|
||||
// 修前的错位:λ 只加在 Step 4 之后的**展示分**上,而候选的取舍早已由 rerank+MMR 定死
|
||||
// ⇒ λ 无论取多大都只改返回的数字、不改结果集合。实测(λ=0.1 vs 0,同一批 14 条 query,
|
||||
// 全部 cache-miss):逐条分数位移与理论吻合(legacy -0.0277 / conf1.0 +0.0294),
|
||||
// 但 top-10 集合与 legacy 占比 33.57% 两轮完全相同 —— 典型的"接上了但接在阀门下游"。
|
||||
//
|
||||
// 现在把偏置移到 MMR 之前:选择与展示共用同一套分数(rel = r.Score,见 MMRSelect)。
|
||||
// λ=0 时本循环不执行 ⇒ 与旧实现逐字节等价,回滚只需把环境变量置 0。
|
||||
if confidenceLambda > 0 {
|
||||
for i := range reranked {
|
||||
idx := reranked[i].Index
|
||||
if idx >= 0 && idx < len(candidates) {
|
||||
conf := coalesceConfidence(candidates[idx].Confidence)
|
||||
reranked[i].Score += confidenceLambda * (conf - confidenceMean)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Step 4: MMR diversity
|
||||
finalIndices := MMRSelect(reranked, candidates, topK*2, diversity)
|
||||
|
||||
@@ -241,17 +304,34 @@ func (p *RecallPipeline) recallUncached(query, namespace string, topK int, diver
|
||||
break
|
||||
}
|
||||
}
|
||||
// 2026-10-08 (Gap#2 来源加权):把信任等级直通给调用方,并按门控参与排序。
|
||||
//
|
||||
// 为什么默认关闭(λ=0):存量 15105 条 confidence 全为 NULL,此时任何加权都只是
|
||||
// 在噪声上做乘法。按 MoA 评审:先跑一周采分布 → 定 λ 与形式 → 再带护栏启用。
|
||||
// 因此 P0 只接线不改行为(no-op),启用是改环境变量,不是改代码。
|
||||
//
|
||||
// 为什么是**加性去中心偏置**而非乘性:
|
||||
// 乘性 base*conf 会把全部 LLM 蒸馏产物整体压沉(memories 表 100% 经蒸馏,
|
||||
// 见 server.go 唯一 InsertMemory 调用点),并与 QualityScore 在同一根轴上二次惩罚;
|
||||
// 加性 base+λ(conf-mean) 只平移均值、不缩方差,对尾部排序影响可控。
|
||||
conf := coalesceConfidence(c.Confidence)
|
||||
// 2026-10-09 (P0 修正):λ 已在 Step 3.7 偏置进 reranked[i].Score,选择与展示
|
||||
// 共用同一套分数;此处不再重复施加(否则 legacy 被双重扣分)。
|
||||
base := score*0.7 + c.QualityScore*0.3
|
||||
results = append(results, models.RecallResult{
|
||||
ID: c.ID,
|
||||
Content: c.Content,
|
||||
Category: c.Category,
|
||||
Score: score*0.7 + c.QualityScore*0.3,
|
||||
Score: base,
|
||||
Timestamp: c.CreatedAt.Format(time.RFC3339),
|
||||
// 2026-09-22:带上来源,供全库召回时分辨记忆归属域
|
||||
Namespace: c.Namespace,
|
||||
AgentID: c.AgentID,
|
||||
// 2026-09-26:把记忆出处直通给调用方 —— 蒸馏/审核记忆的 source 形如 "wiki:<笔记路径>"
|
||||
Source: c.Source,
|
||||
// 2026-10-08 (Gap#2):信任等级直通(NULL 已 COALESCE 成 1.0)
|
||||
Confidence: &conf,
|
||||
ConfidenceSource: c.ConfidenceSource,
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -284,9 +364,9 @@ func (p *RecallPipeline) recallUncached(query, namespace string, topK int, diver
|
||||
}
|
||||
results = deduped
|
||||
|
||||
// Step 6.5: 写搜索缓存(§2.6 — TTL 1h,下次相同 query 直接返回)
|
||||
// Step 6.5: 写搜索缓存(§2.6 — TTL 1h,下次相同 query+topK+diversity 直接返回)
|
||||
if SearchCacheInstance != nil {
|
||||
SearchCacheInstance.Set(query, namespace, results)
|
||||
SearchCacheInstance.Set(query, namespace, topK, diversity, results)
|
||||
}
|
||||
|
||||
// Step 7: 预取推送(CO_OCCURS 权重 > 0.6 的配套记忆 → WebSocket)
|
||||
|
||||
@@ -2,6 +2,7 @@ package storage
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/xiaoxue/memoryweave/internal/models"
|
||||
)
|
||||
@@ -33,7 +34,17 @@ const nsTestQuery = "蒸馏模型日限计数器跨天重置"
|
||||
|
||||
// newNSTestPipeline 构造一个不依赖网络/磁盘的 RecallPipeline:
|
||||
// Embedder 走预置缓存(命中缓存即不发起 HTTP),Reranker 无 key 无 embedder → 优雅降级。
|
||||
func newNSTestPipeline(fake *nsRecordingLance) *RecallPipeline {
|
||||
//
|
||||
// ⚠️ 必须隔离进程级搜索缓存(SearchCacheInstance):
|
||||
// 它是带 Redis L2 的包级单例,**跨 test 运行持久**。本组用例断言的是
|
||||
// 「Search 被调用几次 / 收到什么过滤」,一旦命中缓存,RecallWithMeta 会在查库前
|
||||
// 直接返回 ⇒ Search 调用次数恒为 0 ⇒ 用例假红(实测:置 nil 后 3 例全绿)。
|
||||
func newNSTestPipeline(t *testing.T, fake *nsRecordingLance) *RecallPipeline {
|
||||
t.Helper()
|
||||
saved := SearchCacheInstance
|
||||
SearchCacheInstance = nil
|
||||
t.Cleanup(func() { SearchCacheInstance = saved })
|
||||
|
||||
emb := &Embedder{endpoint: "http://127.0.0.1:1/never", dim: 1024}
|
||||
vec := make([]float32, 1024)
|
||||
for i := range vec {
|
||||
@@ -50,7 +61,7 @@ func TestRecallEmptyNamespaceSearchesAll(t *testing.T) {
|
||||
fake := &nsRecordingLance{rows: []models.MemoryRecord{
|
||||
{ID: "m1", Content: "记忆 A", Category: "distilled"},
|
||||
}}
|
||||
p := newNSTestPipeline(fake)
|
||||
p := newNSTestPipeline(t, fake)
|
||||
|
||||
if _, err := p.Recall(nsTestQuery, "", 5, 0.3); err != nil {
|
||||
t.Fatalf("Recall 出错: %v", err)
|
||||
@@ -73,7 +84,7 @@ func TestRecallExplicitNamespaceStaysScoped(t *testing.T) {
|
||||
fake := &nsRecordingLance{rows: []models.MemoryRecord{
|
||||
{ID: "m1", Content: "记忆 A", Category: "distilled"},
|
||||
}}
|
||||
p := newNSTestPipeline(fake)
|
||||
p := newNSTestPipeline(t, fake)
|
||||
|
||||
if _, err := p.Recall(nsTestQuery, "hermes-main", 5, 0.3); err != nil {
|
||||
t.Fatalf("Recall 出错: %v", err)
|
||||
@@ -95,7 +106,7 @@ func TestRecallSharedNamespaceNoDoubleSearch(t *testing.T) {
|
||||
fake := &nsRecordingLance{rows: []models.MemoryRecord{
|
||||
{ID: "m1", Content: "记忆 A", Category: "distilled"},
|
||||
}}
|
||||
p := newNSTestPipeline(fake)
|
||||
p := newNSTestPipeline(t, fake)
|
||||
|
||||
if _, err := p.Recall(nsTestQuery, "shared", 5, 0.3); err != nil {
|
||||
t.Fatalf("Recall 出错: %v", err)
|
||||
@@ -105,3 +116,26 @@ func TestRecallSharedNamespaceNoDoubleSearch(t *testing.T) {
|
||||
}
|
||||
t.Logf("✓ shared 不重复搜索:%v", fake.nsFilters)
|
||||
}
|
||||
|
||||
// TestRecallCachedPathSkipsSearch 反证用例:证明上面 3 例的隔离是**必要**的。
|
||||
// 装入真实的搜索缓存并预置同 (query,namespace,topK,diversity) 的条目,
|
||||
// Recall 必须走缓存分支、不查库 —— 这正是旧版恒假红的机制(Search 调用次数为 0)。
|
||||
func TestRecallCachedPathSkipsSearch(t *testing.T) {
|
||||
fake := &nsRecordingLance{rows: []models.MemoryRecord{
|
||||
{ID: "m1", Content: "记忆 A", Category: "distilled"},
|
||||
}}
|
||||
emb := &Embedder{endpoint: "http://127.0.0.1:1/never", dim: 1024}
|
||||
p := NewRecallPipeline(emb, fake, &Reranker{})
|
||||
|
||||
saved := SearchCacheInstance
|
||||
SearchCacheInstance = NewSearchCache(100, time.Hour)
|
||||
t.Cleanup(func() { SearchCacheInstance = saved })
|
||||
SearchCacheInstance.Set("任意 query", "hermes-main", 5, 0.3, []models.RecallResult{{ID: "cached"}})
|
||||
|
||||
if _, err := p.Recall("任意 query", "hermes-main", 5, 0.3); err != nil {
|
||||
t.Fatalf("Recall 出错: %v", err)
|
||||
}
|
||||
if len(fake.nsFilters) != 0 {
|
||||
t.Errorf("缓存命中时不应查库,实际 Search 被调用 %d 次", len(fake.nsFilters))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -361,6 +361,65 @@ func (rc *RedisCache) Del(key string) {
|
||||
rc.mu.Unlock()
|
||||
}
|
||||
|
||||
// DelPrefix 按前缀批量删除缓存键(SCAN 游标遍历 + 批量 DEL)。
|
||||
// 不用 KEYS:KEYS 在大 keyspace 下会阻塞 Redis 单线程。
|
||||
// 键的实际形态是 "zhiyi:cache:" + prefix + 其余(见 Get/Set 的键前缀约定)。
|
||||
func (rc *RedisCache) DelPrefix(prefix string) int {
|
||||
deleted := 0
|
||||
|
||||
// 无 Redis 时的进程内兜底层
|
||||
rc.mu.Lock()
|
||||
for k := range rc.fallback {
|
||||
if strings.HasPrefix(k, prefix) {
|
||||
delete(rc.fallback, k)
|
||||
deleted++
|
||||
}
|
||||
}
|
||||
rc.mu.Unlock()
|
||||
|
||||
if rc.client == nil || rc.client.conn == nil {
|
||||
return deleted
|
||||
}
|
||||
|
||||
pattern := "zhiyi:cache:" + prefix + "*"
|
||||
cursor := "0"
|
||||
for {
|
||||
res, err := rc.client.conn.Do("SCAN", cursor, "MATCH", pattern, "COUNT", "500")
|
||||
if err != nil {
|
||||
return deleted
|
||||
}
|
||||
arr, ok := res.([]interface{})
|
||||
if !ok || len(arr) != 2 {
|
||||
return deleted
|
||||
}
|
||||
next, ok := arr[0].(string)
|
||||
if !ok {
|
||||
return deleted
|
||||
}
|
||||
cursor = next
|
||||
|
||||
if keys, ok := arr[1].([]interface{}); ok && len(keys) > 0 {
|
||||
args := []string{"DEL"}
|
||||
for _, k := range keys {
|
||||
if s, ok := k.(string); ok {
|
||||
args = append(args, s)
|
||||
}
|
||||
}
|
||||
if len(args) > 1 {
|
||||
if n, err := rc.client.conn.Do(args...); err == nil {
|
||||
if cnt, ok := n.(int64); ok {
|
||||
deleted += int(cnt)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if cursor == "0" {
|
||||
break
|
||||
}
|
||||
}
|
||||
return deleted
|
||||
}
|
||||
|
||||
// ─── Redis 心跳 ───────────────────────────────────────
|
||||
|
||||
type Heartbeat struct {
|
||||
|
||||
@@ -6,6 +6,8 @@ import (
|
||||
"crypto/sha256"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
@@ -15,6 +17,8 @@ type L2CacheBackend interface {
|
||||
Get(key string) ([]byte, bool)
|
||||
Set(key string, data []byte, ttl time.Duration)
|
||||
Del(key string)
|
||||
// DelPrefix 按前缀批量删除(按 namespace / query 失效用,避免逐键枚举)
|
||||
DelPrefix(prefix string) int
|
||||
}
|
||||
|
||||
// redisL2Adapter 将同包的 RedisCache 适配为 L2CacheBackend
|
||||
@@ -42,6 +46,13 @@ func (a *redisL2Adapter) Del(key string) {
|
||||
}
|
||||
rc.Del(key)
|
||||
}
|
||||
func (a *redisL2Adapter) DelPrefix(prefix string) int {
|
||||
rc := getRedisCacheInstance()
|
||||
if rc == nil {
|
||||
return 0
|
||||
}
|
||||
return rc.DelPrefix(prefix)
|
||||
}
|
||||
|
||||
// getRedisCacheInstance 单例获取 RedisCache(延迟初始化)
|
||||
var (
|
||||
@@ -60,11 +71,11 @@ func getRedisCacheInstance() *RedisCache {
|
||||
// Get 时:L1 miss → 查 Redis L2 → 回填 L1
|
||||
// Set 时:写 L1 + 写 Redis L2(TTL 同步)
|
||||
type SearchCache struct {
|
||||
mu sync.RWMutex
|
||||
entries map[string]*CacheEntry
|
||||
maxSize int
|
||||
ttl time.Duration
|
||||
l2 L2CacheBackend // 可选 L2 缓存(Redis),nil 时只有 L1
|
||||
mu sync.RWMutex
|
||||
entries map[string]*CacheEntry
|
||||
maxSize int
|
||||
ttl time.Duration
|
||||
l2 L2CacheBackend // 可选 L2 缓存(Redis),nil 时只有 L1
|
||||
}
|
||||
|
||||
type CacheEntry struct {
|
||||
@@ -97,8 +108,12 @@ func NewSearchCacheWithRedis(maxSize int, ttl time.Duration) *SearchCache {
|
||||
}
|
||||
|
||||
// Get 获取缓存结果:L1 miss → 查 L2 → 回填 L1
|
||||
func (sc *SearchCache) Get(query, namespace string) ([]byte, bool) {
|
||||
key := cacheKey(query, namespace)
|
||||
//
|
||||
// topK / diversity 参与键构造:召回结果随它们变化(rerank/MMR 取 topK*2、
|
||||
// 最终按 topK 截断),而调用方 topK 各不相同(eval 用 3 / 5,HTTP 用客户端 Limit)
|
||||
// ⇒ 键里不含它们会让不同调用方串用同一条缓存、拿到条数不对的结果。
|
||||
func (sc *SearchCache) Get(query, namespace string, topK int, diversity float64) ([]byte, bool) {
|
||||
key := cacheKey(query, namespace, topK, diversity)
|
||||
|
||||
// L1 查找
|
||||
sc.mu.RLock()
|
||||
@@ -135,8 +150,8 @@ func (sc *SearchCache) Get(query, namespace string) ([]byte, bool) {
|
||||
}
|
||||
|
||||
// Set 写入缓存:写 L1 + 写 L2
|
||||
func (sc *SearchCache) Set(query, namespace string, results interface{}) {
|
||||
key := cacheKey(query, namespace)
|
||||
func (sc *SearchCache) Set(query, namespace string, topK int, diversity float64, results interface{}) {
|
||||
key := cacheKey(query, namespace, topK, diversity)
|
||||
data, err := json.Marshal(results)
|
||||
if err != nil {
|
||||
return
|
||||
@@ -168,37 +183,46 @@ func (sc *SearchCache) Stats() map[string]interface{} {
|
||||
sc.mu.RLock()
|
||||
defer sc.mu.RUnlock()
|
||||
return map[string]interface{}{
|
||||
"size": len(sc.entries),
|
||||
"max_size": sc.maxSize,
|
||||
"ttl": sc.ttl.String(),
|
||||
"size": len(sc.entries),
|
||||
"max_size": sc.maxSize,
|
||||
"ttl": sc.ttl.String(),
|
||||
"l2_enabled": sc.l2 != nil,
|
||||
}
|
||||
}
|
||||
|
||||
// Invalidate 使指定 namespace 的缓存失效(L1 + L2)
|
||||
// Invalidate 使指定 namespace 的缓存失效(L1 + L2)。
|
||||
//
|
||||
// 旧实现把 entry.Results 反序列化成 map[string]interface{} 再取 ["namespace"],
|
||||
// 但 Set 存进去的是 []RecallResult(JSON 数组)⇒ 反序列化恒失败 ⇒ parsed 为 nil
|
||||
// ⇒ 一个键都删不掉:失效是**静默空操作**(探针实测:Invalidate 后缓存仍命中,
|
||||
// 记忆写入后最长 1h TTL 内持续返回陈旧召回)。
|
||||
// 现改为按键布局前缀匹配,不再依赖 payload 结构。
|
||||
func (sc *SearchCache) Invalidate(namespace string) {
|
||||
prefix := cacheKeyNSPrefix(namespace)
|
||||
sc.mu.Lock()
|
||||
defer sc.mu.Unlock()
|
||||
for key, entry := range sc.entries {
|
||||
var parsed map[string]interface{}
|
||||
json.Unmarshal(entry.Results, &parsed)
|
||||
if ns, ok := parsed["namespace"].(string); ok && ns == namespace {
|
||||
for key := range sc.entries {
|
||||
if strings.HasPrefix(key, prefix) {
|
||||
delete(sc.entries, key)
|
||||
if sc.l2 != nil {
|
||||
sc.l2.Del(key)
|
||||
}
|
||||
}
|
||||
}
|
||||
sc.mu.Unlock()
|
||||
if sc.l2 != nil {
|
||||
sc.l2.DelPrefix(prefix)
|
||||
}
|
||||
}
|
||||
|
||||
// InvalidateKey 使指定 key 失效(L1 + L2)
|
||||
// InvalidateKey 使指定 (query, namespace) 的**全部 topK/diversity 变体**失效(L1 + L2)
|
||||
func (sc *SearchCache) InvalidateKey(query, namespace string) {
|
||||
key := cacheKey(query, namespace)
|
||||
prefix := cacheKeyQueryPrefix(query, namespace)
|
||||
sc.mu.Lock()
|
||||
delete(sc.entries, key)
|
||||
for key := range sc.entries {
|
||||
if strings.HasPrefix(key, prefix) {
|
||||
delete(sc.entries, key)
|
||||
}
|
||||
}
|
||||
sc.mu.Unlock()
|
||||
if sc.l2 != nil {
|
||||
sc.l2.Del(key)
|
||||
sc.l2.DelPrefix(prefix)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -233,9 +257,24 @@ func (sc *SearchCache) reaper() {
|
||||
}
|
||||
}
|
||||
|
||||
func cacheKey(query, namespace string) string {
|
||||
h := sha256.Sum256([]byte(query + "|" + namespace))
|
||||
return fmt.Sprintf("%x", h[:16])
|
||||
// cacheKey 构造缓存键。布局刻意做成「可前缀寻址」:
|
||||
//
|
||||
// <namespace> \x00 <queryHash(16B hex)> \x00 <topK> \x00 <diversity>
|
||||
//
|
||||
// 为什么 namespace 放最前:失效按 namespace 做前缀匹配,无需解析 payload。
|
||||
func cacheKey(query, namespace string, topK int, diversity float64) string {
|
||||
h := sha256.Sum256([]byte(query))
|
||||
return fmt.Sprintf("%s\x00%x\x00%d\x00%s", namespace, h[:16], topK,
|
||||
strconv.FormatFloat(diversity, 'f', -1, 64))
|
||||
}
|
||||
|
||||
// cacheKeyNSPrefix 某 namespace 下全部键的前缀
|
||||
func cacheKeyNSPrefix(namespace string) string { return namespace + "\x00" }
|
||||
|
||||
// cacheKeyQueryPrefix 某 (query, namespace) 下全部 topK/diversity 变体的前缀
|
||||
func cacheKeyQueryPrefix(query, namespace string) string {
|
||||
h := sha256.Sum256([]byte(query))
|
||||
return fmt.Sprintf("%s\x00%x\x00", namespace, h[:16])
|
||||
}
|
||||
|
||||
// ─── 全局搜索缓存实例(G9:带 Redis L2) ─────────────────────
|
||||
@@ -243,4 +282,4 @@ var SearchCacheInstance *SearchCache
|
||||
|
||||
func init() {
|
||||
SearchCacheInstance = NewSearchCacheWithRedis(1000, 1*time.Hour)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,135 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/xiaoxue/memoryweave/internal/models"
|
||||
)
|
||||
|
||||
func mustResults(n int) []models.RecallResult {
|
||||
out := make([]models.RecallResult, n)
|
||||
for i := range out {
|
||||
out[i] = models.RecallResult{ID: string(rune('a' + i)), Content: "c"}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func getLen(t *testing.T, c *SearchCache, q, ns string, topK int, div float64) (int, bool) {
|
||||
t.Helper()
|
||||
raw, ok := c.Get(q, ns, topK, div)
|
||||
if !ok {
|
||||
return 0, false
|
||||
}
|
||||
var got []models.RecallResult
|
||||
if err := json.Unmarshal(raw, &got); err != nil {
|
||||
t.Fatalf("缓存内容不是 RecallResult 数组: %v", err)
|
||||
}
|
||||
return len(got), true
|
||||
}
|
||||
|
||||
// 回归(核心):键必须区分 topK。
|
||||
// 旧键 cacheKey(query,namespace) 不含 topK ⇒ eval(topK=3) 与 eval(topK=5)/HTTP(req.Limit)
|
||||
// 会串用同一条缓存,调用方拿到条数不对的结果。
|
||||
func TestSearchCacheKeyDistinguishesTopK(t *testing.T) {
|
||||
c := NewSearchCache(100, time.Hour)
|
||||
c.Set("同一句话", "hermes-main", 5, 0.3, mustResults(5))
|
||||
c.Set("同一句话", "hermes-main", 3, 0.3, mustResults(3))
|
||||
|
||||
n5, ok := getLen(t, c, "同一句话", "hermes-main", 5, 0.3)
|
||||
if !ok {
|
||||
t.Fatal("topK=5 应命中")
|
||||
}
|
||||
if n5 != 5 {
|
||||
t.Errorf("topK=5 应取回 5 条,实际 %d 条 ⇒ topK 未进键", n5)
|
||||
}
|
||||
n3, ok := getLen(t, c, "同一句话", "hermes-main", 3, 0.3)
|
||||
if !ok {
|
||||
t.Fatal("topK=3 应命中")
|
||||
}
|
||||
if n3 != 3 {
|
||||
t.Errorf("topK=3 应取回 3 条,实际 %d 条 ⇒ topK 未进键", n3)
|
||||
}
|
||||
}
|
||||
|
||||
// 回归:键必须区分 diversity(MMR 选择随其变化)。
|
||||
func TestSearchCacheKeyDistinguishesDiversity(t *testing.T) {
|
||||
c := NewSearchCache(100, time.Hour)
|
||||
c.Set("q", "ns", 5, 0.3, mustResults(3))
|
||||
if _, ok := c.Get("q", "ns", 5, 0.5); ok {
|
||||
t.Error("diversity=0.5 命中了 divergence=0.3 的条目 ⇒ diversity 未进键")
|
||||
}
|
||||
if _, ok := c.Get("q", "ns", 5, 0.3); !ok {
|
||||
t.Error("相同 diversity 应命中")
|
||||
}
|
||||
}
|
||||
|
||||
// 回归:namespace 与 query 之间必须无键碰撞(旧键用 '|' 拼接后整体哈希,
|
||||
// 新键用 \x00 分隔且 namespace 在前,需确认前缀关系不误伤)。
|
||||
func TestSearchCacheKeyNoCrossNamespaceCollision(t *testing.T) {
|
||||
c := NewSearchCache(100, time.Hour)
|
||||
c.Set("q", "ns-a", 5, 0.3, mustResults(1))
|
||||
if _, ok := c.Get("q", "ns-ab", 5, 0.3); ok {
|
||||
t.Error("namespace 前缀相近时串了缓存(ns-a vs ns-ab)")
|
||||
}
|
||||
if _, ok := c.Get("q", "ns-a", 5, 0.3); !ok {
|
||||
t.Error("同 namespace 应命中")
|
||||
}
|
||||
}
|
||||
|
||||
// 回归(真实 bug):Invalidate(namespace) 曾对数组型 payload 反序列化失败 ⇒ 静默空操作。
|
||||
// 现为前缀匹配,必须真的删掉,且不得误删其他 namespace。
|
||||
func TestSearchCacheInvalidateByNamespace(t *testing.T) {
|
||||
c := NewSearchCache(100, time.Hour)
|
||||
c.Set("q1", "hermes-main", 5, 0.3, mustResults(2))
|
||||
c.Set("q2", "hermes-main", 3, 0.5, mustResults(2))
|
||||
c.Set("q1", "wiki", 5, 0.3, mustResults(2))
|
||||
|
||||
c.Invalidate("hermes-main")
|
||||
|
||||
if _, ok := c.Get("q1", "hermes-main", 5, 0.3); ok {
|
||||
t.Error("Invalidate(hermes-main) 后 q1 仍命中 ⇒ 失效是空操作")
|
||||
}
|
||||
if _, ok := c.Get("q2", "hermes-main", 3, 0.5); ok {
|
||||
t.Error("Invalidate(hermes-main) 后 q2 仍命中 ⇒ 未按 namespace 全量失效")
|
||||
}
|
||||
if _, ok := c.Get("q1", "wiki", 5, 0.3); !ok {
|
||||
t.Error("Invalidate(hermes-main) 误删了 wiki 的条目")
|
||||
}
|
||||
}
|
||||
|
||||
// 回归:InvalidateKey(query, namespace) 必须清掉该 query 的全部 topK/diversity 变体。
|
||||
func TestSearchCacheInvalidateKeyAllVariants(t *testing.T) {
|
||||
c := NewSearchCache(100, time.Hour)
|
||||
c.Set("q", "ns", 5, 0.3, mustResults(1))
|
||||
c.Set("q", "ns", 3, 0.5, mustResults(1))
|
||||
c.Set("q-别的", "ns", 5, 0.3, mustResults(1))
|
||||
|
||||
c.InvalidateKey("q", "ns")
|
||||
|
||||
if _, ok := c.Get("q", "ns", 5, 0.3); ok {
|
||||
t.Error("InvalidateKey 未清掉 (topK=5,div=0.3) 变体")
|
||||
}
|
||||
if _, ok := c.Get("q", "ns", 3, 0.5); ok {
|
||||
t.Error("InvalidateKey 未清掉 (topK=3,div=0.5) 变体")
|
||||
}
|
||||
if _, ok := c.Get("q-别的", "ns", 5, 0.3); !ok {
|
||||
t.Error("InvalidateKey 误删了别的 query")
|
||||
}
|
||||
}
|
||||
|
||||
// 反证:单一 query 下 namespace 是键的第一段,前缀匹配不会因 query 哈希前缀相同而误删。
|
||||
func TestSearchCacheInvalidatePrefixIsExact(t *testing.T) {
|
||||
c := NewSearchCache(100, time.Hour)
|
||||
// "a" 与 "ab" 做 namespace:Invalidate("a") 不得删掉 "ab"
|
||||
c.Set("q", "a", 5, 0.3, mustResults(1))
|
||||
c.Set("q", "ab", 5, 0.3, mustResults(1))
|
||||
c.Invalidate("a")
|
||||
if _, ok := c.Get("q", "a", 5, 0.3); ok {
|
||||
t.Error("Invalidate(a) 未生效")
|
||||
}
|
||||
if _, ok := c.Get("q", "ab", 5, 0.3); !ok {
|
||||
t.Error("Invalidate(a) 前缀误伤 'ab'(separator 未生效)")
|
||||
}
|
||||
}
|
||||
+60
-3
@@ -14,7 +14,7 @@ use std::sync::Arc;
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use arrow::array::{ArrayRef, BooleanArray, FixedSizeListArray, Float32Array, Float64Array, Int64Array, StringArray};
|
||||
use arrow::array::{Array, ArrayRef, BooleanArray, FixedSizeListArray, Float32Array, Float64Array, Int64Array, StringArray};
|
||||
use arrow::datatypes::{DataType, Field, Schema};
|
||||
use arrow::record_batch::{RecordBatch, RecordBatchIterator};
|
||||
use futures::StreamExt;
|
||||
@@ -31,6 +31,12 @@ pub struct MemoryRecord {
|
||||
pub version_history: String, pub source: String,
|
||||
// 2026-10-08 (t_7e3577c5):血缘字段
|
||||
pub embedding_model: String, pub script_version: String,
|
||||
// 2026-10-08 (Gap#2):信任等级。Option<f64> —— 存量行是 NULL,必须与「显式 0.0」区分。
|
||||
// 漏了这里 ⇒ IPC 响应无该字段 ⇒ Go 侧解包恒 nil ⇒ 加列也白加(静默 no-op)。
|
||||
#[serde(default)]
|
||||
pub confidence: Option<f64>,
|
||||
#[serde(default)]
|
||||
pub confidence_source: String,
|
||||
pub volatile_flag: bool, pub is_deleted: bool,
|
||||
pub depends_on: String, pub derived_from: String,
|
||||
pub last_recalled_at: String, pub created_at: String, pub updated_at: String,
|
||||
@@ -68,6 +74,12 @@ fn arrow_schema() -> Schema {
|
||||
Field::new("last_recalled_at", DataType::Utf8, true),
|
||||
Field::new("created_at", DataType::Utf8, false),
|
||||
Field::new("updated_at", DataType::Utf8, false),
|
||||
// 2026-10-08 (Gap#2):来源加权。confidence=写入方显式声明的信任等级(不从 source 反推);
|
||||
// confidence_source=authoritative|estimated,用于区分「写入时声明」与「事后回填猜测」,
|
||||
// 是日后批量纠偏的唯一手段(否则无法定向打 estimated 行)。
|
||||
// 均 nullable:存量行保持 NULL,读路径 COALESCE→1.0,使加权上线首日即 no-op。
|
||||
Field::new("confidence", DataType::Float64, true),
|
||||
Field::new("confidence_source", DataType::Utf8, true),
|
||||
])
|
||||
}
|
||||
|
||||
@@ -92,8 +104,19 @@ impl LanceDBOps {
|
||||
let mut missing: Vec<(String, String)> = Vec::new();
|
||||
for f in target.fields() {
|
||||
if tbl_schema.field_with_name(f.name()).is_err() {
|
||||
// 缺失列补 Utf8 nullable,默认空串
|
||||
missing.push((f.name().to_string(), "''".to_string()));
|
||||
// 缺失列一律补**可空**(2026-10-08 修正:原先字符串给 '',
|
||||
// 而 add_columns 的 SQL 字面量默认值会把列标成 **non-null** ——
|
||||
// 与 arrow_schema 的 nullable=true 冲突,写入 NULL 会被 lance 拒绝:
|
||||
// "contained null values even though the field is marked non-null"
|
||||
// 实测对照:embedding_model/script_version(历史迁移列)都是 nullable=true。
|
||||
let default = match f.data_type() {
|
||||
DataType::Float64 => "CAST(NULL AS DOUBLE)",
|
||||
DataType::Float32 => "CAST(NULL AS FLOAT)",
|
||||
DataType::Int64 => "CAST(NULL AS BIGINT)",
|
||||
DataType::Boolean => "CAST(NULL AS BOOLEAN)",
|
||||
_ => "CAST(NULL AS STRING)",
|
||||
};
|
||||
missing.push((f.name().to_string(), default.to_string()));
|
||||
}
|
||||
}
|
||||
if !missing.is_empty() {
|
||||
@@ -154,6 +177,9 @@ impl LanceDBOps {
|
||||
last_recalled_at: String::new(),
|
||||
created_at: col_str(&batch, i, "created_at"),
|
||||
updated_at: col_str(&batch, i, "updated_at"),
|
||||
// 2026-10-08 (Gap#2):读侧接线。col_f64_opt 保留 NULL 语义
|
||||
confidence: col_f64_opt(&batch, i, "confidence"),
|
||||
confidence_source: col_str(&batch, i, "confidence_source"),
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -203,6 +229,9 @@ impl LanceDBOps {
|
||||
last_recalled_at: String::new(),
|
||||
created_at: col_str(&batch, i, "created_at"),
|
||||
updated_at: col_str(&batch, i, "updated_at"),
|
||||
// 2026-10-08 (Gap#2):读侧接线。col_f64_opt 保留 NULL 语义
|
||||
confidence: col_f64_opt(&batch, i, "confidence"),
|
||||
confidence_source: col_str(&batch, i, "confidence_source"),
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -229,6 +258,9 @@ impl LanceDBOps {
|
||||
let mut volatile_flags = Vec::new(); let mut is_deleteds = Vec::new();
|
||||
let mut depends_ons = Vec::new(); let mut derived_froms = Vec::new();
|
||||
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();
|
||||
let mut confidence_sources: Vec<&str> = Vec::new();
|
||||
|
||||
for r in &records {
|
||||
ids.push(r["id"].as_str().unwrap_or(""));
|
||||
@@ -259,6 +291,12 @@ impl LanceDBOps {
|
||||
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(""));
|
||||
// confidence:显式声明才写,否则 NULL(不从 source 反推)——
|
||||
// **NULL 本身就是「未声明」的语义信号**,confidence_source 只是标签。
|
||||
confidences.push(r["confidence"].as_f64());
|
||||
// confidence_source 恒给字符串(未声明 ⇒ ""):迁移来的列可能是 non-null,
|
||||
// 送 None 会被 lance 拒收。语义仍由 confidence=NULL 承担。
|
||||
confidence_sources.push(r["confidence_source"].as_str().unwrap_or(""));
|
||||
}
|
||||
|
||||
// 构建 vector 列 (FixedSizeList<Float32, 1024>)
|
||||
@@ -297,6 +335,9 @@ impl LanceDBOps {
|
||||
Arc::new(StringArray::from(last_recalled_ats)),
|
||||
Arc::new(StringArray::from(created_ats)),
|
||||
Arc::new(StringArray::from(updated_ats)),
|
||||
// 2026-10-08 (Gap#2):顺序必须与 arrow_schema() 严格一致
|
||||
Arc::new(Float64Array::from(confidences)),
|
||||
Arc::new(StringArray::from(confidence_sources)),
|
||||
];
|
||||
|
||||
let batch = RecordBatch::try_new(schema.clone(), columns)?;
|
||||
@@ -638,6 +679,9 @@ impl LanceDBOps {
|
||||
last_recalled_at: col_str(&batch, i, "last_recalled_at"), // ✅ 真读(勿学 scan_all 的 String::new())
|
||||
created_at: col_str(&batch, i, "created_at"),
|
||||
updated_at: col_str(&batch, i, "updated_at"),
|
||||
// 2026-10-08 (Gap#2):读侧接线。col_f64_opt 保留 NULL 语义
|
||||
confidence: col_f64_opt(&batch, i, "confidence"),
|
||||
confidence_source: col_str(&batch, i, "confidence_source"),
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -658,6 +702,8 @@ impl LanceDBOps {
|
||||
"useful_count", "not_useful_count", "freshness", "version",
|
||||
"source", "volatile_flag", "is_deleted",
|
||||
"last_recalled_at", "created_at", "updated_at",
|
||||
// 2026-10-08 (Gap#2):scan_all 用投影裁剪,漏加 ⇒ col_f64_opt 取不到列 ⇒ 恒 None
|
||||
"confidence", "confidence_source",
|
||||
];
|
||||
|
||||
let mut results = Box::pin(rt().block_on(
|
||||
@@ -697,6 +743,9 @@ impl LanceDBOps {
|
||||
last_recalled_at: String::new(),
|
||||
created_at: col_str(&batch, i, "created_at"),
|
||||
updated_at: col_str(&batch, i, "updated_at"),
|
||||
// 2026-10-08 (Gap#2):读侧接线。col_f64_opt 保留 NULL 语义
|
||||
confidence: col_f64_opt(&batch, i, "confidence"),
|
||||
confidence_source: col_str(&batch, i, "confidence_source"),
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -751,6 +800,14 @@ fn col_f64(b: &RecordBatch, r: usize, c: &str) -> f64 {
|
||||
.map(|a| a.value(r))
|
||||
.unwrap_or(0.0)
|
||||
}
|
||||
/// col_f64_opt 2026-10-08 (Gap#2):可空浮点读法。
|
||||
/// 与 col_f64 的关键差别:**列不存在**或**该行是 NULL** 都返回 None,而不是 0.0。
|
||||
/// 对 confidence 这是必须的 —— 0.0 会被读成「极不信任」,而 NULL 表示「未声明」。
|
||||
fn col_f64_opt(b: &RecordBatch, r: usize, c: &str) -> Option<f64> {
|
||||
b.column_by_name(c)
|
||||
.and_then(|col| col.as_any().downcast_ref::<Float64Array>())
|
||||
.and_then(|a| if a.is_null(r) { None } else { Some(a.value(r)) })
|
||||
}
|
||||
fn col_i64(b: &RecordBatch, r: usize, c: &str) -> i64 {
|
||||
b.column_by_name(c)
|
||||
.and_then(|col| col.as_any().downcast_ref::<Int64Array>())
|
||||
|
||||
Reference in New Issue
Block a user