AskaEth
d6ed9cfa2b
主人要「现在就想看到在 736 条上它会建议合并哪些」，而不是等 23:00 的低峰首跑。
- 后端：Tidier.DryRun + TidyService.DryRun，**复用同一条扫描→召回→判定→构造候选路径**（抽出 tidyRoundPolicy，生产轮与试算轮只差「候选出口」与「召回降级策略」两处；禁止第二套实现）
- 🔴 零副作用：不写 memory_tidy_candidates / memory_tidy_runs，绝不改 memory_items 任何列；机械证明 = 跑前跑后 cyrene.db 与 -wal 的 SHA256 一致（测试断言「表行数不变」）
- 失败码诚实：500 tidy_unavailable / 409 tidy_disabled / 500 tidy_recall_unavailable（**embedding 不可用不降级成 0 条**）/ 503 tidy_timeout / 500 tidy_dry_run_failed；**空数组唯一含义 = 算过了没建议**
- 端点 POST /api/v1/admin/memories/tidy/dry-run（writeAuth）；面板「记忆」页加试算卡（点了才跑、无轮询、三态分开、明写「仅试算，不会改动任何记忆」）
- 真机 e2e（741 条活库副本）：76 ms / 15 分区 / 53 候选 / 41 skipped / 1 vetoed(low_confidence)；样例 merge reason=fact_subset score=1.0000、link reason=similar_but_not_contained score=0.9248
- 变异：dry-run 顺手写库 ⇒ 两条测试红（行数 0→2）；自造阈值 ⇒ 红；失败回 200 空数组 ⇒ 红
- 如实登记：Compared=688 被 400/分区护栏截断、候选上限 20/分区被打满 ⇒ 护栏值需在 741 条量级重定（非 bug）
426 lines
17 KiB
Go
426 lines
17 KiB
Go
package memory
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"log/slog"
|
||
"sort"
|
||
"strings"
|
||
"time"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding"
|
||
)
|
||
|
||
// 本文件是记忆整理（design/23-记忆整理与去重）的整理器与编排：逐分区扫描 →
|
||
// 候选召回 → 包含判定 → 规则否决 → 候选落库 + 运行留痕。
|
||
//
|
||
// 三条不可走样的口径（design/23 §0/§5.1）：
|
||
// 1. **相似度只用来召回候选对**，永不直接当合并判据；
|
||
// 2. **合并的唯一条件是「事实集合的包含关系」**（A ⊆ B 才并入 B）；
|
||
// 相似但并列的事实只建链接，绝不合并；
|
||
// 3. **规则否决一票否决**（跨作用域/跨角色/跨项目/群聊来源/时效污染/
|
||
// 数值不一致/否定冲突/低置信），命中即踢出候选。
|
||
//
|
||
// 第一版**只产建议**：RunOnce 唯一的写入是候选表与运行表，
|
||
// 绝不改写、绝不删除任何一条记忆（design/23 §3 非目标 + §7）。
|
||
|
||
// TidyAction 是整理建议的动作类别（design/23 §10 的 action 枚举）。
|
||
type TidyAction string
|
||
|
||
const (
|
||
// TidyMerge 表示 A 的事实集合是 B 的子集，建议把 A 并入 B（保留 B）。
|
||
// 它仍需人工确认（design/23 §7），第一版绝不自动应用。
|
||
TidyMerge TidyAction = "merge"
|
||
// TidyLink 表示两条相似但**互不包含**，只建链接/分组，绝不合并
|
||
// （design/23 §1.3 的 [1]/[3]「红牛 vs 冰美式」）。
|
||
TidyLink TidyAction = "link"
|
||
// TidyArchive 表示含绝对日期/时刻的一次性事件，建议归档而非合并
|
||
// （design/23 §5.1 T0-5）。
|
||
TidyArchive TidyAction = "archive"
|
||
// TidyReclassify 表示 kind 误分类（design/23 §5.1 T0-6），只产候选不自动改。
|
||
TidyReclassify TidyAction = "reclassify"
|
||
// TidyKeep 是「拿不准 = 什么都不做」（design/23 §5.2-10）。它是合法动作，
|
||
// 但第一版不为它落库（只进否决计数与日志），以免候选队列被噪声填满。
|
||
TidyKeep TidyAction = "keep"
|
||
)
|
||
|
||
// TidyVetoReason 是规则否决的原因标签：进日志、进运行表的原因分布
|
||
// （design/23 §9「每轮记否决原因分布」），也是测试断言的抓手。
|
||
type TidyVetoReason string
|
||
|
||
const (
|
||
// VetoNone 表示未命中任何否决规则。
|
||
VetoNone TidyVetoReason = ""
|
||
// VetoScope 是跨作用域硬门（design/23 §5.2-7）：user/persona/project 之间永不合。
|
||
VetoScope TidyVetoReason = "scope"
|
||
// VetoPersona 是跨角色硬门（design/23 §5.2-8）。
|
||
VetoPersona TidyVetoReason = "persona"
|
||
// VetoProject 是跨项目硬门（design/23 §5.2-8）。
|
||
VetoProject TidyVetoReason = "project"
|
||
// VetoSource 是群聊/陌生人来源（design/23 §5.2-9、§6）：只读不写。
|
||
VetoSource TidyVetoReason = "source"
|
||
// VetoTemporal 是时效污染（design/23 §5.2-1）：一次性事件/有期限状态
|
||
// 不等于长期事实（[8]「9月8日晚」≠ 熬夜习惯）。
|
||
VetoTemporal TidyVetoReason = "temporal"
|
||
// VetoNumeric 是数值/配置不一致（design/23 §5.2-2）：数值不同即不同事实。
|
||
VetoNumeric TidyVetoReason = "numeric"
|
||
// VetoConflict 是否定/冲突（design/23 §5.2-3）：X vs 非X 是覆盖不是重复。
|
||
VetoConflict TidyVetoReason = "conflict"
|
||
// VetoLowConfidence 是低置信否决门（design/23 §5.2-6）。
|
||
VetoLowConfidence TidyVetoReason = "low_confidence"
|
||
)
|
||
|
||
// ErrNilStore 在整理器没有收到记忆存储时返回。
|
||
var ErrNilStore = errors.New("memory: tidy needs a store")
|
||
|
||
// ErrTidyEmbedCount 在 embedder 返回的向量条数与请求条数不一致时返回：
|
||
// 缺失向量会让相似度与条目错位，宁可整体回退到词项余弦。
|
||
var ErrTidyEmbedCount = errors.New("memory/tidy: embedder returned a mismatched vector count")
|
||
|
||
// TidyRun 是一轮整理的留痕（design/23 §10 memory_tidy_runs）。
|
||
type TidyRun struct {
|
||
ID string
|
||
StartedAt time.Time
|
||
FinishedAt time.Time
|
||
Scanned int
|
||
Candidates int
|
||
Applied int
|
||
Skipped int
|
||
Errors int
|
||
Vetoed int
|
||
VetoReasons map[string]int
|
||
RuleVersion string
|
||
}
|
||
|
||
// TidyStats 是一轮整理的观测口径：**扫描 N / 候选 M / 建议 K / 否决原因分布**
|
||
// （design/23 §9）。它同时写进日志与运行表，绝不静默。
|
||
type TidyStats struct {
|
||
Scanned int
|
||
Partitions int
|
||
Compared int
|
||
Pairs int
|
||
Candidates int
|
||
Duplicates int
|
||
Skipped int
|
||
Vetoed int
|
||
VetoReasons map[string]int
|
||
// Errors 是本轮的失败计数（分区失败 + 候选写入失败），进运行表。
|
||
Errors int
|
||
// Truncated 是被上限截断的分区标签（扫描或比对上限），供运维放大配额。
|
||
Truncated []string
|
||
}
|
||
|
||
// TidyDeps 是注入给整理器的依赖。Store 必需；其余可空并优雅降级。
|
||
type TidyDeps struct {
|
||
// Store 是记忆存储。必需 —— 整理只读写它，绝不接触会话历史。
|
||
Store *Store
|
||
// Embedder 是候选召回的相似度来源。为 nil 时回退到零依赖的
|
||
// 字符二元组余弦（离线可用、测试确定性），见 tidy_recall.go。
|
||
Embedder embedding.Embedder
|
||
// Audit 是审计出口（design/23 §8 的 memory.tidy.* 动作）。可选：
|
||
// 组合根把它接到网关的 AuditEvent；为 nil 时只记日志。
|
||
Audit func(who, what, result string)
|
||
// Now 是运行时间戳的墙钟来源，便于测试注入。
|
||
Now func() time.Time
|
||
// Log 是基础日志器，来源标记为 "memory/tidy"。
|
||
Log *slog.Logger
|
||
}
|
||
|
||
// Tidier 是记忆整理器。它并发安全**仅限**「一轮之内不重入」：
|
||
// 服务层用一把互斥锁保证同一时刻只有一轮（design/23 §9 单写者）。
|
||
type Tidier struct {
|
||
cfg TidyConfig
|
||
deps TidyDeps
|
||
log *slog.Logger
|
||
now func() time.Time
|
||
}
|
||
|
||
// NewTidier 由 cfg 与 deps 构建整理器。Store 为 nil 时返回 ErrNilStore。
|
||
func NewTidier(cfg TidyConfig, deps TidyDeps) (*Tidier, error) {
|
||
if deps.Store == nil {
|
||
return nil, ErrNilStore
|
||
}
|
||
cfg = normalizeTidyConfig(cfg)
|
||
now := deps.Now
|
||
if now == nil {
|
||
now = time.Now
|
||
}
|
||
return &Tidier{
|
||
cfg: cfg,
|
||
deps: deps,
|
||
log: logger.WithSource(baseTidyLog(deps.Log), "memory/tidy"),
|
||
now: now,
|
||
}, nil
|
||
}
|
||
|
||
// Config 返回生效的整理配置（已套默认值）。
|
||
func (t *Tidier) Config() TidyConfig { return t.cfg }
|
||
|
||
// tidyCandidateSink 是**本轮的候选出口**：一个分区比完就把候选交给它。
|
||
// 生产轮是落库（persistCandidates），试算轮只收集（tidy_dryrun.go）——
|
||
// 规则引擎本身与出口无关，这正是「禁止写第二套实现」的落点。
|
||
type tidyCandidateSink func(ctx context.Context, cands []TidyCandidate, stats *TidyStats)
|
||
|
||
// tidyRecall 是**本轮的召回策略**：给定已截断的参与比对条目，产出候选对。
|
||
// 它把「相似度来源不可用时怎么办」的差异收在一处：生产轮降级到零依赖词项
|
||
// 余弦，试算轮如实上报错误。
|
||
type tidyRecall func(ctx context.Context, p tidyPartition, items []MemoryItem, stats *TidyStats) ([]tidyPair, error)
|
||
|
||
// tidyRoundPolicy 是本轮整理的出口策略：候选去哪 + 召回来源不可用时怎么办。
|
||
// 「生产轮」与「试算轮」的**全部**差异都在这两个字段上，规则引擎其余部分
|
||
// 逐字共用。
|
||
type tidyRoundPolicy struct {
|
||
sink tidyCandidateSink
|
||
recall tidyRecall
|
||
}
|
||
|
||
// RunOnce 跑一轮整理，并写一条运行留痕。它返回本轮的观测计数。
|
||
//
|
||
// 失败隔离（design/23 §9）：单个分区失败不影响其它分区，失败计数进运行表与
|
||
// 日志；只有「所有分区都失败」或运行留痕本身写不进去时才回错误。
|
||
func (t *Tidier) RunOnce(ctx context.Context) (TidyStats, error) {
|
||
started := t.now().UTC()
|
||
stats, partitions, err := t.beginRound(ctx)
|
||
if err != nil {
|
||
return stats, err
|
||
}
|
||
|
||
failures := t.scanPartitions(ctx, partitions, &stats, tidyRoundPolicy{
|
||
sink: t.persistCandidates,
|
||
recall: t.fallbackRecall,
|
||
})
|
||
|
||
run := TidyRun{
|
||
// 运行 id 在这里就定下来：留痕与日志必须指向同一条运行记录。
|
||
ID: mustNewTidyID(),
|
||
StartedAt: started,
|
||
FinishedAt: t.now().UTC(),
|
||
Scanned: stats.Scanned,
|
||
Candidates: stats.Candidates,
|
||
Applied: 0, // 第一版绝不自动应用（design/23 §7）
|
||
Skipped: stats.Skipped + stats.Duplicates,
|
||
Errors: stats.Errors,
|
||
Vetoed: stats.Vetoed,
|
||
VetoReasons: stats.VetoReasons,
|
||
RuleVersion: TidyRuleVersion,
|
||
}
|
||
if err := t.deps.Store.persistTidyRun(ctx, run); err != nil {
|
||
failures = append(failures, err)
|
||
}
|
||
|
||
t.logRun(stats, run)
|
||
|
||
switch {
|
||
case len(failures) == 0:
|
||
return stats, nil
|
||
case len(failures) == len(partitions) && len(partitions) > 0:
|
||
return stats, fmt.Errorf("memory/tidy: every partition failed: %w", errors.Join(failures...))
|
||
default:
|
||
return stats, fmt.Errorf("memory/tidy: run finished with errors: %w", errors.Join(failures...))
|
||
}
|
||
}
|
||
|
||
// logRun 输出本轮的可观测口径：扫描 N / 候选 M / 建议 K / 否决原因分布。
|
||
// 它带 source（memory/tidy），因此这条汇总永远可回溯到整理子系统。
|
||
func (t *Tidier) logRun(stats TidyStats, run TidyRun) {
|
||
fields := []any{
|
||
"run_id", run.ID,
|
||
"rule_version", run.RuleVersion,
|
||
"partitions", stats.Partitions,
|
||
"scanned", run.Scanned,
|
||
"compared", stats.Compared,
|
||
"pairs", stats.Pairs,
|
||
"candidates", run.Candidates,
|
||
"duplicates", stats.Duplicates,
|
||
"skipped", stats.Skipped,
|
||
"vetoed", run.Vetoed,
|
||
"veto_reasons", tidyReasonsLabel(run.VetoReasons),
|
||
"errors", run.Errors,
|
||
"duration_ms", run.FinishedAt.Sub(run.StartedAt).Milliseconds(),
|
||
"recall_mode", tidyRecallMode(t.deps.Embedder != nil),
|
||
}
|
||
if len(stats.Truncated) > 0 {
|
||
fields = append(fields, "truncated", strings.Join(stats.Truncated, ","))
|
||
}
|
||
t.log.Info("memory tidy run finished", fields...)
|
||
}
|
||
|
||
// beginRound 开一轮：准备口径容器并列出待整理分区。分区列表拿不到时整轮失败，
|
||
// 且**不写运行留痕** —— 那一轮根本没开始（口径与 design §9 一致）。
|
||
func (t *Tidier) beginRound(ctx context.Context) (TidyStats, []tidyPartition, error) {
|
||
stats := TidyStats{VetoReasons: make(map[string]int)}
|
||
partitions, err := t.deps.Store.listTidyPartitions(ctx)
|
||
if err != nil {
|
||
return stats, nil, fmt.Errorf("memory/tidy: list partitions: %w", err)
|
||
}
|
||
stats.Partitions = len(partitions)
|
||
return stats, partitions, nil
|
||
}
|
||
|
||
// scanPartitions 逐个分区跑规则并把候选交给 policy.sink。它返回失败列表
|
||
// （单个分区失败不阻塞其它分区，design §9）；成功时返回 nil。
|
||
func (t *Tidier) scanPartitions(ctx context.Context, partitions []tidyPartition, stats *TidyStats, policy tidyRoundPolicy) []error {
|
||
var failures []error
|
||
for _, p := range partitions {
|
||
if err := t.runPartition(ctx, p, stats, policy); err != nil {
|
||
stats.Errors++
|
||
failures = append(failures, err)
|
||
t.log.Warn("memory tidy partition failed", "partition", p.label(), "error", err)
|
||
}
|
||
}
|
||
return failures
|
||
}
|
||
|
||
// fallbackRecall 是**生产轮**的召回：相似度来源不可用时降级到零依赖的词项
|
||
// 余弦（离线/无 embedding 配置时可跑，宁可比语义路径更保守），并记一条 WARN。
|
||
// 两个来源**不可混用阈值**：阈值只有一个（design §13 开放点 2 的 0.80）。
|
||
func (t *Tidier) fallbackRecall(ctx context.Context, p tidyPartition, items []MemoryItem, stats *TidyStats) ([]tidyPair, error) {
|
||
pairs, err := t.recallPairs(ctx, p, items, stats)
|
||
if err == nil {
|
||
return pairs, nil
|
||
}
|
||
t.log.Warn("memory tidy embedding recall unavailable, falling back to lexical",
|
||
"partition", p.label(), "error", err)
|
||
return collectTidyPairs(items, t.cfg.RecallThreshold, lexicalSimFunc(items), tidyRecallLexical, stats), nil
|
||
}
|
||
|
||
// runPartition 整理一个分区（persona_id + scope + project_id）。
|
||
// 分区之间**永不比较**（design/23 §6 硬约束）。候选去处与召回降级策略都由
|
||
// policy 决定，因此生产轮与试算轮走的是**同一条**扫描→召回→判定→构造候选
|
||
// 的路径。
|
||
func (t *Tidier) runPartition(ctx context.Context, p tidyPartition, stats *TidyStats, policy tidyRoundPolicy) error {
|
||
items, err := t.deps.Store.listTidyPartitionItems(ctx, p, t.cfg.ScanLimit)
|
||
if err != nil {
|
||
return fmt.Errorf("memory/tidy: list partition %s: %w", p.label(), err)
|
||
}
|
||
stats.Scanned += len(items)
|
||
if len(items) == t.cfg.ScanLimit {
|
||
// 扫描上限命中：本分区更大的部分这一轮不看（design §9「超限分片，下一轮
|
||
// 继续」）。本实现只做「最新优先」的截断、不推进分片游标，因此这里必须
|
||
// 留下痕迹，否则「配置了 5000 却永远只看最新 5000」会无人知晓。
|
||
stats.Truncated = append(stats.Truncated, p.label()+"(scan)")
|
||
t.log.Warn("memory tidy partition hit the scan limit",
|
||
"partition", p.label(), "limit", t.cfg.ScanLimit)
|
||
}
|
||
|
||
comparable, archives := t.classifyItems(items, stats)
|
||
|
||
// 截断与计数只做一次：召回策略拿到的是**已截断**的条目，回退重算不重复计数。
|
||
compared := t.limitCompareItems(p, comparable, stats)
|
||
pairs, err := policy.recall(ctx, p, compared, stats)
|
||
if err != nil {
|
||
return fmt.Errorf("memory/tidy: recall partition %s: %w", p.label(), err)
|
||
}
|
||
|
||
candidates := archives
|
||
for _, pr := range pairs {
|
||
verdict := judgePair(pr.a, pr.b, pr.score)
|
||
if verdict.Veto != VetoNone {
|
||
stats.Vetoed++
|
||
stats.VetoReasons[string(verdict.Veto)]++
|
||
continue
|
||
}
|
||
candidates = append(candidates, buildCandidates(pr, verdict, t.now().UTC())...)
|
||
}
|
||
// 每分区候选上限（design §9）对**全部候选**生效（含归档建议），
|
||
// 超出的只登记不入队，并记进 stats.Truncated（绝不静默）。
|
||
if len(candidates) > t.cfg.CandidateLimit {
|
||
stats.Skipped += len(candidates) - t.cfg.CandidateLimit
|
||
stats.Truncated = append(stats.Truncated, p.label()+"(candidates)")
|
||
candidates = candidates[:t.cfg.CandidateLimit]
|
||
}
|
||
policy.sink(ctx, candidates, stats)
|
||
return nil
|
||
}
|
||
|
||
// classifyItems 把分区内条目分流成**可比对**与**归档候选**（跳过的不产出）：
|
||
// - 群聊/陌生人来源 → 排除在扫描范围外（design/23 §6，只读不写）；
|
||
// - 含绝对日期/时刻 → 一次性事件，走归档而非合并（[8]「9月8日晚22:38」），
|
||
// 且**不参与任何 merge**（design §5.1-5）；
|
||
// - 过短内容 → 没有可判定的事实要素，跳过（不产噪声候选）。
|
||
func (t *Tidier) classifyItems(items []MemoryItem, stats *TidyStats) (comparable []MemoryItem, archives []TidyCandidate) {
|
||
comparable = make([]MemoryItem, 0, len(items))
|
||
for _, it := range items {
|
||
switch t.classifyForTidy(it, stats) {
|
||
case tidySkip:
|
||
continue
|
||
case tidyArchive:
|
||
archives = append(archives, archiveCandidate(it, t.now().UTC()))
|
||
continue
|
||
default:
|
||
comparable = append(comparable, it)
|
||
}
|
||
}
|
||
return comparable, archives
|
||
}
|
||
|
||
// tidyDisposition 是扫描时一条记忆的处置分流。
|
||
type tidyDisposition int
|
||
|
||
const (
|
||
tidyCompare tidyDisposition = iota
|
||
tidyArchive
|
||
tidySkip
|
||
)
|
||
|
||
// classifyForTidy 决定一条记忆进入比对、进归档候选，还是被排除：
|
||
// - 群聊/陌生人来源 → 排除在扫描范围外（design/23 §6，只读不写）；
|
||
// - 含绝对日期/时刻 → 一次性事件，走归档而非合并（[8]「9月8日晚22:38」）；
|
||
// - 过短内容 → 没有可判定的事实要素，跳过（不产噪声候选）。
|
||
func (t *Tidier) classifyForTidy(it MemoryItem, stats *TidyStats) tidyDisposition {
|
||
if nonOwnerSource(it.SourceConversation) {
|
||
stats.Vetoed++
|
||
stats.VetoReasons[string(VetoSource)]++
|
||
return tidySkip
|
||
}
|
||
if hasAbsoluteTimestamp(it.Content) {
|
||
return tidyArchive
|
||
}
|
||
if len([]rune(it.Content)) < minTidyContentRunes {
|
||
return tidySkip
|
||
}
|
||
return tidyCompare
|
||
}
|
||
|
||
// persistCandidates 落库一批候选：幂等键已存在时只计入 duplicates
|
||
// （design/23 §9「重复跑不重复产候选」）。
|
||
func (t *Tidier) persistCandidates(ctx context.Context, cands []TidyCandidate, stats *TidyStats) {
|
||
for _, c := range cands {
|
||
created, err := t.deps.Store.insertTidyCandidate(ctx, c)
|
||
if err != nil {
|
||
stats.Errors++
|
||
t.log.Warn("memory tidy candidate insert failed", "action", string(c.Action), "error", err)
|
||
continue
|
||
}
|
||
if !created {
|
||
stats.Duplicates++
|
||
continue
|
||
}
|
||
stats.Candidates++
|
||
}
|
||
}
|
||
|
||
// tidyReasonsLabel 把否决原因分布渲染成稳定的日志字符串（按原因排序）。
|
||
func tidyReasonsLabel(reasons map[string]int) string {
|
||
if len(reasons) == 0 {
|
||
return "none"
|
||
}
|
||
keys := make([]string, 0, len(reasons))
|
||
for k := range reasons {
|
||
keys = append(keys, k)
|
||
}
|
||
sort.Strings(keys)
|
||
parts := make([]string, 0, len(keys))
|
||
for _, k := range keys {
|
||
parts = append(parts, fmt.Sprintf("%s=%d", k, reasons[k]))
|
||
}
|
||
return strings.Join(parts, ",")
|
||
}
|
||
|
||
// baseTidyLog 在 log 非 nil 时返回它，否则返回丢弃式日志器。
|
||
func baseTidyLog(log *slog.Logger) *slog.Logger {
|
||
return discardLogger(log)
|
||
}
|