Compare commits

...
Author SHA1 Message Date
xiaowei f1859f2ff9 perf(pagerank): O(V+E) 累加实现替代 O(V²) 全源遍历
旧实现每目标节点遍历全部源节点找边匹配 (943-961):
- 17628 nodes × 20 迭代 = 6.2亿次内层操作 → 单次 PageRank ~3 分钟
- 重启后多套触发器零值连锁 fire full consolidation → CPU 100% 持续 10+ 分钟

新实现先归一化每源节点总权重, 再沿出边把贡献累加到目标 (O(V+E)):
- 17628 nodes PageRank + 图谱维护: 3分钟 → <1 秒
- 重启风暴从 10+ 分钟 CPU 100% → 秒级无感

实测: 01:16:40 图谱维护开始 → 01:16:41 PageRank 更新 17628 nodes 完成
2026-09-06 01:20:21 +08:00
xiaowei d4343b1c47 fix(recall): Embedder http client 8s 超时 + decay 分批 AC-1 日志
- Embedder http.Client{} 无超时 → 远端 bge 半开 TCP 时 recall 永久挂起
  (gateway 10s 断 → 织忆异常误报)。加 8s 超时快速失败降级。
- decay 每 tick 限量 2000 条(最久未访问优先) + scanned/forgotten 日志
  (AC-1 验收证据)
2026-09-06 01:08:42 +08:00
xiaowei e6b1e0417c fix(forgetting): P1 防风暴 - 限长/分桶/抽样下推/decay分批
febc2c9 全表扫描 O(n²) 风暴的根治防护(生产基线 ba44e20+ 之上,不含 febc2c9 扫描代码):

1. mergeSimilar 按 category 分桶 + 单桶 maxMergePerBucket=500 截断
   - 原双层循环全量 O(n²),全表时触发 620% CPU
   - 分桶后 O(Σ桶²) << O(n²),语义不变(原 catI!=catJ continue 等价分桶)
2. scanConflicts 显式 created_at 倒序取最近 500 条
   - 原注释'最近100条'实际全量 O(n·50)
3. checkTimestampSanity 先截断 maxSanitySample=1000 再均匀抽样
   - 原全量拉取仅为 5% 抽样
4. decay 每 tick 限量 maxDecayBatch=2000,按最久未访问优先
5. 存储层输出契约统一 RFC3339 string + 补 category/created_at
   - lancedb_ipc: last_recalled_at 此前直接放 time.Time → server.go .(string)
     断言恒失败 → lastAccessed 永远默认30天前 → 遗忘判定失真(本 bug + 数据
     年龄未到 = 遗忘从不触发的双根因之一)
   - sqlite: 补 category/created_at + LIMIT 100→200(对齐 lancedb 契约)
   - 新增 memoryTimeVal/fmtTimeRFC3339 helper 兼容多类型

