87 lines
3.4 KiB
Go
87 lines
3.4 KiB
Go
package memory
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"time"
|
||
)
|
||
|
||
// decayWritebackKey 是「是否允许把检索命中的衰减值物化回写」的 context 键。
|
||
// 默认允许；只读上下文（群聊 / 陌生人，design/03 §5「能查不能写」）经
|
||
// WithoutDecayWriteback 关掉它，使受限检索对记忆库零写入。
|
||
type decayWritebackKey struct{}
|
||
|
||
// WithoutDecayWriteback 返回一个禁用衰减物化回写的 context：受限（只读）
|
||
// 检索必须用它，避免在「查记忆只读不写」的群聊 / 陌生人上下文里改动记忆行。
|
||
func WithoutDecayWriteback(ctx context.Context) context.Context {
|
||
return context.WithValue(ctx, decayWritebackKey{}, false)
|
||
}
|
||
|
||
// decayWritebackEnabled 报告 ctx 是否允许衰减物化回写（未显式关闭时为 true）。
|
||
func decayWritebackEnabled(ctx context.Context) bool {
|
||
v, ok := ctx.Value(decayWritebackKey{}).(bool)
|
||
if !ok {
|
||
return true
|
||
}
|
||
return v
|
||
}
|
||
|
||
// decayRow 是一条待物化的衰减记录：命中条目 id 与其当前衰减值。
|
||
type decayRow struct {
|
||
id string
|
||
decay float64
|
||
}
|
||
|
||
// persistDecay 把本轮检索**真正命中**（relevance > 0）的条目的衰减值
|
||
// 物化回写 decay_weight，并把它们的 updated_at 推进到 now。它是 design/03
|
||
// §2 与裁决 Q3（2026-10-04）的落地：「updated_at 即衰减源 —— 读时按它计算
|
||
// 衰减；检索命中时把结果物化回写 decay_weight」「decay_weight 列语义＝
|
||
// 上次物化的衰减值」。一次命中即一次「使用」，因此同时刷新 updated_at，使
|
||
// 下一次读取从同一基准继续衰减，而不会把已物化的值再乘一遍（避免双重衰减）。
|
||
//
|
||
// query 为空时的纯新近度排序不是命中，故不物化；只读上下文
|
||
// （WithoutDecayWriteback）也不物化，以守住「群聊/陌生人不写记忆」。
|
||
//
|
||
// 物化失败绝不使检索失败：decay_weight 只是排序/展示用的派生列（design §2
|
||
// 「不作为真值源」），失败只记一条带来源的告警。
|
||
func (r *Retriever) persistDecay(ctx context.Context, results []Result, relevance map[string]float64, now time.Time) {
|
||
if !decayWritebackEnabled(ctx) {
|
||
return
|
||
}
|
||
rows := make([]decayRow, 0, len(results))
|
||
for _, res := range results {
|
||
if relevance[res.Item.ID] <= 0 {
|
||
continue
|
||
}
|
||
rows = append(rows, decayRow{id: res.Item.ID, decay: r.decayFactor(res.Item, now)})
|
||
}
|
||
if len(rows) == 0 {
|
||
return
|
||
}
|
||
if err := r.store.writeDecay(ctx, rows, now); err != nil {
|
||
r.log.Warn("memory: decay weight writeback failed", "error", err)
|
||
}
|
||
}
|
||
|
||
// writeDecay 在单个事务里把每条命中的 decay_weight 更新为其物化值，并把
|
||
// updated_at 推进到 at（命中即使用）。批量、参数化的 UPDATE，绝不拼接 SQL；
|
||
// 任一条失败即整批回滚，保持「要么都物化、要么都不动」。
|
||
func (s *Store) writeDecay(ctx context.Context, rows []decayRow, at time.Time) error {
|
||
tx, err := s.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return fmt.Errorf("memory: begin decay writeback tx: %w", err)
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
for _, row := range rows {
|
||
if _, err := tx.ExecContext(ctx,
|
||
`UPDATE memory_items SET decay_weight = ?, updated_at = ? WHERE id = ?`,
|
||
row.decay, formatTime(at), row.id); err != nil {
|
||
return fmt.Errorf("memory: update decay weight %q: %w", row.id, err)
|
||
}
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return fmt.Errorf("memory: commit decay writeback tx: %w", err)
|
||
}
|
||
return nil
|
||
}
|