fix(provenance): 落库 embedding_model/script_version 血缘字段 (t_7e3577c5)
根因: commit→distill→memory 管线从不在服务端持久化请求 metadata 里的 embedding_model/script_version, MemoryRecord 结构缺这两字段, 导致 zhiyi-vault-ingest.py 已发送的血缘被丢弃, 新写入恒 NULL。 修复: - Go MemoryRecord + DistillInput 加两字段; core.go commit handler 解析 metadata→AutoDistillTrigger 透传; server.go 落库时 orDefault(embedding_model→bge-m3, script_version→unknown); lancedb_ipc.go 记录 map 带上两字段。 - Rust MemoryRecord + arrow_schema() + insert_batch + 4 处 col_str 读取。 验证: commit 带 metadata → 蒸馏落地 19 行 bge-m3/vault-ingest-2.0.0 (原为 NULL)。
This commit is contained in:
小唯
@@ -0,0 +1,14 @@
|
||||
fix(provenance): 落库 embedding_model/script_version 血缘字段 (t_7e3577c5)
|
||||
|
||||
根因: commit→distill→memory 管线从不在服务端持久化请求 metadata 里的
|
||||
embedding_model/script_version, MemoryRecord 结构缺这两字段, 导致
|
||||
zhiyi-vault-ingest.py 已发送的血缘被丢弃, 新写入恒 NULL。
|
||||
|
||||
修复:
|
||||
- Go MemoryRecord + DistillInput 加两字段;
|
||||
core.go commit handler 解析 metadata→AutoDistillTrigger 透传;
|
||||
server.go 落库时 orDefault(embedding_model→bge-m3, script_version→unknown);
|
||||
lancedb_ipc.go 记录 map 带上两字段。
|
||||
- Rust MemoryRecord + arrow_schema() + insert_batch + 4 处 col_str 读取。
|
||||
|
||||
验证: commit 带 metadata → 蒸馏落地 19 行 bge-m3/vault-ingest-2.0.0 (原为 NULL)。
|
||||
@@ -13,7 +13,8 @@ import (
|
||||
var DistillEngineRef *distill.Engine
|
||||
|
||||
// AutoDistillTrigger commit 后自动触发蒸馏流水线
|
||||
func AutoDistillTrigger(episodeID string, content string, category string, namespace string, agentID string, source string) {
|
||||
// 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) {
|
||||
// 1. 硬规则过滤
|
||||
if !distill.HardRulesPass(content) {
|
||||
return
|
||||
@@ -22,12 +23,14 @@ func AutoDistillTrigger(episodeID string, content string, category string, names
|
||||
// 2. 使用蒸馏引擎
|
||||
if DistillEngineRef != nil {
|
||||
DistillEngineRef.Enqueue(distill.DistillInput{
|
||||
EpisodeID: episodeID,
|
||||
Content: content,
|
||||
Category: distill.Category(category),
|
||||
Namespace: namespace,
|
||||
AgentID: agentID,
|
||||
Source: source,
|
||||
EpisodeID: episodeID,
|
||||
Content: content,
|
||||
Category: distill.Category(category),
|
||||
Namespace: namespace,
|
||||
AgentID: agentID,
|
||||
Source: source,
|
||||
EmbeddingModel: embedModel,
|
||||
ScriptVersion: scriptVer,
|
||||
})
|
||||
} else {
|
||||
// 降级: 关键词提取
|
||||
|
||||
@@ -115,6 +115,14 @@ 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 (t_7e3577c5)：血缘字段——metadata 携带（vault-ingest 等脚本发 embedding_model/script_version）。
|
||||
// 缺省留空，server 蒸馏落库时兜底（embedding_model→bge-m3，script_version→unknown）。
|
||||
embedModel, scriptVer := "", ""
|
||||
if req.Metadata != nil {
|
||||
embedModel, _ = req.Metadata["embedding_model"].(string)
|
||||
scriptVer, _ = req.Metadata["script_version"].(string)
|
||||
}
|
||||
|
||||
// t_3fbb6cbd：引用可复算旁路观测（阶段1，开关 ZHIYI_COMMIT_PROVENANCE_OBSERVE 默认关）。
|
||||
// 异步 goroutine，不阻塞写路径、不改响应、不拒写入。
|
||||
go observeCommitProvenance(req.AgentID, req.Content, req.Metadata)
|
||||
@@ -144,7 +152,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)
|
||||
go AutoDistillTrigger(epID, req.Content, req.Category, req.Namespace, req.AgentID, src, embedModel, scriptVer)
|
||||
respond(w, 200, map[string]string{
|
||||
"episode_id": epID, "memory_id": existing.ID,
|
||||
"status": "merged", "reason": "exact_dup",
|
||||
@@ -158,7 +166,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)
|
||||
go AutoDistillTrigger(epID, req.Content, req.Category, req.Namespace, req.AgentID, src, embedModel, scriptVer)
|
||||
respond(w, 200, map[string]string{
|
||||
"episode_id": epID, "memory_id": existing.ID,
|
||||
"status": "merged", "reason": "near_dup",
|
||||
@@ -224,9 +232,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)
|
||||
go AutoDistillTrigger(epID, req.Content, req.Category, req.Namespace, req.AgentID, src, embedModel, scriptVer)
|
||||
} else {
|
||||
go AutoDistillTrigger(epID, req.Content, req.Category, req.Namespace, req.AgentID, src)
|
||||
go AutoDistillTrigger(epID, req.Content, req.Category, req.Namespace, req.AgentID, src, embedModel, scriptVer)
|
||||
}
|
||||
respond(w, 201, respData)
|
||||
}
|
||||
|
||||
@@ -216,7 +216,11 @@ func NewServer() http.Handler {
|
||||
DerivedFrom: input.EpisodeID,
|
||||
Freshness: "fresh",
|
||||
Source: srcOrLLMDistill(input.Source), // 2026-09-26：有笔记出处→"wiki:<路径>"；否则沿用 llm_distill（2026-09-07 方案C provenance）
|
||||
}
|
||||
// 2026-10-08 (t_7e3577c5)：血缘字段——写入方经 metadata 携带；缺省兜底
|
||||
// embedding_model→bge-m3（实际嵌入模型），script_version→unknown。
|
||||
EmbeddingModel: orDefault(input.EmbeddingModel, "bge-m3"),
|
||||
ScriptVersion: orDefault(input.ScriptVersion, "unknown"),
|
||||
}
|
||||
log.Printf("[srcprobe] 蒸馏落库 episode=%s input.Source=%q → mem.Source=%q", input.EpisodeID, input.Source, mem.Source)
|
||||
if err := ldb.InsertMemory(mem); err != nil {
|
||||
log.Printf("[distill] commit fact failed: %v", err)
|
||||
@@ -1116,7 +1120,7 @@ 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, req.Source, "", "")
|
||||
respondJSON(w, 200, map[string]string{"status": "queued", "episode_id": epID})
|
||||
})
|
||||
// 蒸馏队列状态端点（G8-G9 配套）
|
||||
@@ -1902,3 +1906,13 @@ func srcOrLLMDistill(src string) string {
|
||||
}
|
||||
return "llm_distill"
|
||||
}
|
||||
|
||||
// orDefault 血缘字段缺省兜底：空值 → 默认值。
|
||||
// 2026-10-08 (t_7e3577c5)：embedding_model→bge-m3，script_version→unknown，
|
||||
// 使未发 metadata 的写入路径（worklog/cbm/会话蒸馏）也落非空血缘。
|
||||
func orDefault(v, def string) string {
|
||||
if v == "" {
|
||||
return def
|
||||
}
|
||||
return v
|
||||
}
|
||||
|
||||
@@ -44,6 +44,9 @@ type DistillInput struct {
|
||||
AgentID string
|
||||
// 2026-09-26：来源（笔记路径等）—— 透传到蒸馏产物 MemoryRecord.Source，recall 时带回
|
||||
Source string
|
||||
// 2026-10-08 (t_7e3577c5)：血缘字段——透传到蒸馏产物 MemoryRecord 对应列
|
||||
EmbeddingModel string
|
||||
ScriptVersion string
|
||||
}
|
||||
|
||||
// DistillResult 蒸馏输出
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
package governance
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// TestNormalizeAlias_EnglishOldNames 2026-10-03 新增
|
||||
// 背景：q003「织忆图谱201个旧实体仍叫旧名」实查发现 n_xiaowei count=7 ——
|
||||
// 中文旧名已清净（小唯=0）但英文旧名从未归一（map 只有中文 key 且大小写敏感）。
|
||||
func TestNormalizeAlias_EnglishOldNames(t *testing.T) {
|
||||
cases := []struct {
|
||||
in string
|
||||
want string
|
||||
}{
|
||||
// 中文旧名（原有能力，防回归）
|
||||
{"小唯", "小怡"},
|
||||
{"小唯/07-Wiki/织忆", "小怡/07-Wiki/织忆"},
|
||||
// 英文旧名 · 大小写各变体（本次修复重点）
|
||||
{"xiaowei", "小怡"},
|
||||
{"Xiaowei", "小怡"},
|
||||
{"XIAOWEI", "小怡"},
|
||||
{"XiaoWei", "小怡"},
|
||||
// 复合标识符（含旧名的整串）
|
||||
{"soulful-xiaowei-main", "soulful-小怡-main"},
|
||||
{"my-xiaowei-main", "my-小怡-main"},
|
||||
{"Soulful-Xiaowei-Main", "soulful-小怡-main"}, // 归一后统一规范形式（小写）
|
||||
// 不应被误伤
|
||||
{"小怡", "小怡"},
|
||||
{"xiaoyi", "xiaoyi"},
|
||||
{"Docker", "Docker"},
|
||||
{"", ""},
|
||||
}
|
||||
for _, c := range cases {
|
||||
got := normalizeAlias(c.in)
|
||||
if got != c.want {
|
||||
t.Errorf("normalizeAlias(%q) = %q, want %q", c.in, got, c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestNormalizeAlias_NoInfiniteLoop 子串替换必须终止（实现用了 for+Index 循环）
|
||||
func TestNormalizeAlias_NoInfiniteLoop(t *testing.T) {
|
||||
for _, in := range []string{
|
||||
"xiaowei-xiaowei-xiaowei",
|
||||
"XIAOWEI_Xiaowei_xiaowei",
|
||||
"小唯小唯小唯",
|
||||
"soulful-xiaowei-main-soulful-xiaowei-main",
|
||||
} {
|
||||
_ = normalizeAlias(in)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUpdateFromDistill_EnglishOldNameNotRevived 端到端：
|
||||
// 蒸馏产物带英文旧名 → 写入图谱后不得再出现 n_xiaowei，应归一到「小怡」
|
||||
func TestUpdateFromDistill_EnglishOldNameNotRevived(t *testing.T) {
|
||||
g := NewInMemoryGraph()
|
||||
up := NewAutoGraphUpdater(g)
|
||||
up.UpdateFromDistill(&DistillInput{
|
||||
EpisodeID: "ep_test_en",
|
||||
Entities: []string{"Xiaowei", "xiaowei", "soulful-xiaowei-main"},
|
||||
Facts: []string{"Xiaowei 负责运维"},
|
||||
Namespace: "ns_en",
|
||||
})
|
||||
nodes := g.ListNodes("ns_en")
|
||||
if len(nodes) == 0 {
|
||||
t.Fatal("应建出节点，实际 0 个")
|
||||
}
|
||||
for _, n := range nodes {
|
||||
name, _ := n["name"].(string)
|
||||
id, _ := n["id"].(string)
|
||||
if id == "n_xiaowei" || id == "n_Xiaowei" || name == "xiaowei" || name == "Xiaowei" {
|
||||
t.Errorf("英文旧名不应再入图：id=%s name=%s", id, name)
|
||||
}
|
||||
}
|
||||
found := false
|
||||
for _, n := range nodes {
|
||||
if n["name"] == "小怡" {
|
||||
found = true
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Errorf("英文旧名应归一到「小怡」，实际: %v", nodes)
|
||||
}
|
||||
_ = strings.Contains // 保持 strings 引用
|
||||
}
|
||||
@@ -56,6 +56,10 @@ type MemoryRecord struct {
|
||||
VolatileFlag bool `json:"volatile_flag"` // 频繁变更标记（≥3次修正自动设置）
|
||||
RecallCount int `json:"recall_count"`
|
||||
Freshness string `json:"freshness"` // fresh, stale, verified
|
||||
// 2026-10-08 (t_7e3577c5)：血缘字段。embedding_model=实际嵌入模型（bge-m3）；
|
||||
// script_version=写入脚本版本标识。写入方经 metadata 携带，server 蒸馏落库。
|
||||
EmbeddingModel string `json:"embedding_model"`
|
||||
ScriptVersion string `json:"script_version"`
|
||||
LastRecalledAt time.Time `json:"last_recalled_at"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
|
||||
@@ -157,6 +157,7 @@ func (rc *RustLanceDBClient) InsertMemory(m models.MemoryRecord) error {
|
||||
"freshness": m.Freshness, "version": 1, "version_history": "[]",
|
||||
"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) 血缘
|
||||
"created_at": m.CreatedAt.Format(time.RFC3339),
|
||||
"updated_at": m.UpdatedAt.Format(time.RFC3339),
|
||||
}
|
||||
|
||||
@@ -28,6 +28,8 @@ pub struct MemoryRecord {
|
||||
pub recall_count: i64, pub useful_count: i64, pub not_useful_count: i64,
|
||||
pub freshness: String, pub version: i64,
|
||||
pub version_history: String, pub source: String,
|
||||
// 2026-10-08 (t_7e3577c5)：血缘字段
|
||||
pub embedding_model: String, pub script_version: 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,
|
||||
@@ -56,6 +58,8 @@ fn arrow_schema() -> Schema {
|
||||
Field::new("version", DataType::Int64, false),
|
||||
Field::new("version_history", DataType::Utf8, true),
|
||||
Field::new("source", DataType::Utf8, true),
|
||||
Field::new("embedding_model", DataType::Utf8, true),
|
||||
Field::new("script_version", DataType::Utf8, true),
|
||||
Field::new("volatile_flag", DataType::Boolean, false),
|
||||
Field::new("is_deleted", DataType::Boolean, false),
|
||||
Field::new("depends_on", DataType::Utf8, true),
|
||||
@@ -117,6 +121,8 @@ impl LanceDBOps {
|
||||
version: col_i64(&batch, i, "version"),
|
||||
version_history: String::new(),
|
||||
source: col_str(&batch, i, "source"),
|
||||
embedding_model: col_str(&batch, i, "embedding_model"),
|
||||
script_version: col_str(&batch, i, "script_version"),
|
||||
volatile_flag: col_bool(&batch, i, "volatile_flag"),
|
||||
is_deleted: col_bool(&batch, i, "is_deleted"),
|
||||
depends_on: String::new(),
|
||||
@@ -164,6 +170,8 @@ impl LanceDBOps {
|
||||
version: col_i64(&batch, i, "version"),
|
||||
version_history: String::new(),
|
||||
source: col_str(&batch, i, "source"),
|
||||
embedding_model: col_str(&batch, i, "embedding_model"),
|
||||
script_version: col_str(&batch, i, "script_version"),
|
||||
volatile_flag: col_bool(&batch, i, "volatile_flag"),
|
||||
is_deleted: col_bool(&batch, i, "is_deleted"),
|
||||
depends_on: String::new(),
|
||||
@@ -193,6 +201,7 @@ impl LanceDBOps {
|
||||
let mut recall_counts = Vec::new(); let mut useful_counts = Vec::new(); let mut not_useful_counts = Vec::new();
|
||||
let mut freshnesses = Vec::new(); let mut versions = Vec::new();
|
||||
let mut version_histories = Vec::new(); let mut sources = Vec::new();
|
||||
let mut embedding_models = Vec::new(); let mut script_versions = Vec::new();
|
||||
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();
|
||||
@@ -217,6 +226,8 @@ impl LanceDBOps {
|
||||
versions.push(r["version"].as_i64().unwrap_or(1));
|
||||
version_histories.push(r["version_history"].as_str().unwrap_or("[]"));
|
||||
sources.push(r["source"].as_str().unwrap_or(""));
|
||||
embedding_models.push(r["embedding_model"].as_str().unwrap_or(""));
|
||||
script_versions.push(r["script_version"].as_str().unwrap_or(""));
|
||||
volatile_flags.push(r["volatile_flag"].as_bool().unwrap_or(false));
|
||||
is_deleteds.push(r["is_deleted"].as_bool().unwrap_or(false));
|
||||
depends_ons.push(r["depends_on"].as_str().unwrap_or("[]"));
|
||||
@@ -253,6 +264,8 @@ impl LanceDBOps {
|
||||
Arc::new(Int64Array::from(versions)),
|
||||
Arc::new(StringArray::from(version_histories)),
|
||||
Arc::new(StringArray::from(sources)),
|
||||
Arc::new(StringArray::from(embedding_models)),
|
||||
Arc::new(StringArray::from(script_versions)),
|
||||
Arc::new(BooleanArray::from(volatile_flags)),
|
||||
Arc::new(BooleanArray::from(is_deleteds)),
|
||||
Arc::new(StringArray::from(depends_ons)),
|
||||
@@ -566,6 +579,8 @@ impl LanceDBOps {
|
||||
version: col_i64(&batch, i, "version"),
|
||||
version_history: String::new(),
|
||||
source: col_str(&batch, i, "source"),
|
||||
embedding_model: col_str(&batch, i, "embedding_model"),
|
||||
script_version: col_str(&batch, i, "script_version"),
|
||||
volatile_flag: col_bool(&batch, i, "volatile_flag"),
|
||||
is_deleted: col_bool(&batch, i, "is_deleted"),
|
||||
depends_on: String::new(),
|
||||
@@ -623,6 +638,8 @@ impl LanceDBOps {
|
||||
version: col_i64(&batch, i, "version"),
|
||||
version_history: String::new(),
|
||||
source: col_str(&batch, i, "source"),
|
||||
embedding_model: col_str(&batch, i, "embedding_model"),
|
||||
script_version: col_str(&batch, i, "script_version"),
|
||||
volatile_flag: col_bool(&batch, i, "volatile_flag"),
|
||||
is_deleted: col_bool(&batch, i, "is_deleted"),
|
||||
depends_on: String::new(),
|
||||
|
||||
Reference in New Issue
Block a user