测试: go build ./... 全绿;gofmt 仅历史文件遗留(未全文件重排,保持最小 diff)
2026-09-06 00:56:06 +08:00
6 changed files with 149 additions and 38 deletions
+57 -20
View File
@@ -4,6 +4,7 @@ package routes
import (
"fmt"
"log"
"sort"
"strings"
"time"
@@ -207,7 +208,10 @@ func (cp *ConsolidationPipeline) RunMerge() (int, error) {
return cp.mergeSimilar()
}
// mergeSimilar 合并相似记忆(向量相似度 > 0.8 → 保留最新)
// mergeSimilar 合并相似记忆(相同 category + 高内容重叠 → 保留较新)
// 2026-09-06 P1 fix: 原实现双层循环全量 O(n²) —— febc2c9 全表扫描时触发 620% CPU 风暴。
// 改为按 category 分桶后桶内两两比较(原逻辑 catI != catJ continue 本就等价),
// 单桶超 maxMergePerBucket 截断,总计算量 O(Σ桶²) << O(n²)。
func (cp *ConsolidationPipeline) mergeSimilar() (int, error) {
// 获取全部 distilled 记忆
memories, err := cp.ldb.GetCandidatesForForgetting()
@@ -215,23 +219,36 @@ func (cp *ConsolidationPipeline) mergeSimilar() (int, error) {
return 0, err
}
const maxMergePerBucket = 500
merged := 0
// 简单启发式:相同 category + 高内容重叠 → 合并
for i := 0; i < len(memories); i++ {
for j := i + 1; j < len(memories); j++ {
catI := strVal(memories[i]["category"])
catJ := strVal(memories[j]["category"])
if catI != catJ {
continue
}
contentI := strVal(memories[i]["content"])
contentJ := strVal(memories[j]["content"])
if overlap := contentOverlap(contentI, contentJ); overlap > 0.8 {
// 保留较新的
idI := strVal(memories[i]["id"])
idJ := strVal(memories[j]["id"])
_ = cp.ldb.SoftDelete(idJ, fmt.Sprintf("merged_into_%s", idI))
merged++
// 按 category 分桶
buckets := make(map[string][]map[string]interface{})
for _, m := range memories {
cat := strVal(m["category"])
buckets[cat] = append(buckets[cat], m)
}
// 桶内两两比较(同桶才可能合并,语义与原全量双层一致)
for _, mems := range buckets {
if len(mems) > maxMergePerBucket {
// 单桶爆炸防护:只处理最近 maxMergePerBucket 条(created_at 倒序)
sort.SliceStable(mems, func(a, b int) bool {
return memoryInt64Val(mems[a], "created_at") > memoryInt64Val(mems[b], "created_at")
})
mems = mems[:maxMergePerBucket]
}
for i := 0; i < len(mems); i++ {
for j := i + 1; j < len(mems); j++ {
contentI := strVal(mems[i]["content"])
contentJ := strVal(mems[j]["content"])
if overlap := contentOverlap(contentI, contentJ); overlap > 0.8 {
// 保留较新的
idI := strVal(mems[i]["id"])
idJ := strVal(mems[j]["id"])
_ = cp.ldb.SoftDelete(idJ, fmt.Sprintf("merged_into_%s", idI))
merged++
}
}
}
}
@@ -239,12 +256,21 @@ func (cp *ConsolidationPipeline) mergeSimilar() (int, error) {
}
// Step 2: 扫描所有待解决冲突
// 2026-09-06 P1 fix: 原注释"最近 100 条"但实际对全量做实体提取+对比 → 全表时 O(n·50)。
// 显式按 created_at 取最近 maxConflictScan 条再扫。
func (cp *ConsolidationPipeline) scanConflicts() (int, error) {
// 获取最近 100 条记忆
// 获取候选并按最近优先截断(created_at 倒序)
memories, err := cp.ldb.GetCandidatesForForgetting()
if err != nil {
return 0, err
}
const maxConflictScan = 500
if len(memories) > maxConflictScan {
sort.SliceStable(memories, func(a, b int) bool {
return memoryInt64Val(memories[a], "created_at") > memoryInt64Val(memories[b], "created_at")
})
memories = memories[:maxConflictScan]
}
count := 0
for i := 0; i < len(memories); i++ {
@@ -305,17 +331,28 @@ func (cp *ConsolidationPipeline) updateGraph() (int, error) {
return before - after, nil
}
// checkTimestampSanity 抽样检查记忆时间戳合理性(5% sampling,不阻塞)
// checkTimestampSanity 抽样检查记忆时间戳合理性(均匀抽样,不阻塞)
// 返回 (可疑数量, 抽样总数)。epoch-0(< 2024-01-01)视为可疑。
// 2026-09-06 P1 fix: 原实现全量拉取只为 5% 抽样 → 先截断最近 maxSanitySample
// 条再做均匀 step 抽样,避免全量内存/IO 与排序成本。
func (cp *ConsolidationPipeline) checkTimestampSanity() (suspicious, total int) {
const epochThreshold int64 = 1704067200 // 2024-01-01 00:00:00 UTC
const maxSanitySample = 1000
memories, err := cp.ldb.GetCandidatesForForgetting()
if err != nil || len(memories) == 0 {
return 0, 0
}
// 全量不超过 100 条
// 超过 maxSanitySample 时按 created_at 取最新,再均匀抽样
if len(memories) > maxSanitySample {
sort.SliceStable(memories, func(a, b int) bool {
return memoryInt64Val(memories[a], "created_at") > memoryInt64Val(memories[b], "created_at")
})
memories = memories[:maxSanitySample]
}
// 抽样最多 100 条
sampleSize := len(memories)
if sampleSize > 100 {
sampleSize = 100
+54 -1
View File
@@ -1245,9 +1245,23 @@ func NewServer() http.Handler {
graphStore.Prune(0.15)
case "decay":
// 扫描所有记忆,按衰减率计算 freshness
// 2026-09-06 P1 fix: 每 tick 限量评估(maxDecayBatch),优先处理
// 最久未访问的(last_recalled_at 最早 → 最该遗忘),防全表单次拉取
// 内存翻倍 + 长时间阻塞触发器循环。
memories, e := ldb.GetCandidatesForForgetting()
decayScanned := 0
decayForgotten := 0
if e == nil {
const maxDecayBatch = 2000
if len(memories) > maxDecayBatch {
sort.SliceStable(memories, func(a, b int) bool {
return memoryTimeVal(memories[a], "last_recalled_at", "created_at") <
memoryTimeVal(memories[b], "last_recalled_at", "created_at")
})
memories = memories[:maxDecayBatch]
}
for _, m := range memories {
decayScanned++
// 解析 last_recalled_at
lastAccessed := time.Now().Add(-30 * 24 * time.Hour) // 默认30天前
if t, ok := m["last_recalled_at"].(string); ok && t != "" {
@@ -1277,9 +1291,14 @@ func NewServer() http.Handler {
forgetter.ScanAndForget(id)
// 实际执行软删除(持久化到 LanceDB)
_ = ldb.SoftDelete(id, "auto_forget")
decayForgotten++
}
}
}
}
}
// AC-1 验收日志(P1 修复 type 断言后 last_recalled_at 可解析;batch<=2000)
if decayScanned > 0 {
log.Printf("[decay] scanned=%d forgotten=%d (batch<=2000, 最久未访问优先)", decayScanned, decayForgotten)
}
case "gap_scan":
// 检查已有缺口:过期 7 天的自动关闭
@@ -1422,6 +1441,40 @@ func AgentTypeDecayOrDefault(agentType string) float64 {
return 0.015
}
// memoryTimeVal 取记忆时间戳用于排序(主 key 优先,空则回退副 key;均空返回 0)。
// 2026-09-06 P1: decay 分批按"最久未访问优先"排序需要;兼容 RFC3339 string / time.Time / int。
func memoryTimeVal(m map[string]interface{}, primary, fallback string) int64 {
for _, key := range []string{primary, fallback} {
v, ok := m[key]
if !ok || v == nil {
continue
}
switch val := v.(type) {
case string:
if val == "" {
continue
}
if t, err := time.Parse(time.RFC3339, val); err == nil {
return t.Unix()
}
if n, err := strconv.ParseInt(val, 10, 64); err == nil {
return n
}
case time.Time:
if !val.IsZero() {
return val.Unix()
}
case float64:
return int64(val)
case int64:
return val
case int:
return int64(val)
}
}
return 0
}
// runStartupChecks 启动时数据目录一致性检查
// 检测废弃路径(如 /home/muc/data)并警告,防止数据源混乱
func runStartupChecks(backend string) {
+19 -14
View File
@@ -941,21 +941,26 @@ func (gs *SQLiteGraphStore) PageRank(damping float64, iterations int) map[string
}
for iter := 0; iter < iterations; iter++ {
newRanks := make(map[string]float64)
for _, node := range nodes {
rank := base
for src, edges := range outEdges {
totalWt := 0.0
for _, e := range edges {
totalWt += e.weight
}
for _, e := range edges {
if e.target == node && totalWt > 0 {
rank += damping * ranks[src] * e.weight / totalWt
}
}
// 正确 PageRank 实现 O(V+E):先归一化每个源节点总权重,
// 再沿出边把贡献直接累加到目标节点(旧实现每目标遍历全源 O(V²),
// 17628 节点 × 20 迭代 = 6.2 亿次 → 3 分钟;现在亚秒级)
contrib := make(map[string]float64, len(outEdges))
for src, edges := range outEdges {
totalWt := 0.0
for _, e := range edges {
totalWt += e.weight
}
newRanks[node] = rank
if totalWt <= 0 {
continue
}
share := damping * ranks[src] / totalWt
for _, e := range edges {
contrib[e.target] += share * e.weight
}
}
newRanks := make(map[string]float64, len(nodes))
for _, node := range nodes {
newRanks[node] = base + contrib[node]
}
ranks = newRanks
}
+2 -1
View File
@@ -9,6 +9,7 @@ import (
"net/http"
"os"
"sync"
"time"
)
// Embedder bge-m3 编码客户端。优先使用本地 vLLM(端口 8000),fallback 到模力方舟 API。
@@ -33,7 +34,7 @@ func NewEmbedder(endpoint string) *Embedder {
endpoint: endpoint,
modelName: getModelName(),
apiKey: os.Getenv("MOLIFANG_API_KEY"),
httpClient: &http.Client{},
httpClient: &http.Client{Timeout: 8 * time.Second},
dim: 1024,
}
}
+14 -1
View File
@@ -497,7 +497,9 @@ func (rc *RustLanceDBClient) GetCandidatesForForgetting() ([]map[string]interfac
"id": id,
"content": m.Content,
"namespace": m.Namespace,
"last_recalled_at": m.LastRecalledAt,
"category": m.Category,
"last_recalled_at": fmtTimeRFC3339(m.LastRecalledAt),
"created_at": fmtTimeRFC3339(m.CreatedAt),
"recall_count": m.RecallCount,
"tier": m.Tier,
"importance": m.Importance,
@@ -507,6 +509,17 @@ func (rc *RustLanceDBClient) GetCandidatesForForgetting() ([]map[string]interfac
return out, nil
}
// fmtTimeRFC3339 统一时间输出契约:RFC3339 字符串;零值输出 ""。
// 2026-09-06 P1 修复:此前直接放 time.Time → server.go decay 的 .(string) 断言
// 恒失败 → lastAccessed 永远"默认30天前" → 遗忘判定失真(叠加数据年龄未到)。
// 调用方一律按 string 解析(parseTimeStr / time.Parse RFC3339)。
func fmtTimeRFC3339(t time.Time) string {
if t.IsZero() {
return ""
}
return t.Format(time.RFC3339)
}
func mapStr(m map[string]interface{}, key string) string {
if v, ok := m[key]; ok {
if s, ok := v.(string); ok {
+3 -1
View File
@@ -485,7 +485,7 @@ func (sc *SQLiteClient) GetCandidatesForForgetting() ([]map[string]interface{},
defer sc.mu.RUnlock()
rows, err := sc.query(
"SELECT id, content, namespace, last_recalled_at, recall_count, tier, importance FROM memories WHERE is_deleted=0 AND tier!='core' ORDER BY last_recalled_at ASC LIMIT 100",
"SELECT id, content, namespace, category, last_recalled_at, recall_count, tier, importance, created_at FROM memories WHERE is_deleted=0 AND tier!='core' ORDER BY last_recalled_at ASC LIMIT 200",
)
if err != nil {
return nil, err
@@ -496,10 +496,12 @@ func (sc *SQLiteClient) GetCandidatesForForgetting() ([]map[string]interface{},
"id": r["id"],
"content": r["content"],
"namespace": r["namespace"],
"category": r["category"],
"last_recalled_at": r["last_recalled_at"],
"recall_count": r["recall_count"],
"tier": r["tier"],
"importance": r["importance"],
"created_at": r["created_at"],
})
}
return results, nil