自验证机制：consolidate 聚类下限 + timestamp 前置检查 + 启动数据目录检查 + 修复图谱扩展字段名
1. consolidation_pipe.go: - 前置：抽样检查记忆时间戳（< 2024-01-01 视为可疑 epoch-0） - 后置：clusters <= 1 时记录 ERROR + patterns 标记异常 2. server.go: runStartupChecks() 启动时检测废弃路径 + socket 可达性 3. graph_expander.go: Navigate 返回字段从 "target"/"source" 修正为 "to"/"from"
This commit is contained in:
xiaowei
parent
612d915eec
commit
3d054e7590
|
|
@ -50,7 +50,14 @@ func (cp *ConsolidationPipeline) Run() (*ConsolidationReport, error) {
|
|||
// mode: "cluster_only" | "full" | "prune_only"
|
||||
// full 模式触发 LLM 质量回溯 + 衰减校准（仅低频调度使用）
|
||||
func (cp *ConsolidationPipeline) RunWithMode(mode string) (*ConsolidationReport, error) {
|
||||
// ─── 尝试 Rust sidecar ──────────────────────────────
|
||||
// ─── 前置检查：timestamp 合理性（防止 epoch-0 数据污染聚类结果）────────
|
||||
if mode == "full" || mode == "cluster_only" {
|
||||
if suspicious, total := cp.checkTimestampSanity(); suspicious > 0 {
|
||||
log.Printf("[WARN] timestamp 检查: %d/%d 条记忆时间戳可疑（可能是 epoch-0），结果仅供参考", suspicious, total)
|
||||
}
|
||||
}
|
||||
|
||||
// ─── 尝试 Rust sidecar ───────────────────────────────────────
|
||||
var report *ConsolidationReport
|
||||
if cp.dataDir != "" && cp.sqlitePath != "" {
|
||||
// 用指定模式调用 Rust sidecar
|
||||
|
|
@ -64,6 +71,13 @@ func (cp *ConsolidationPipeline) RunWithMode(mode string) (*ConsolidationReport,
|
|||
Patterns: []string{fmt.Sprintf("decay_rates=%v", rustReport.DecayRates)},
|
||||
GraphPruned: 0,
|
||||
}
|
||||
|
||||
// ─── 后置检查：聚类数量下限 ─────────────────────────────
|
||||
if rustReport.Clusters <= 1 {
|
||||
log.Printf("[ERROR] consolidate 生成 clusters=%d（疑似 epoch-0 数据或聚类失效），结果可能无效", rustReport.Clusters)
|
||||
report.Patterns = append(report.Patterns, fmt.Sprintf("⚠️ clusters=%d 可能异常", rustReport.Clusters))
|
||||
}
|
||||
|
||||
if rustReport.Quality != nil {
|
||||
report.Patterns = append(report.Patterns,
|
||||
fmt.Sprintf("quality_score=%.2f low_info=%d hallucinations=%d",
|
||||
|
|
@ -270,6 +284,56 @@ func (cp *ConsolidationPipeline) updateGraph() (int, error) {
|
|||
return before - after, nil
|
||||
}
|
||||
|
||||
// checkTimestampSanity 抽样检查记忆时间戳合理性（5% sampling，不阻塞）
|
||||
// 返回 (可疑数量, 抽样总数)。epoch-0（< 2024-01-01）视为可疑。
|
||||
func (cp *ConsolidationPipeline) checkTimestampSanity() (suspicious, total int) {
|
||||
const epochThreshold int64 = 1704067200 // 2024-01-01 00:00:00 UTC
|
||||
|
||||
memories, err := cp.ldb.GetCandidatesForForgetting()
|
||||
if err != nil || len(memories) == 0 {
|
||||
return 0, 0
|
||||
}
|
||||
|
||||
// 全量不超过 100 条
|
||||
sampleSize := len(memories)
|
||||
if sampleSize > 100 {
|
||||
sampleSize = 100
|
||||
}
|
||||
step := len(memories) / sampleSize
|
||||
if step < 1 {
|
||||
step = 1
|
||||
}
|
||||
|
||||
for i := 0; i < len(memories); i += step {
|
||||
total++
|
||||
ts := memoryInt64Val(memories[i], "created_at")
|
||||
if ts > 0 && ts < epochThreshold {
|
||||
suspicious++
|
||||
}
|
||||
}
|
||||
return suspicious, total
|
||||
}
|
||||
|
||||
// memoryInt64Val 安全取 int64（处理 string/int/float）
|
||||
func memoryInt64Val(m map[string]interface{}, key string) int64 {
|
||||
v, ok := m[key]
|
||||
if !ok {
|
||||
return 0
|
||||
}
|
||||
switch val := v.(type) {
|
||||
case int64:
|
||||
return val
|
||||
case float64:
|
||||
return int64(val)
|
||||
case int:
|
||||
return int64(val)
|
||||
case string:
|
||||
t, _ := time.Parse(time.RFC3339, val)
|
||||
return t.Unix()
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
// ─── 报告 ─────────────────────────────────────────────
|
||||
|
||||
type ConsolidationReport struct {
|
||||
|
|
|
|||
|
|
@ -87,6 +87,10 @@ func NewServer() http.Handler {
|
|||
ldb = storage.NewMemLanceClient(emb)
|
||||
log.Printf("[zhiyid] 存储后端: 内存（零依赖）")
|
||||
}
|
||||
|
||||
// 启动时数据目录一致性检查（防止路径混乱导致读取废弃数据）
|
||||
runStartupChecks(backend)
|
||||
|
||||
// G7.2: 注册 LDB getter（供 skill crystallize 使用）
|
||||
routes.RegisterLDBGetter(func() storage.LanceDB { return ldb })
|
||||
|
||||
|
|
@ -1109,3 +1113,39 @@ func AgentTypeDecayOrDefault(agentType string) float64 {
|
|||
}
|
||||
return 0.015
|
||||
}
|
||||
|
||||
// runStartupChecks 启动时数据目录一致性检查
|
||||
// 检测废弃路径（如 /home/muc/data）并警告，防止数据源混乱
|
||||
func runStartupChecks(backend string) {
|
||||
canonicalDataDir := "/var/lib/memoryweave"
|
||||
deprecatedPaths := []string{
|
||||
"/home/muc/data",
|
||||
"/home/muc/.local/share/memoryweave",
|
||||
}
|
||||
|
||||
for _, dep := range deprecatedPaths {
|
||||
if _, err := os.Stat(dep); err == nil {
|
||||
log.Printf("[WARN] 检测到废弃数据目录 %s，仍有数据残留（当前 canonical: %s）", dep, canonicalDataDir)
|
||||
}
|
||||
}
|
||||
|
||||
// LanceDB 后端：检查 Rust sidecar socket 是否可达
|
||||
if backend == "lancedb" {
|
||||
sockPath := os.Getenv("LANCEDB_SOCKET")
|
||||
if sockPath == "" {
|
||||
sockPath = "/tmp/zhiyi-ipc.sock"
|
||||
}
|
||||
if _, err := os.Stat(sockPath); os.IsNotExist(err) {
|
||||
log.Printf("[WARN] LanceDB socket 不存在 (%s)，Rust sidecar 可能未运行", sockPath)
|
||||
} else {
|
||||
log.Printf("[startup] LanceDB socket 就绪: %s", sockPath)
|
||||
}
|
||||
}
|
||||
|
||||
// canonical 目录存在性检查
|
||||
if _, err := os.Stat(canonicalDataDir); os.IsNotExist(err) {
|
||||
log.Printf("[WARN] canonical 数据目录不存在: %s（首次部署？）", canonicalDataDir)
|
||||
} else {
|
||||
log.Printf("[startup] 数据目录正常: %s", canonicalDataDir)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -26,10 +26,10 @@ func (g *InMemoryGraph) ExpandFromResults(results []models.RecallResult, namespa
|
|||
continue
|
||||
}
|
||||
for _, p := range paths {
|
||||
target, _ := p["target"].(string)
|
||||
source, _ := p["source"].(string)
|
||||
to, _ := p["to"].(string)
|
||||
from, _ := p["from"].(string)
|
||||
|
||||
for _, id := range []string{target, source} {
|
||||
for _, id := range []string{to, from} {
|
||||
if id != "" && !seen[id] {
|
||||
seen[id] = true
|
||||
expanded = append(expanded, models.RecallResult{
|
||||
|
|
|
|||
Loading…
Reference in New Issue