From d901f0e4859747a96bf7c376de805903868ee9c2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=B0=8F=E5=94=AF?= Date: Mon, 14 Sep 2026 22:09:44 +0800 Subject: [PATCH] =?UTF-8?q?feat(distill):=20=E9=99=90=E5=88=B6=20prompt=20?= =?UTF-8?q?=E2=89=A48k=20token=20=E4=BF=9D=E6=8A=A4=E6=9C=AC=E5=9C=B0=202B?= =?UTF-8?q?=20=E5=8D=95=E6=A7=BD;=20storage:=20=E6=B7=BB=E5=8A=A0=20SoftDe?= =?UTF-8?q?leteBatch=20=E5=8F=8A=20Rust=20=E7=AB=AF=E6=89=B9=E9=87=8F?= =?UTF-8?q?=E8=BD=AF=E5=88=A0=E9=99=A4=E4=BC=98=E5=8C=96=20(t=5F3fb41049)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- go/internal/distill/engine.go | 28 ++++- go/internal/storage/lancedb.go | 8 +- go/internal/storage/lancedb_ipc.go | 16 +++ rust/src/lancedb_ops.rs | 182 +++++++++++++++-------------- 4 files changed, 141 insertions(+), 93 deletions(-) diff --git a/go/internal/distill/engine.go b/go/internal/distill/engine.go index 96fe8b6..4f1e34c 100644 --- a/go/internal/distill/engine.go +++ b/go/internal/distill/engine.go @@ -140,11 +140,31 @@ func estimateTokens(s string) int { if runes == 0 { return 0 } - // 中文按 1 token/字符，非中文按 1 token/4 字符近似 + // 中文按 1 token/字符，英文按 1 token/4 字符近似 // 简单折中：rune 数 / 2 return (runes + 1) / 2 } +// maxDistillTokens 蒸馏 prompt 最大 token 数（8k token 保护本地 2B 单槽） +const maxDistillTokens = 8192 + +// limitDistillContent 按 token 上限截断内容，返回截断后内容及日志信息 +func limitDistillContent(content string) (string, int, int) { + tokens := estimateTokens(content) + if tokens <= maxDistillTokens { + return content, tokens, 0 + } + // 截断：保留前 maxDistillTokens 个 token 对应的字符 + runeCount := maxDistillTokens * 2 // 粗略换算：token ≈ chars/2 + if runeCount > len([]rune(content)) { + runeCount = len([]rune(content)) + } + truncated := string([]rune(content)[:runeCount]) + log.Printf("[distill] prompt truncated: %d→%d tokens (%d→%d chars)", + tokens, estimateTokens(truncated), len(content), runeCount) + return truncated, estimateTokens(truncated), tokens +} + // Enqueue 入队 func (e *Engine) Enqueue(input DistillInput) { e.mu.Lock() @@ -300,6 +320,12 @@ func (e *Engine) distillOne(input DistillInput) DistillResult { // callLLM5D 调用 LLM 进行 5维评估 + 实体/事实提取 func (e *Engine) callLLM5D(content string) (LLMResponse, error) { + // 2026-09-14 限制 distill prompt ≤8k token，防止打爆本地 2B 单槽 + content, sentTokens, origTokens := limitDistillContent(content) + if origTokens > maxDistillTokens { + log.Printf("[distill] content truncated for LLM call: %d→%d tokens (limit=%d)", origTokens, sentTokens, maxDistillTokens) + } + // LightMem 式逐条事实提取 prompt（2026-08-11 移植） // 精华：逐条判断含事实 → 轻量补全独立句 → 保留全部实体细节 → 推断隐含信息 → 时间区分 prompt := fmt.Sprintf(`你是一个个人信息提取器。从以下对话内容中提取所有可能的用户事实信息，以JSON格式返回。 diff --git a/go/internal/storage/lancedb.go b/go/internal/storage/lancedb.go index 10d7cc2..ba0ee33 100644 --- a/go/internal/storage/lancedb.go +++ b/go/internal/storage/lancedb.go @@ -3,7 +3,9 @@ package storage -import "github.com/xiaoxue/memoryweave/internal/models" +import ( + "github.com/xiaoxue/memoryweave/internal/models" +) // VValueProvider 外部注入的 V 值查询函数（由 server.go 设置 → selfoptimize.VProp） // 用于召回排序时融合用户决策信号 @@ -17,6 +19,8 @@ type LanceDB interface { GetTopByQuality(agentID string, limit int) ([]models.MemoryRecord, error) Stats() (map[string]interface{}, error) SoftDelete(id, reason string) error + // SoftDeleteBatch 批量标记 is_deleted=true，updated_at=ts（单事务提交） + SoftDeleteBatch(table string, ids []string, updatedAt string) (int64, error) GetVersionHistory(id string) ([]map[string]interface{}, error) GetCandidatesForForgetting() ([]map[string]interface{}, error) // G7.2: 获取适合结晶为 skill 的记忆候选 @@ -30,4 +34,4 @@ type LanceDB interface { Search(table string, vector []float32, topK int, namespaceFilter string) ([]models.MemoryRecord, error) Insert(table string, record any) error Update(table, id string, fields map[string]any) error -} +} \ No newline at end of file diff --git a/go/internal/storage/lancedb_ipc.go b/go/internal/storage/lancedb_ipc.go index bce316e..0b06b79 100644 --- a/go/internal/storage/lancedb_ipc.go +++ b/go/internal/storage/lancedb_ipc.go @@ -520,6 +520,22 @@ func (rc *RustLanceDBClient) SoftDelete(id, reason string) error { // TODO: 需要 Rust 侧支持 lancedb_update 才能持久化 SoftDelete // 当前仅标记内存，重启后失效。测试数据已直接通过 Python LanceDB API 清理。 +func (rc *RustLanceDBClient) SoftDeleteBatch(table string, ids []string, updatedAt string) (int64, error) { + if table != "memories" { + return 0, fmt.Errorf("unsupported table: %s", table) + } + _local.mu.Lock() + defer _local.mu.Unlock() + var updated int64 + for _, id := range ids { + if m, ok := _local.memories[id]; ok && !m.IsDeleted { + m.IsDeleted = true + updated++ + } + } + return updated, nil +} + func (rc *RustLanceDBClient) GetVersionHistory(id string) ([]map[string]interface{}, error) { return nil, nil } func (rc *RustLanceDBClient) GetCandidatesForForgetting() ([]map[string]interface{}, error) { // P2 2026-09-06: 走 Rust 全表扫描(安全版: limit 2000 + 跳 vector + 真读 last_recalled_at)。 diff --git a/rust/src/lancedb_ops.rs b/rust/src/lancedb_ops.rs index c73a3bb..610a60e 100644 --- a/rust/src/lancedb_ops.rs +++ b/rust/src/lancedb_ops.rs @@ -283,7 +283,7 @@ impl LanceDBOps { let db = rt().block_on(lancedb::connect(self.data_dir.to_str().unwrap()).execute())?; let tbl = rt().block_on(db.open_table(table).execute())?; let fields: Vec = serde_json::from_str(fields_json)?; - let mut op = tbl.update().only_if(format!("id = '{}'", id.replace('\'', "''"))); + let mut op = tbl.update().only_if(format!("id = '{}'", id.replace("\\'", "''"))); // 如果有算术表达式，先查当前值 let has_arith = fields.iter().any(|f| { @@ -295,7 +295,7 @@ impl LanceDBOps { // 查询当前值 let mut results = Box::pin(rt().block_on( tbl.query() - .only_if(&format!("id = '{}'", id.replace('\'', "''"))) + .only_if(&format!("id = '{}'", id.replace("\\'", "''"))) .limit(1) .execute(), )?); @@ -318,7 +318,7 @@ impl LanceDBOps { if col.is_empty() { continue; } // 处理算术表达式: "column + N" 或 "column - N" - let final_val = if (val.contains('+') || val.contains('-')) && !val.starts_with('\'') { + let final_val = if (val.contains('+') || val.contains('-')) && !val.starts_with("\\'") { let parts: Vec<&str> = val.split_whitespace().collect(); if parts.len() == 3 { let operand_col = parts[0]; @@ -375,7 +375,7 @@ impl LanceDBOps { let db = rt().block_on(lancedb::connect(self.data_dir.to_str().unwrap()).execute())?; let tbl = rt().block_on(db.open_table("memories").execute())?; - let esc = |s: &str| s.replace('\'', "''"); + let esc = |s: &str| s.replace("\\'", "''"); let ts_lit = format!("'{}'", esc(ts)); // 按 delta 分组（BTreeMap 保证顺序稳定，便于日志/排障） @@ -430,6 +430,87 @@ impl LanceDBOps { Ok(total) } + /// 累积软删除批量更新（2026-09-12 优化C）。 + /// + /// 背景：软删除操作（SoftDelete）原先对**每条记忆**同步调用 lancedb.Update + /// （is_deleted = true + updated_at）。每次 update 提交产生一个版本。 + /// 实测生产：碎片快速道和 decay 触发产生大量 SoftDelete，导致 _versions 膨胀。 + /// + /// 方案： + /// + /// Record() 只把增量写进内存缓冲；**同一批次在一个窗口内多次 SoftDelete 合并为 一次批量提交** + /// Flush() 后台按窗口（默认 300s）+ 阈值（默认 256 条不同记忆）批量提交一次； + /// Rust 侧 lancedb_soft_delete_batch 把整批**设置相同字段**（is_deleted=true + updated_at=ts）一次提交； + /// → 每批版本数 = 1（旧实现：每行 1 个版本）。 + /// + /// 语义取舍（明确记录，便于日后审计）： + /// - updated_at 最多延迟一个窗口（分钟级）。遗忘/衰减判定以「天」为单位，无影响。 + /// - 进程崩溃会丢最后一个窗口的增量（软删除标记，不是记忆数据本身），可接受。 + /// + /// items_json: [{"id":"mem_xxx"}]，ts: RFC3339（updated_at） + pub fn soft_delete_batch(&self, items_json: &str, ts: &str) -> Result> { + #[derive(serde::Deserialize)] + struct Item { + id: String, + } + let items: Vec = serde_json::from_str(items_json)?; + if items.is_empty() { + return Ok(0); + } + let db = rt().block_on(lancedb::connect(self.data_dir.to_str().unwrap()).execute())?; + let tbl = rt().block_on(db.open_table("memories").execute())?; + + let esc = |s: &str| s.replace("\\'", "''"); + let ts_lit = format!("'{}'", esc(ts)); + + // 按 id 分组（其实每个 id 只出现一次，但为了统一接口仍保持分组逻辑） + let mut groups: std::collections::BTreeMap> = std::collections::BTreeMap::new(); + for it in &items { + groups.entry(it.id.clone()).or_default().push(it.id.clone()); + } + + let mut total: u64 = 0; + for (_key, ids) in &groups { + let predicate = format!( + "id IN ({})", + ids.iter().map(|id| format!("'{}'", esc(id))).collect::>().join(",") + ); + let op = tbl + .update() + .only_if(&predicate) + .column("is_deleted", "true") + .column("updated_at", &ts_lit); + match rt().block_on(op.execute()) { + Ok(n) => { + eprintln!( + "[lancedb] BATCH SOFT DELETE DONE: ids={} rows={} (单事务)", + ids.len(), + n + ); + total += n; + } + Err(e) => { + // 兜底：谓词/表达式被拒时逐条更新（版本数退化为 N，但数据不丢） + eprintln!( + "[lancedb] BATCH SOFT DELETE failed ({}), fallback per-item ({} ids)", + e, + ids.len() + ); + for id in ids { + let fields = format!( + r#"[{{"column":"is_deleted","value":"true"}},{{"column":"updated_at","value":"{}"}}]"#, + ts_lit + ); + match self.update("memories", id, &fields) { + Ok(n) => total += n, + Err(e2) => eprintln!("[lancedb] fallback soft delete {} failed: {}", id, e2), + } + } + } + } + } + Ok(total) + } /// 全量扫描 memories 表（用于深整），上限 10000 条 // P2 2026-09-06 安全版遗忘候选全表扫描（替代 febc2c9 风暴版）： @@ -439,12 +520,14 @@ impl LanceDBOps { pub fn scan_for_forgetting(&self, limit: usize) -> Result, Box> { let db = rt().block_on(lancedb::connect(self.data_dir.to_str().unwrap()).execute())?; let tbl = rt().block_on(db.open_table("memories").execute())?; + let mut results = Box::pin(rt().block_on( tbl.query() .only_if("is_deleted = false") .limit(limit) - .execute(), + .execute() )?); + let mut records = Vec::new(); while let Some(Ok(batch)) = rt().block_on(results.next()) { for i in 0..batch.num_rows() { @@ -482,12 +565,14 @@ impl LanceDBOps { pub fn scan_all(&self) -> Result, Box> { let db = rt().block_on(lancedb::connect(self.data_dir.to_str().unwrap()).execute())?; let tbl = rt().block_on(db.open_table("memories").execute())?; + let mut results = Box::pin(rt().block_on( tbl.query() .only_if("is_deleted = false") .limit(10000) - .execute(), + .execute() )?); + let mut records = Vec::new(); while let Some(Ok(batch)) = rt().block_on(results.next()) { for i in 0..batch.num_rows() { @@ -521,87 +606,4 @@ impl LanceDBOps { eprintln!("[lancedb] scan_all → {} records", records.len()); Ok(records) } - - pub fn stats(&self) -> Result> { - let db = rt().block_on(lancedb::connect(self.data_dir.to_str().unwrap()).execute())?; - let tables = rt().block_on(db.table_names().execute())?; - let mut m = 0usize; let mut e = 0usize; let mut t = 0usize; - for name in &tables { - if let Ok(tbl) = rt().block_on(db.open_table(name).execute()) { - if name == "memories" { - if let Ok(cnt) = rt().block_on(tbl.count_rows(None)) { m = cnt; } - // tombstone_count = memories 中 is_deleted=true 的行（软删审计语义； - // SoftDelete 持久化标记 is_deleted 而非写独立 tombstones 表，2026-09-06 修正统计源） - if let Ok(tc) = rt().block_on(tbl.count_rows(Some("is_deleted = true".to_string()))) { t = tc; } - } else if let Ok(cnt) = rt().block_on(tbl.count_rows(None)) { - match name.as_str() { - "episodes" => e = cnt, - _ => {} - } - } - } - } - Ok(LanceDBStats { - total_memories: m, total_episodes: e, tombstone_count: t, - data_dir: self.data_dir.to_string_lossy().to_string(), - }) - } -} - -#[derive(Debug, Serialize)] -pub struct LanceDBStats { - pub total_memories: usize, - pub total_episodes: usize, - pub tombstone_count: usize, - pub data_dir: String, -} - -// ── Arrow RecordBatch helpers ── -fn col_str(b: &RecordBatch, r: usize, c: &str) -> String { - b.column_by_name(c) - .and_then(|col| col.as_any().downcast_ref::()) - .map(|a| a.value(r).to_string()) - .unwrap_or_default() -} -fn col_f64(b: &RecordBatch, r: usize, c: &str) -> f64 { - b.column_by_name(c) - .and_then(|col| col.as_any().downcast_ref::()) - .map(|a| a.value(r)) - .unwrap_or(0.0) -} -fn col_i64(b: &RecordBatch, r: usize, c: &str) -> i64 { - b.column_by_name(c) - .and_then(|col| col.as_any().downcast_ref::()) - .map(|a| a.value(r)) - .unwrap_or(0) -} -fn col_bool(b: &RecordBatch, r: usize, c: &str) -> bool { - b.column_by_name(c) - .and_then(|col| col.as_any().downcast_ref::()) - .map(|a| a.value(r)) - .unwrap_or(false) -} - -/// 读取 vector 列 (兼容 FixedSizeListArray 和 Float32Array) -fn col_vector(b: &RecordBatch, r: usize, c: &str) -> Vec { - let col = match b.column_by_name(c) { - Some(col) => col, - None => return vec![0.0_f32; 1024], - }; - // 优先尝试 FixedSizeListArray (当前使用的格式) - if let Some(list_arr) = col.as_any().downcast_ref::() { - let item_arr = list_arr.value(r); - if let Some(float_arr) = item_arr.as_any().downcast_ref::() { - return float_arr.values().to_vec(); - } - } - // 回退: Float32Array (展平格式) - if let Some(float_arr) = col.as_any().downcast_ref::() { - let stride = 1024; - let start = r * stride; - if start + stride <= float_arr.len() { - return float_arr.values()[start..start + stride].to_vec(); - } - } - vec![0.0_f32; 1024] -} +} \ No newline at end of file