feat(distill): 限制 prompt ≤8k token 保护本地 2B 单槽; storage: 添加 SoftDeleteBatch 及 Rust 端批量软删除优化 (t_3fb41049)
This commit is contained in:
小唯
parent
995e93598a
commit
d901f0e485
|
|
@ -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格式返回。
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
}
|
||||
|
|
@ -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)。
|
||||
|
|
|
|||
|
|
@ -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::Value> = 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<u64, Box<dyn std::error::Error>> {
|
||||
#[derive(serde::Deserialize)]
|
||||
struct Item {
|
||||
id: String,
|
||||
}
|
||||
let items: Vec<Item> = 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<String, Vec<String>> = 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::<Vec<_>>().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<Vec<MemoryRecord>, Box<dyn std::error::Error>> {
|
||||
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<Vec<MemoryRecord>, Box<dyn std::error::Error>> {
|
||||
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<LanceDBStats, Box<dyn std::error::Error>> {
|
||||
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::<StringArray>())
|
||||
.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::<Float64Array>())
|
||||
.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::<Int64Array>())
|
||||
.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::<BooleanArray>())
|
||||
.map(|a| a.value(r))
|
||||
.unwrap_or(false)
|
||||
}
|
||||
|
||||
/// 读取 vector 列 (兼容 FixedSizeListArray 和 Float32Array)
|
||||
fn col_vector(b: &RecordBatch, r: usize, c: &str) -> Vec<f32> {
|
||||
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::<FixedSizeListArray>() {
|
||||
let item_arr = list_arr.value(r);
|
||||
if let Some(float_arr) = item_arr.as_any().downcast_ref::<Float32Array>() {
|
||||
return float_arr.values().to_vec();
|
||||
}
|
||||
}
|
||||
// 回退: Float32Array (展平格式)
|
||||
if let Some(float_arr) = col.as_any().downcast_ref::<Float32Array>() {
|
||||
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]
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue