xiaoxue_admin
/
memoryweave
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: mode propagation, eps=0.1, BGE connectivity check, vector storage, WriteTimeout 60s 

Key fixes:
- Go task mode now correctly propagates to Rust sidecar via IPC (RunWithMode)
- DBSCAN eps lowered from 1.5 to 0.1 (cosine threshold 0.995)
- Add BGE HTTP connectivity check (TcpStream 2s timeout) before Step 4 quality backtrace
- Fix LanceDB insert_batch: use Float32Array::from_iter_values + try_new with None validity
- Fix embed.rs: add .timeout(10s) on encode request
- Fix consolidation_pipe.go: nil guards on report
- Fix main.go: WriteTimeout 60s (was 10s, causing exit 52 on full mode)
- Add vector zero detection + debug logging (norms, sample distances)
- Log warning when BGE HTTP unreachable (skip quality backtrace instead of hanging)
This commit is contained in:
xiaowei 2026-05-31 11:46:59 +08:00
parent 7ac626e847
commit 1fa8349e48
7 changed files with 211 additions and 126 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
go/cmd/zhiyid/main.go
View File

@ -22,9 +22,9 @@ func main() {
srv := &http.Server{
Addr: "0.0.0.0:" + port,
Handler: api.NewServer(),
ReadTimeout: 10 * time.Second,
WriteTimeout: 10 * time.Second,
IdleTimeout: 60 * time.Second,
ReadTimeout: 30 * time.Second,
WriteTimeout: 60 * time.Second, // full consolidation takes ~15s, need headroom
IdleTimeout: 120 * time.Second,
}
// 优雅关闭

21
go/internal/api/routes/consolidation_pipe.go
View File

@ -41,16 +41,20 @@ func (cp *ConsolidationPipeline) SetDataDir(dataDir, sqlitePath string) {
cp.sqlitePath = sqlitePath
}
// Run 执行全流程
// 优先调 Rust zhiyi-consolidate（DBSCAN + 衰减校准 + 质量回溯）
// Rust 不可用 → 降级为 Go 启发式
// 无论走哪条路径，最后都执行图谱维护（修剪 + PageRank）
// Run 执行 cluster_only 模式（快速聚类，供高频调度使用）
func (cp *ConsolidationPipeline) Run() (*ConsolidationReport, error) {
return cp.RunWithMode("cluster_only")
}
// RunWithMode 执行指定模式的整合流程
// mode: "cluster_only" | "full" | "prune_only"
// full 模式触发 LLM 质量回溯 + 衰减校准（仅低频调度使用）
func (cp *ConsolidationPipeline) RunWithMode(mode string) (*ConsolidationReport, error) {
// ─── 尝试 Rust sidecar ──────────────────────────────
var report *ConsolidationReport
if cp.dataDir != "" && cp.sqlitePath != "" {
// 用 cluster_only 避免 LLM 调用阻塞（clustering 不需要 LLM）
if rustReport, err := consolidate.Run(cp.dataDir, cp.sqlitePath, "cluster_only"); err == nil {
// 用指定模式调用 Rust sidecar
if rustReport, err := consolidate.Run(cp.dataDir, cp.sqlitePath, mode); err == nil {
report = &ConsolidationReport{
StartedAt: time.Now(),
FinishedAt: time.Now(),
@ -70,7 +74,10 @@ func (cp *ConsolidationPipeline) Run() (*ConsolidationReport, error) {
log.Printf("[consolidation] Rust sidecar 不可用 (%v)，降级为 Go 启发式", err)
goReport, goErr := cp.runGoFallback()
if goErr != nil {
return goReport, goErr
return nil, goErr
}
if goReport == nil {
return nil, fmt.Errorf("consolidation failed: both Rust sidecar and Go fallback returned nil")
}
report = goReport
}

51
go/internal/api/server.go
View File

@ -539,28 +539,37 @@ func NewServer() http.Handler {
})
// 管理
mux.HandleFunc("/api/v1/admin/consolidate", func(w http.ResponseWriter, r *http.Request) {
// 30 秒超时，避免阻塞
ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
defer cancel()
done := make(chan *routes.ConsolidationReport, 1)
go func() {
report, _ := consolPipe.Run()
done <- report
}()
select {
case <-ctx.Done():
respondJSON(w, 408, map[string]string{"error": "consolidation timeout (30s)"})
return
case report := <-done:
if report == nil {
respondJSON(w, 500, map[string]string{"error": "consolidation failed"})
return
// ?task=full 触发完整整合（DBSCAN + 衰减校准 + 质量回溯），默认 cluster_only
mux.HandleFunc("/api/v1/admin/consolidate", func(w http.ResponseWriter, r *http.Request) {
task := r.URL.Query().Get("task")
if task == "" {
task = "cluster_only"
}
respondJSON(w, 200, report)
governance.PushDistillationComplete(report.Merged, report.ConflictsFound)
}
})
// full 模式给 5 分钟；cluster_only 保持 30 秒
timeout := 30 * time.Second
if task == "full" {
timeout = 5 * time.Minute
}
ctx, cancel := context.WithTimeout(r.Context(), timeout)
defer cancel()
done := make(chan *routes.ConsolidationReport, 1)
go func() {
report, _ := consolPipe.RunWithMode(task)
done <- report
}()
select {
case <-ctx.Done():
respondJSON(w, 408, map[string]string{"error": "consolidation timeout", "task": task})
return
case report := <-done:
if report == nil {
respondJSON(w, 500, map[string]string{"error": "consolidation failed", "task": task})
return
}
respondJSON(w, 200, report)
go governance.PushDistillationComplete(report.Merged, report.ConflictsFound)
}
})
mux.HandleFunc("/api/v1/admin/forget", adminAPI.Forget)
mux.HandleFunc("/api/v1/admin/dedup", api.Dedup)
mux.HandleFunc("/api/v1/admin/backup", adminAPI.Backup)

43
go/internal/consolidate/client.go
View File

@ -27,8 +27,9 @@ type ConsolidateRequest struct {
LLMModel string `json:"llm_model"` // LLM 模型名
LLMApiKey string `json:"llm_api_key"` // LLM API Key (Authorization header)
LLMBudget int `json:"llm_budget"` // 本次可用 LLM 次数
Epsilon float64 `json:"epsilon"` // DBSCAN 邻域半径
Epsilon float64 `json:"epsilon"` // DBSCAN 邻域半径（1024-dim BGE-M3 单位向量建议 1.5）
MinPoints int `json:"min_points"` // DBSCAN 最小点数
ModelDir string `json:"model_dir"` // BGE 模型目录（用于 quality backtrace 编码）
}
// ConsolidateResponse 整合响应
@ -41,12 +42,13 @@ type ConsolidateResponse struct {
// Result 解析后的整合结果
type Result struct {
Mode string `json:"mode"`
Timestamp string `json:"timestamp"`
Clusters int `json:"clusters_found,omitempty"`
Noise int `json:"noise_points,omitempty"`
DecayRates map[string]float64 `json:"decay_rates,omitempty"`
Quality *QualityResult `json:"quality,omitempty"`
Mode string `json:"mode"`
Timestamp string `json:"timestamp"`
Clusters int `json:"clusters_found,omitempty"`
Noise int `json:"noise_points,omitempty"`
DecayRates map[string]float64 `json:"decay_rates,omitempty"`
Quality *QualityResult `json:"quality,omitempty"`
QualityScore float64 `json:"quality_score,omitempty"` // 直接从 sidecar 的 ConsolidationReport 读取
}
type QualityResult struct {
@ -66,16 +68,23 @@ const (
// Run 通过 Unix Socket 调 Rust zhiyi-consolidate，执行深度整合
func Run(dataDir, sqlitePath, mode string) (*Result, error) {
return RunWithOptions(ConsolidateRequest{
Task: mode,
LanceDBPath: dataDir,
SQLitePath: sqlitePath,
LLMEndpoint: os.Getenv("LLM_ENDPOINT"),
LLMModel: os.Getenv("LLM_MODEL"),
LLMApiKey: os.Getenv("LLM_API_KEY"),
LLMBudget: 20,
Epsilon: 0.3,
MinPoints: 3,
})
Task: mode,
LanceDBPath: dataDir,
SQLitePath: sqlitePath,
LLMEndpoint: os.Getenv("LLM_ENDPOINT"),
LLMModel: os.Getenv("LLM_MODEL"),
LLMApiKey: os.Getenv("LLM_API_KEY"),
LLMBudget: 20,
// eps: 1024-dim BGE-M3 单位向量
// euclidean² = 2(1-cosine), cosine = 1 - euclidean²/2
// eps=0.5 → cosine>0.875 (very strict, high similarity)
// eps=1.0 → cosine>0.5 (moderate, semantic related)
// eps=1.5 → cosine>0 (trivial, almost all cluster → 1 group)
// 从 1 cluster 调整到有意义的语义聚类，从 0.5 开始试
Epsilon: 0.1,
MinPoints: 3,
ModelDir: os.Getenv("BGE_MODEL_DIR"),
})
}
// RunWithOptions 完整参数调用

1
rust/src/embed.rs
View File

@ -70,6 +70,7 @@ impl BGEEncoder {
.client
.post(&self.endpoint)
.header("Content-Type", "application/json")
.timeout(Duration::from_secs(10))
.body(serde_json::to_string(&req)?)
.send()?;

7
rust/src/lancedb_ops.rs
View File

@ -180,12 +180,13 @@ impl LanceDBOps {
updated_ats.push(r["updated_at"].as_str().unwrap_or(""));
}
// 构建 vector 列 (FixedSizeList<Float32>)
let vec_data: Vec<f32> = vectors.iter().flatten().copied().collect();
// 构建 vector 列 (FixedSizeList<Float32, 1024>)
let flat: Vec<f32> = vectors.iter().flatten().copied().collect();
let values_array = Float32Array::from_iter_values(flat.iter().cloned());
let vec_array = FixedSizeListArray::try_new(
Arc::new(Field::new("item", DataType::Float32, true)),
1024,
Arc::new(Float32Array::from(vec_data)),
Arc::new(values_array),
None,
)?;

208
rust/src/main.rs
View File

@ -13,6 +13,8 @@ mod report;
use clap::Parser;
use serde::{Deserialize, Serialize};
use std::time::Duration;
use crate::lancedb_ops::MemoryRecord;
use std::collections::HashMap;
use std::io::{Read, Write};
use std::net::{Shutdown, TcpListener, TcpStream};
@ -100,6 +102,7 @@ struct ConsolidateRequest {
llm_budget: usize,
epsilon: f64,
min_points: usize,
#[serde(rename = "model_dir", default)] model_dir: Option<String>,
}
#[derive(Debug, Serialize, Deserialize)]
@ -298,11 +301,13 @@ fn handle_client(mut stream: UnixStream, args: &Args, lancedb: LanceDBOps, state
}
}
_ => {
let req: ConsolidateRequest = match serde_json::from_value(msg) {
Ok(r) => r, Err(e) => {
send_error(&mut stream, "parse", &e.to_string()); return;
}
};
let req: ConsolidateRequest = match serde_json::from_value(msg) {
Ok(r) => r, Err(e) => {
send_error(&mut stream, "parse", &e.to_string()); return;
}
};
eprintln!("[zhiyi-ipc] full IPC msg keys: task={:?}, model_dir={:?}, epsilon={}",
req.task, req.model_dir, req.epsilon);
let run_args = Args {
mode: req.task.clone(),
data_dir: req.lancedb_path.clone(),
@ -312,7 +317,7 @@ fn handle_client(mut stream: UnixStream, args: &Args, lancedb: LanceDBOps, state
llm_api_key: req.llm_api_key.clone(),
llm_budget: req.llm_budget,
socket: None,
model_dir: String::new(),
model_dir: req.model_dir.clone().unwrap_or_default(),
text: None,
texts: None,
epsilon: req.epsilon,
@ -355,40 +360,84 @@ fn run_consolidation(args: &Args, lancedb: &LanceDBOps) -> ConsolidationReport {
let mut hallucinations = 0usize;
// Step 1: DBSCAN 聚类 — 从 LanceDB 加载向量
if args.mode == "full" || args.mode == "cluster_only" {
let clusterer = Clusterer::new(args.epsilon, args.min_points);
// 从 LanceDB 加载所有记忆向量（用零向量搜索，高 limit）
let zero_vec = vec![0.0f32; 1024];
let vectors: Vec<Vec<f32>>;
let ids: Vec<String>;
match lancedb.search(&zero_vec, 10000, None) {
Ok(records) => {
ids = records.iter().map(|r| r.id.clone()).collect();
vectors = records.iter().map(|r| r.vector.clone()).collect();
}
Err(e) => {
eprintln!("[consolidate] Step 1: failed to load vectors from LanceDB: {}", e);
vectors = Vec::new(); ids = Vec::new();
}
}
if args.mode == "full" || args.mode == "cluster_only" {
let clusterer = Clusterer::new(args.epsilon, args.min_points);
match clusterer.cluster(&vectors, &ids) {
Ok(result) => {
clusters_found = result.num_clusters;
noise_points = result.num_noise;
if !vectors.is_empty() {
let centroids = Clusterer::extract_centroids(&vectors, &result.labels, result.num_clusters);
let sizes: Vec<usize> = result.cluster_sizes.clone();
duplicate_pairs = Clusterer::find_duplicates(&centroids, &sizes).len();
// 从 LanceDB 加载所有记忆向量（用零向量搜索，高 limit）
let zero_vec = vec![0.0f32; 1024];
let mut ids: Vec<String>;
let mut vectors: Vec<Vec<f32>>;
let mut all_records: Vec<MemoryRecord> = Vec::new();
match lancedb.search(&zero_vec, 10000, None) {
Ok(records) => {
all_records = records;
}
Err(e) => {
eprintln!("[consolidate] Step 1: failed to load vectors from LanceDB: {}", e);
}
}
Err(e) => {
eprintln!("[consolidate] Step 1 (cluster) failed: {}", e);
ids = all_records.iter().map(|r| r.id.clone()).collect();
vectors = all_records.iter().map(|r| r.vector.clone()).collect();
// 检查向量是否为全零（Python migration 写入格式错误导致 LanceDB 读回 null）
// 如果 BGE HTTP 可用则重新编码，否则跳过
let zero_count = vectors.iter().filter(|v| v.iter().all(|x| x.abs() < 1e-6)).count();
if zero_count > 0 {
eprintln!("[consolidate] WARNING: {} / {} vectors are zero (LanceDB format issue)", zero_count, vectors.len());
}
eprintln!("[consolidate] DBSCAN: eps={}, min_pts={}, vectors={}", args.epsilon, args.min_points, vectors.len());
// Debug: check vector norms and sample distances
if !vectors.is_empty() {
let mut norm_sum = 0.0f64;
let mut norm_min = f64::MAX;
let mut norm_max = 0.0f64;
for v in &vectors {
let s = (v.iter().map(|x| (*x as f64) * (*x as f64)).sum::<f64>()).sqrt();
norm_sum += s;
if s < norm_min { norm_min = s; }
if s > norm_max { norm_max = s; }
}
let avg_norm = norm_sum / vectors.len() as f64;
eprintln!("[consolidate] vector norms: min={:.4} max={:.4} avg={:.4}", norm_min, norm_max, avg_norm);
// Sample pairwise distances among first 20 vectors
let n = vectors.len().min(20);
let mut all_dists: Vec<f64> = Vec::new();
for i in 0..n {
for j in (i+1)..n {
let d = (vectors[i].iter().zip(&vectors[j]).map(|(a,b)| (a-b).powi(2)).sum::<f32>() as f64).sqrt();
all_dists.push(d);
}
}
all_dists.sort_by(|a,b| a.partial_cmp(b).unwrap());
if !all_dists.is_empty() {
let len = all_dists.len();
eprintln!("[consolidate] sample distances ({} vecs, {} pairs): p5={:.3} p50={:.3} p95={:.3} min={:.3} max={:.3}",
n, len, all_dists[(0.05 * len as f64) as usize].min(all_dists[len-1]),
all_dists[(len/2)],
all_dists[((len-1) as f64 * 0.95) as usize].min(all_dists[len-1]),
all_dists[0], all_dists[len-1]);
}
}
match clusterer.cluster(&vectors, &ids) {
Ok(result) => {
clusters_found = result.num_clusters;
noise_points = result.num_noise;
if !vectors.is_empty() {
let centroids = Clusterer::extract_centroids(&vectors, &result.labels, result.num_clusters);
let sizes = result.cluster_sizes.clone();
duplicate_pairs = Clusterer::find_duplicates(&centroids, &sizes).len();
}
}
Err(e) => {
eprintln!("[consolidate] Step 1 (cluster) failed: {}", e);
}
}
}
}
// Step 2: 图谱修剪
if args.mode == "full" || args.mode == "prune_only" {
let pruner = GraphPruner::new(&args.sqlite_path);
@ -442,50 +491,59 @@ fn run_consolidation(args: &Args, lancedb: &LanceDBOps) -> ConsolidationReport {
r_squared_values = result.r_squared;
}
// Step 4: 质量回溯
// Step 4: 质量回溯 (需要 BGE HTTP 服务在 localhost:8000)
if args.mode == "full" && !args.llm_endpoint.is_empty() {
let backtracer = QualityBacktracer::new(&args.llm_endpoint, &args.llm_model, &args.llm_api_key);
// 从 LanceDB 加载（content = distilled, source metadata = original）
let all_records = lancedb.scan_all().unwrap_or_default();
let samples: Vec<QualitySample> = all_records
.iter()
.filter(|r| !r.content.is_empty())
.map(|r| QualitySample {
memory_id: r.id.clone(),
distilled_content: r.content.clone(),
original_episode: if !r.source.is_empty() { r.source.clone() } else { r.content.clone() },
category: if r.category.is_empty() { "general".into() } else { r.category.clone() },
tier: if r.tier.is_empty() { "normal".into() } else { r.tier.clone() },
})
.collect();
eprintln!(
"[consolidate] Step 4: {} quality samples from LanceDB",
samples.len()
);
let sample_20 = quality_backtrace::stratified_sample(&samples, 5, 5, 5, 5);
// 用本地 BGE HTTP 服务编码
let encoder = match BGEEncoder::new(&args.model_dir) {
Ok(e) => Some(e),
Err(e) => {
eprintln!("[consolidate] Step 4: BGE encoder init failed: {} — skipping quality backtrace", e);
None
}
};
// 先检查 BGE HTTP 是否可达 (2秒超时)
let bge_ok = std::net::TcpStream::connect_timeout(
&std::net::SocketAddr::from(([127, 0, 0, 1], 8000)),
std::time::Duration::from_secs(2),
).is_ok();
if !bge_ok {
eprintln!("[consolidate] Step 4: BGE HTTP (port 8000) unreachable — skipping quality backtrace");
} else {
let backtracer = QualityBacktracer::new(&args.llm_endpoint, &args.llm_model, &args.llm_api_key);
// 从 LanceDB 加载（content = distilled, source metadata = original）
let all_records = lancedb.scan_all().unwrap_or_default();
let samples: Vec<QualitySample> = all_records
.iter()
.filter(|r| !r.content.is_empty())
.map(|r| QualitySample {
memory_id: r.id.clone(),
distilled_content: r.content.clone(),
original_episode: if !r.source.is_empty() { r.source.clone() } else { r.content.clone() },
category: if r.category.is_empty() { "general".into() } else { r.category.clone() },
tier: if r.tier.is_empty() { "normal".into() } else { r.tier.clone() },
})
.collect();
eprintln!(
"[consolidate] Step 4: {} quality samples from LanceDB",
samples.len()
);
let sample_20 = quality_backtrace::stratified_sample(&samples, 5, 5, 5, 5);
if let Some(ref enc) = encoder {
let embed_fn = &|text: &str| -> Result<Vec<f32>, Box<dyn std::error::Error>> {
enc.encode(text)
// 用本地 BGE HTTP 服务编码
let encoder = match BGEEncoder::new(&args.model_dir) {
Ok(e) => Some(e),
Err(e) => {
eprintln!("[consolidate] Step 4: BGE encoder init failed: {} — skipping quality backtrace", e);
None
}
};
match backtracer.backtrace(&sample_20, embed_fn) {
Ok(result) => {
quality_score = result.score;
low_info = result.low_info_loss;
hallucinations = result.hallucinations;
}
Err(e) => {
eprintln!("[consolidate] Step 4 (quality) failed: {}", e);
if let Some(ref enc) = encoder {
let embed_fn = &|text: &str| -> Result<Vec<f32>, Box<dyn std::error::Error>> {
enc.encode(text)
};
match backtracer.backtrace(&sample_20, embed_fn) {
Ok(result) => {
quality_score = result.score;
low_info = result.low_info_loss;
hallucinations = result.hallucinations;
}
Err(e) => {
eprintln!("[consolidate] Step 4 (quality) failed: {}", e);
}
}
}
}