From fbad02899a516517a565fd79fbaa63ef0174665c Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sun, 4 Oct 2026 14:20:22 +0800 Subject: [PATCH] =?UTF-8?q?fix(memory,loop,capacity):=20=E6=89=93=E9=80=9A?= =?UTF-8?q?=E8=AE=B0=E5=BF=86=E9=93=BE=E4=B8=89=E6=9D=A1=E7=A9=BA=E5=A3=B3?= =?UTF-8?q?=EF=BC=88P0-3=20/=20P0-4=20/=20P0-17=EF=BC=8C=E5=AE=A1=E8=AE=A1?= =?UTF-8?q?=20=C2=A74.3=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - P0-3 持久向量索引**不再空壳**：Store.Add 落库后真做 Embed+Add（失败只记日志、绝不让记忆写入失败）； NewRetriever 拿到 WithVectorIndex 时**自动接上写入路径**（无需改组合根）；索引无命中 → WARN + **回退内存余弦**； 新增每进程一次的**回填**（索引确实为空时把已落库条目向量化，分批 64、上限 512） - P0-4 工作记忆**按会话接线**：回合落定时写入（owner 会话 + 带会话 id 的窗口，LRU 上限 64）， 组装上下文时按会话读取 ⇒ 前导块真的出现；原单窗口 API 语义不变 - P0-17 L4 学习回环**真回写**：新增 capacity.Judge（只跑判定、不施人格覆盖/不发事件/不改状态）； loop 在回合落定"只写 owner 会话 + 跳过子代理分支"，写 ScopeUser（L4 的读域）+ 结构化溯源 - 顺带红线加固：意图记录是主人**原话**的机器可读形态 ⇒ **绝不注入群聊/陌生人块**（design/03 §5） - 复核：三条各自的包 build/vet/test 绿；「真生效」断言含生产构造路径（vector.New + 真实 SQLite 后端 + n-gram embedder）、跑完一轮窗口非空、L4 回写后能唤醒一个自身无强信号的表达； **5 条变异反证**（其中两条专门打「受限会话零写入」与「群聊不泄露主人原话」）均已还原 - 未做（登记）：Index.Search 无 ID 过滤（客户端按候选集过滤，极端情况走回退）；每条记忆一次 Embed --- internal/runtime/capacity/recall.go | 26 ++ internal/runtime/loop/memory.go | 123 ++++++++- internal/runtime/memory/retrieve.go | 130 ++++++--- internal/runtime/memory/retrieve_index.go | 110 ++++++++ internal/runtime/memory/retrieve_semantic.go | 44 +++ internal/runtime/memory/store.go | 18 +- internal/runtime/memory/store_vector.go | 68 +++++ internal/runtime/memory/vector_sync_test.go | 220 +++++++++++++++ internal/runtime/memory/working.go | 255 ++++++++++++++---- .../runtime/memory/working_session_test.go | 107 ++++++++ 10 files changed, 1020 insertions(+), 81 deletions(-) create mode 100644 internal/runtime/memory/retrieve_index.go create mode 100644 internal/runtime/memory/retrieve_semantic.go create mode 100644 internal/runtime/memory/store_vector.go create mode 100644 internal/runtime/memory/vector_sync_test.go create mode 100644 internal/runtime/memory/working_session_test.go diff --git a/internal/runtime/capacity/recall.go b/internal/runtime/capacity/recall.go index 2106d78..62cc4e2 100644 --- a/internal/runtime/capacity/recall.go +++ b/internal/runtime/capacity/recall.go @@ -1,6 +1,7 @@ package capacity import ( + "context" "math" "strconv" "strings" @@ -56,6 +57,31 @@ const ( intentRecordPrefix = "intent:" ) +// Judge 只跑意图分层（L1 → L2 → L3 → L4），返回该轮的原始判定 +// （design §2.4 闭环学习的写回对象）：它不施加人格覆盖、不产出事件、 +// 不改动任何状态，也不写任何东西。 +// +// 它是 Decide 的「只看判定」视图：回合落定时调它一次即可拿到该轮判定 +// 用于回写记忆（RecordIntent），既不必重跑整次决策，也不会重复触发 +// capacity_awaken 广播。判定层被禁用或没有可判文本时返回零值 +// （IsTask=false），调用方据此跳过写入。 +func (d *Decider) Judge(ctx context.Context, in Input) Judgment { + if !d.cfg.Enabled || len(in.Messages) == 0 { + return Judgment{} + } + state, conf, tool := d.intent(ctx, in) + return judgmentFor(in, state, conf, tool) +} + +// IsIntentRecord 报告 content 是否为 RecordIntent 写回的规范意图记录 +// （design §2.4）。群聊/陌生人的记忆注入据此把主人原话的机器可读形态 +// 挡在只读共享块之外：那是 L4 的学习素材与「主人敏感信息」， +// 不是给群聊消费的非敏感共享记忆。 +func IsIntentRecord(content string) bool { + _, ok := parseIntentRecord(content) + return ok +} + // RecordIntent formats a judgment into the canonical intent-record string written // back to memory (design §2.4). Only task judgments are worth learning — a quiet / // non-task judgment returns "" so callers can skip persisting noise. The format is diff --git a/internal/runtime/loop/memory.go b/internal/runtime/loop/memory.go index 86cd909..8e3487d 100644 --- a/internal/runtime/loop/memory.go +++ b/internal/runtime/loop/memory.go @@ -5,6 +5,7 @@ import ( "strings" "time" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/capacity" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" @@ -77,6 +78,10 @@ func (l *Loop) memoryBlock(ctx context.Context, messages []model.Message) string // （design/03 §5「不泄露主人敏感信息」+ 主人 2026-10-03 // 裁决「用户习惯记忆与不敏感数据：可查、不可写」）。 // +// 同样按敏感项排除的是 L4 意图记录（capacity.RecordIntent 的回写格式， +// 内容是主人的逐句原话）：它虽落在共享的用户习惯域以便 L4 闭环学习， +// 但群聊/陌生人只该看到「不敏感的习惯结论」，不该看到主人原话。 +// // 召回本身包在一个 memory span 里（裁决 11.4.2，design/19 §1 // 旁路节点）：span 只记元数据（作用域/命中条数/注入块字节数/耗时）， // 绝不记召回了什么内容。 @@ -95,9 +100,17 @@ func (l *Loop) retrieveMemoryBlock(ctx context.Context, env SessionEnv, personaI } shared := make([]memory.Result, 0, len(results)) for _, r := range results { - if sharedScope(r.Item.Scope) { - shared = append(shared, r) + if !sharedScope(r.Item.Scope) { + continue } + // L4 意图记录（capacity.RecordIntent 的回写格式）是主人原话的 + // 机器可读形态，属主人敏感信息：共享域可查的是「不敏感的用户 + // 习惯」，主人的逐句原话绝不注入群聊/陌生人上下文 + // （design/03 §5「不泄露主人敏感信息」）。 + if capacity.IsIntentRecord(r.Item.Content) { + continue + } + shared = append(shared, r) } if len(shared) == 0 { l.finishMemoryRecallSpan(span, "restricted", 0, 0, nil) @@ -162,7 +175,11 @@ func (l *Loop) assembleContext(ctx context.Context, messages []model.Message, ro leading = append(leading, l.currentTimeBlock()) } if l.working != nil { - if s := l.working.Summarize(); s != "" { + // 读**本会话**的窗口（回合落定时由 recordWorkingMemory 写入）： + // 进程内只有一个 WorkingMemory，按会话读才不至于把别的会话 + // （尤其是群聊/陌生人）的工作记忆混进主人的上下文。该会话 + // 尚无窗口（如首轮之前）时摘要为空，前导块自然不出现。 + if s := l.working.SummarizeSession(loopSession(ctx)); s != "" { leading = append(leading, model.TextMessage(model.RoleSystem, "工作记忆: "+s)) } } @@ -237,7 +254,13 @@ func (l *Loop) extractMemory(ctx context.Context, messages []model.Message, step // 被提取（只读隔离，design/02 §3 / design/11 §2），缺少 // extractor 或 store 时为空操作。提取失败与 store 失败会 // 带循环的来源记录日志，绝不向上传播。 +// +// 它同时是 loop 侧**回合落定**的记忆钩子入口（run 路径每个已落定 +// 的 turn 调用它一次）：无论提取器是否配置，都在此先落下工作记忆 +// 与 L4 判定回写（见 recordTurnMemory）。两者各自带门（工作记忆写 +// 会话窗口、L4 只写 owner 会话），因此不依赖这里的 extractor 门。 func (l *Loop) extractMemoryAsync(ctx context.Context, messages []model.Message, step int) { + l.recordTurnMemory(ctx, messages, step) env := sessionEnvFrom(ctx) if env.Restricted() { l.log.Debug("memory extract skipped: read-only context", "step", step, "env", env.ActorEnv) @@ -318,6 +341,100 @@ func (l *Loop) persistMemories(ctx context.Context, items []memory.MemoryItem, s l.finishMemoryWriteSpan(span, written, failed, nil) } +// recordTurnMemory 是回合落定时 loop 拥有的记忆钩子：把该会话的当前 +// 对话窗口写进工作记忆（三层记忆的层①），并把该轮的 L4 判定回写进 +// 记忆（design/14 §2.2 L4 / §2.4 闭环学习）。它由 extractMemoryAsync +// 在每个已落定的 turn 上调用一次 —— 落点是 run 路径既有的 turn 结束 +// 钩子，因此 sub-agent 与主会话走同一条路径。 +// +// 两个钩子各自带门（见 recordWorkingMemory / recordIntent），因此 +// 未配置 capacity、未配置检索器、或处于群聊/陌生人环境时它们 +// 安静跳过，绝不因为「钩子跑了」而多写一行记忆。 +func (l *Loop) recordTurnMemory(ctx context.Context, messages []model.Message, step int) { + l.recordWorkingMemory(ctx, messages) + l.recordIntent(ctx, messages, step) +} + +// recordWorkingMemory 把该会话的对话窗口记进工作记忆（design/03 §0 +// 层①「当前会话的对话窗口（滑动窗口 + 摘要）」），使 assembleContext +// 的「工作记忆: …」前导块与 Summarize 真的有内容 —— 接线之前生产 +// 代码从不调用 Add，窗口恒空、摘要恒 ""（审计 P0-4）。 +// +// 两道门： +// - 只写**主人会话**（MemWritable）：群聊/陌生人的对话不入工作记忆 +// （design/02 §3、design/11 §2 的只读隔离）。 +// - 只写带会话 id 的窗口：进程内只有一个 WorkingMemory 实例，无 +// 会话键的写入会污染别的会话的读点。 +// +// 传入的是该回合的完整对话（含 checkpoint 还原出的历史），窗口因此 +// 就是「最近 N 条」滑窗；会话恢复时同一条路径把历史回填进窗口， +// 而不是从空白重启（design §1 durable execution）。 +func (l *Loop) recordWorkingMemory(ctx context.Context, messages []model.Message) { + if l.working == nil || len(messages) == 0 { + return + } + if !sessionEnvFrom(ctx).MemWritable { + return + } + sessionID := loopSession(ctx) + if sessionID == "" { + return + } + l.working.RecordSession(sessionID, messages) +} + +// recordIntent 把该轮的 L4 判定写回记忆（design/14 §2.2 L4「写入记忆… +// 让意图理解越用越准」、§2.4 闭环学习）。接线之前 capacity.RecordIntent +// 在生产代码里零调用，判定层因此只读不写，「越用越准」永远不成立 +// （审计 P0-17）。 +// +// 红线：**只写 owner 会话**。群聊/陌生人有独立的记忆域且只读不写 +// （design/02 §3、design/11 §2、design/03 §5），子代理的内部回合也 +// 不是用户本人的表达习惯（经 flowBranch 识别），两者都不写。 +// 判定层禁用、无角色、缺 store 时为空操作；判定为非任务时 +// capacity.RecordIntent 返回 ""，跳过写入（不堆噪声记忆）。 +// +// 条目落进**用户习惯域**（ScopeUser，design/03 §3「用户习惯记忆全局 +// 共享」）：L4 学的正是主人的任务表达习惯，写入与读取因此在同一个 +// 作用域里闭环（capacity L4 只采信 user/persona/project 域的召回）。 +func (l *Loop) recordIntent(ctx context.Context, messages []model.Message, step int) { + if l.capacity == nil || !l.capacity.Enabled() || l.memory == nil || l.retriever == nil { + return + } + if !sessionEnvFrom(ctx).MemWritable { + return + } + if _, isBranch := flowBranch(ctx); isBranch { + return + } + personaID := l.personaFor(ctx) + if personaID == "" { + return + } + judgment := l.capacity.Judge(ctx, capacity.Input{ + Messages: capacityTexts(messages), + Awakening: l.awakeningFor(ctx), + Recall: l.capacityRecall(ctx, messages), + }) + content := l.capacity.RecordIntent(judgment) + if content == "" { + return + } + item := memory.MemoryItem{ + PersonaID: personaID, + Scope: memory.ScopeUser, + Kind: memory.KindRelationship, + Content: content, + Confidence: judgment.Confidence, + SourceConversation: memory.FormatSourceConversation(loopSession(ctx), loopTurn(ctx)), + } + if _, err := l.memory.Add(ctx, item); err != nil { + l.log.Warn("memory: L4 intent writeback failed", "step", step, "error", err) + return + } + l.log.Debug("memory: L4 intent writeback stored", "step", step, "task_kind", judgment.TaskKind) +} + // queryFromMessages 从对话导出检索查询：最后一个用户 // turn 的文本，回退到最后一条消息的文本，再回退到 ""。 func queryFromMessages(messages []model.Message) string { diff --git a/internal/runtime/memory/retrieve.go b/internal/runtime/memory/retrieve.go index 4f9a733..7250d68 100644 --- a/internal/runtime/memory/retrieve.go +++ b/internal/runtime/memory/retrieve.go @@ -3,11 +3,13 @@ package memory import ( "context" "fmt" - "math" + "log/slog" "sort" "strings" + "sync" "time" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory/projectindex" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory/vector" @@ -58,7 +60,7 @@ type Result struct { // 加共享的用户/项目作用域（design §3/§4，见 visibleCandidates）—— // 按相关性（关键词，以及配置了 embedder 或向量索引时的 // 语义向量）加权衰减（decay_weight × 时间衰减）排序后返回 -// 前 K 条。它是无状态的、并发安全；排名在 +// 前 K 条。它并发安全（索引装配状态与一次性回填有锁保护）；排名在 // 每次调用时从存储重算，因此结果始终反映持久化的 // 状态。 type Retriever struct { @@ -69,6 +71,14 @@ type Retriever struct { keywordWeight float64 semanticWeight float64 recencyHalfLife time.Duration + log *slog.Logger + + // seedMu 保护一次性索引回填状态（seeded 与 fallbackWarned）： + // 回填每进程最多一次，而语义回退的告警也只报一次，避免每个 + // Retrieve 都写日志。 + seedMu sync.Mutex + seeded bool + fallbackWarned bool } // RetrieverOption 定制由 NewRetriever 构建的 Retriever。 @@ -78,19 +88,36 @@ type RetrieverOption func(*Retriever) // 为 nil（默认）时，检索退化为仅关键词匹配 // 加衰减。若同时经 WithVectorIndex 接入了向量索引，则 // 索引优先：它是持久化语义路径（design §3 -// 可选增强），此时 WithEmbedder 改为即时计算相似度。 +// 可选增强）。索引没有作答时（索引空、无候选命中或索引报错） +// 语义分回退到此内存路径，而绝不因「接了索引」就交出一片 +// 空语义分。 func WithEmbedder(e embedding.Embedder) RetrieverOption { return func(r *Retriever) { r.embedder = e } } +// WithRetrieverLogger 设置检索器的 logger，使语义路径的降级 +// （索引报错、回退内存余弦）可被发现（AGENTS §10 日志必须标来源）。 +// 未设置时默认用该 store 的 logger（来源 "memory"），因此没有 +// 额外接线的调用方也不会静默降级。 +func WithRetrieverLogger(log *slog.Logger) RetrieverOption { + return func(r *Retriever) { + r.log = logger.WithSource(log, "memory/retriever") + } +} + // WithVectorIndex 把持久化的 vector.Index 接入检索。设置后， // 语义相关性由 idx.Search 在已存向量上作答，而非 // 内存内向量化，因此配置的 pgvector/SQLite 向量存储被 // 真正启用（design §3「真正启用向量检索」）。为 nil（默认）时， -// 除非设置了 WithEmbedder，否则语义路径跳过。调用方负责 -// 让索引与已存条目保持同步（写入时 embed + add）。 +// 除非设置了 WithEmbedder，否则语义路径跳过。 +// +// 它同时把索引接到该 store 的**写入路径**（见 Store.indexItem），使 +// 「写入时 embed + add」不再是调用方可能漏做的一步：条目写进 +// memory_items 的同时也写进索引。索引中只有全局向量，检索侧 +// 只认本次候选集内的命中，并对空索引做一次性回填（见 +// Retriever.ensureIndexSeeded）。 func WithVectorIndex(idx vector.Index) RetrieverOption { return func(r *Retriever) { r.vindex = idx @@ -138,19 +165,39 @@ func WithRecencyHalfLife(days float64) RetrieverOption { // NewRetriever 在 store 之上构建 Retriever。nil store 非法，会在 // 调用时 panic；调用方应传入非 nil 的 store。 +// +// 接入向量索引（WithVectorIndex）时，它随即将该索引接到 store 的 +// 写入路径，使此后每条落库的记忆都被向量化并写入索引 —— 索引与 +// 条目因此不会只同步一半（审计 P0-3：装配了索引却没有任何 +// Index.Add 调用，语义检索恒空）。logger 未显式给出时沿用 store 的 +// logger，使降级可见。 func NewRetriever(store *Store, opts ...RetrieverOption) *Retriever { r := &Retriever{ store: store, keywordWeight: defaultKeywordWeight, semanticWeight: defaultSemanticWeight, recencyHalfLife: time.Duration(defaultRecencyHalfLifeDays * float64(hoursPerDay*time.Hour)), + log: retrieverLog(store), } for _, opt := range opts { opt(r) } + if store != nil { + store.attachVectorIndex(r.vindex) + } return r } +// retrieverLog 返回检索器默认使用的 logger：store 自带的那一个 +// （NewStore 已把它标为来源 "memory"，且绝不为 nil），store 缺失时 +// 用一个不回显的丢弃 logger，避免检索路径出现无来源裸日志。 +func retrieverLog(store *Store) *slog.Logger { + if store == nil { + return logger.WithSource(discardLogger(nil), "memory") + } + return store.log +} + // Retrieve 为 personaID 返回最匹配 query 的最多 topK 条，按 // 相关性（关键词 + 可选语义）加权衰减（decay_weight × // 时间衰减）排序。候选集是 design/03 §3/§4 的可见域并集：该角色自己的条目 @@ -281,14 +328,24 @@ func (r *Retriever) keywordScores(items []MemoryItem, query string) map[string]f // semanticScores 计算每个条目对 query 的语义相关性。接入了向量 // 索引（WithVectorIndex）时，它经 idx.Search 从持久化索引 -// 作答 —— 真正的向量路径（design §3）。否则，在配置了 embedder -// 时，它在内存内对 query 与候选做向量化并使用夹取的余弦 -// 相似度。两者都没有时返回空 map，因此语义相关性 +// 作答 —— 真正的向量路径（design §3）。索引没有给出任何候选命中时 +// （索引空、索引报错或 query 为空），它**回退**到内存余弦：配置了 +// 增强却仍拿到一片 0 分，等于增强既无效又遮住了可用的路径 +// （审计 P0-3；design/03 §3「真正启用向量检索」、design/04 §7）。 +// 两条路径都没有时返回空 map，因此语义相关性 // 不贡献任何分数。 func (r *Retriever) semanticScores(ctx context.Context, items []MemoryItem, query string) map[string]float64 { scores := make(map[string]float64, len(items)) if r.vindex != nil { - return r.vectorIndexScores(ctx, items, query) + // 持久索引为空时先做一次性回填（每进程最多一次），使既有 + // 条目也进入语义路径，而不是只有新写的条目可被检索。 + r.ensureIndexSeeded(ctx) + if indexed := r.vectorIndexScores(ctx, items, query); len(indexed) > 0 { + return indexed + } + if r.embedder != nil { + r.warnSemanticFallback() + } } if r.embedder == nil { return scores @@ -300,10 +357,12 @@ func (r *Retriever) semanticScores(ctx context.Context, items []MemoryItem, quer } vectors, err := r.embedder.Embed(ctx, texts) if err != nil { + r.log.Warn("memory: semantic embed of candidates failed, semantic scores skipped", "error", err) return scores } queryVectors, err := r.embedder.Embed(ctx, []string{query}) if err != nil || len(queryVectors) == 0 { + r.log.Warn("memory: semantic embed of query failed, semantic scores skipped", "error", err) return scores } qvec := queryVectors[0] @@ -321,9 +380,13 @@ func (r *Retriever) semanticScores(ctx context.Context, items []MemoryItem, quer // vectorIndexScores 从持久化向量索引作答语义相关性： // 它用索引对 query 做向量化，检索最相似的若干 id，并 -// 返回它们的余弦相似度，以最高命中为基准归一化到 [0,1]。由于 -// idx.Search 在全部已存 id 上排名，不在索引中的条目没有 -// 语义得分。query 为空或索引报错时退化为空 map。 +// 返回它们的余弦相似度，以候选命中的最高分为基准归一化到 [0,1]。 +// +// 索引是**全局**的（承载所有角色/作用域的向量），而本次检索的候选 +// 只是其中一个子集：不属于候选的命中不参与排名，也不得充当 +// 归一化基准（也不得因占满 topK 名额而把候选挤出命中）。因此 +// 这里只保留候选集内的命中。query 为空、索引报错或候选内 +// 无命中时退化为空 map —— 调用方据此回退内存余弦。 func (r *Retriever) vectorIndexScores(ctx context.Context, items []MemoryItem, query string) map[string]float64 { scores := make(map[string]float64, len(items)) if strings.TrimSpace(query) == "" { @@ -331,14 +394,35 @@ func (r *Retriever) vectorIndexScores(ctx context.Context, items []MemoryItem, q } qvecs, err := r.vindex.Embed(ctx, []string{query}) if err != nil || len(qvecs) == 0 { + r.log.Warn("memory: vector index query embed failed, falling back to in-memory semantic", + "error", err) return scores } hits, err := r.vindex.Search(ctx, qvecs[0], len(items)) - if err != nil || len(hits) == 0 { + if err != nil { + r.log.Warn("memory: vector index search failed, falling back to in-memory semantic", + "error", err) return scores } - max := hits[0].Score + candidates := make(map[string]struct{}, len(items)) + for _, it := range items { + candidates[it.ID] = struct{}{} + } + matched := make([]vector.Hit, 0, len(hits)) + max := 0.0 for _, h := range hits { + if _, ok := candidates[h.ID]; !ok { + continue + } + matched = append(matched, h) + if h.Score > max { + max = h.Score + } + } + if len(matched) == 0 { + return scores + } + for _, h := range matched { if max > 0 { scores[h.ID] = h.Score / max } else { @@ -371,24 +455,6 @@ func keywordText(it MemoryItem) string { return it.Content + " " + strings.Join(it.Tags, " ") } -// cosineSim 计算两个向量的余弦相似度，由调用方夹到 [0,1]。 -// 长度为零或不一致的向量得 0。 -func cosineSim(a, b embedding.Vector) float64 { - if len(a) == 0 || len(b) == 0 || len(a) != len(b) { - return 0 - } - var dot, na, nb float64 - for i := range a { - dot += float64(a[i]) * float64(b[i]) - na += float64(a[i]) * float64(a[i]) - nb += float64(b[i]) * float64(b[i]) - } - if na == 0 || nb == 0 { - return 0 - } - return dot / (math.Sqrt(na) * math.Sqrt(nb)) -} - // Inject 为 personaID 检索最多 topK 条相关记忆，并把它们格式化为 // 紧凑的上下文块，可供注入提示词。没有检索到任何内容时 // 返回空字符串，因此没有相关记忆时调用方可以忽略 diff --git a/internal/runtime/memory/retrieve_index.go b/internal/runtime/memory/retrieve_index.go new file mode 100644 index 0000000..6c29c14 --- /dev/null +++ b/internal/runtime/memory/retrieve_index.go @@ -0,0 +1,110 @@ +package memory + +import ( + "context" +) + +// 本文件是「持久语义索引的一次性回填」：让既有条目也进入语义路径。 +// +// 写入侧（store_vector.go）保证**此后**每条记忆都被写进索引，但进程 +// 启动前就已落库的 memory_items 没有对应向量。若不回填，语义路径 +// 只覆盖新写入的少数条目，旧条目在索引里查不到 —— 对「配置了增强 +// 就必须真生效」（AGENTS §9）而言仍是半个空壳。 +// +// 回填触发条件极窄：每进程首次需要语义打分时探测一次索引，**只在索 +// 引确实为空**（一条向量都没有）时回填全部条目，之后永不重复。因此 +// 它不会在正常运行的每次检索上花代价，也不会重复嵌入已入索引的向量。 + +// indexProbeText 是探测「索引是否为空」用的固定文本：它的向量内容 +// 无关紧要（只想知道索引里有没有任何向量），取常量以保证确定性。 +const indexProbeText = "probe" + +// indexBackfillBatch 是单次 Embed 调用喂给索引的条数上限，避免把 +// 整库文本一次性交给远端 embedding API。 +const indexBackfillBatch = 64 + +// maxBackfillItems 是单次回填处理的最大条目数。它给「一次启动就把 +// 十万条历史记忆全量向量化」这种代价封顶；超出部分只记日志，不 +// 阻塞检索（Store.List 最新在前，因此被截掉的是最旧的一批）。 +const maxBackfillItems = 512 + +// ensureIndexSeeded 每进程最多执行一次：探测持久索引是否为空，为空 +// 则把已落库的条目回填进去。未接入索引时是空操作；探测或嵌入失败 +// 只记日志且不再重试（重试只会把同一条故障日志刷满每一次检索）。 +func (r *Retriever) ensureIndexSeeded(ctx context.Context) { + if r.vindex == nil { + return + } + r.seedMu.Lock() + defer r.seedMu.Unlock() + if r.seeded { + return + } + r.seeded = true + + probe, err := r.vindex.Embed(ctx, []string{indexProbeText}) + if err != nil || len(probe) == 0 { + r.log.Warn("memory: vector index probe failed, semantic index not seeded", "error", err) + return + } + hits, err := r.vindex.Search(ctx, probe[0], 1) + if err != nil { + r.log.Warn("memory: vector index probe search failed, semantic index not seeded", "error", err) + return + } + if len(hits) > 0 { + // 索引里已有向量：它由写入侧或上一次运行填过，无需回填。 + return + } + r.backfillIndex(ctx) +} + +// backfillIndex 把已落库的条目向量化并写入索引。它分批 Embed，逐条 +// Add；单条失败只记日志（其余条目继续），因此一次失败的 Add 不会让 +// 整个回填白跑。按 maxBackfillItems 截断时记一条日志说明被跳过 +// 的量，避免「配置了增强却只覆盖一部分」无人知晓。 +func (r *Retriever) backfillIndex(ctx context.Context) { + items, err := r.store.List(ctx) + if err != nil { + r.log.Warn("memory: list items for vector backfill failed", "error", err) + return + } + if len(items) == 0 { + return + } + total := len(items) + if total > maxBackfillItems { + r.log.Warn("memory: vector backfill truncated, oldest items left unindexed", + "total", total, "limit", maxBackfillItems) + items = items[:maxBackfillItems] + } + + indexed, failed := 0, 0 + for start := 0; start < len(items); start += indexBackfillBatch { + end := min(start+indexBackfillBatch, len(items)) + batch := items[start:end] + texts := make([]string, len(batch)) + for i, it := range batch { + texts[i] = it.Content + } + vecs, err := r.vindex.Embed(ctx, texts) + if err != nil { + r.log.Warn("memory: vector backfill embed failed, seeding stopped", + "indexed", indexed, "error", err) + return + } + for i, v := range vecs { + if i >= len(batch) { + break + } + if err := r.vindex.Add(ctx, batch[i].ID, v); err != nil { + failed++ + r.log.Warn("memory: vector backfill add failed", "id", batch[i].ID, "error", err) + continue + } + indexed++ + } + } + r.log.Info("memory: vector index backfilled from persisted items", + "indexed", indexed, "failed", failed, "total", total) +} diff --git a/internal/runtime/memory/retrieve_semantic.go b/internal/runtime/memory/retrieve_semantic.go new file mode 100644 index 0000000..4da51ab --- /dev/null +++ b/internal/runtime/memory/retrieve_semantic.go @@ -0,0 +1,44 @@ +package memory + +import ( + "math" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding" +) + +// 本文件收拢语义打分的辅助件：内存余弦相似度与「索引没答上、改走 +// 内存回退」的一次性告警。它们的用量小但分散，独立成文件是为了让 +// retrieve.go 保持可读（文件 ≤500 行，AGENTS §10）。 + +// cosineSim 计算两个向量的余弦相似度，由调用方夹到 [0,1]。 +// 长度为零或不一致的向量得 0。 +func cosineSim(a, b embedding.Vector) float64 { + if len(a) == 0 || len(b) == 0 || len(a) != len(b) { + return 0 + } + var dot, na, nb float64 + for i := range a { + dot += float64(a[i]) * float64(b[i]) + na += float64(a[i]) * float64(a[i]) + nb += float64(b[i]) * float64(b[i]) + } + if na == 0 || nb == 0 { + return 0 + } + return dot / (math.Sqrt(na) * math.Sqrt(nb)) +} + +// warnSemanticFallback 报告一次「接入了向量索引但在候选集内无命中， +// 已改走内存余弦」。它每个检索器只报一次：该状态在一次运行里 +// 基本不变（索引尚未写入或 embedder 维度不匹配），逐次 Retrieve +// 重复告警只会淹没有用日志。索引报错本身由 vectorIndexScores 以 +// Warn 记录，这里用 Debug 区分「已知会回退」与「真的出错了」。 +func (r *Retriever) warnSemanticFallback() { + r.seedMu.Lock() + defer r.seedMu.Unlock() + if r.fallbackWarned { + return + } + r.fallbackWarned = true + r.log.Debug("memory: vector index had no candidate hit, using in-memory cosine fallback") +} diff --git a/internal/runtime/memory/store.go b/internal/runtime/memory/store.go index b19a7b0..5ace4cd 100644 --- a/internal/runtime/memory/store.go +++ b/internal/runtime/memory/store.go @@ -38,9 +38,11 @@ import ( "io" "log/slog" "strings" + "sync" "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory/vector" ) // Scope 分类一条长期记忆归属于谁。它与 Kind 正交， @@ -139,12 +141,18 @@ const ( // // 存储自己拥有 schema：NewStore 幂等地创建 memory_items 表及其 // 索引。并发交由 *sql.DB，按约定它可安全 -// 并发使用，因此 Store 不再持锁。每个查询都 +// 并发使用，因此除语义索引的接入外，Store 不再持锁。每个查询都 // 参数化（?）并显式列出列（无 SELECT *），读取带 // context.Cancel。 type Store struct { db *sql.DB log *slog.Logger + + // vecMu 保护 vec，使持久语义索引能在装配后并发接入，而写入 + // 路径每写一条记忆就 embed + add 到它（见 indexItem）。vec 为 + // nil 表示没有配置持久索引，检索据此回退到内存余弦。 + vecMu sync.RWMutex + vec vector.Index } // rowScanner 同时被 *sql.Row 与 *sql.Rows 满足，因此单个 scanItem @@ -186,6 +194,13 @@ func NewStore(ctx context.Context, db *sql.DB, log *slog.Logger) (*Store, error) // Add 持久化 item，分配 id 与缺失的时间戳，并把 // Scope/Kind/Confidence/DecayWeight 规范化到默认值。它返回已落库的 // 条目，id 与时间戳均已填充。 +// +// 接入持久语义索引（见 attachVectorIndex）时，落库后同一条目也会 +// 被向量化并写入索引，使索引与 memory_items 同步 —— 语义检索路径因此 +// 真的有向量可查，而不是对一个恒空索引搜索（design/03 §3「真正启用 +// 向量检索」、design/04 §7）。索引写入失败只记日志：条目已经在 +// 关系表里，语义增强的失败绝不使记忆写入失败，检索侧也会回退到 +// 内存余弦（见 Retriever.semanticScores）。 func (s *Store) Add(ctx context.Context, item MemoryItem) (MemoryItem, error) { if err := item.normalize(); err != nil { return MemoryItem{}, err @@ -209,6 +224,7 @@ func (s *Store) Add(ctx context.Context, item MemoryItem) (MemoryItem, error) { if err != nil { return MemoryItem{}, fmt.Errorf("memory: insert item: %w", err) } + s.indexItem(ctx, item) return item, nil } diff --git a/internal/runtime/memory/store_vector.go b/internal/runtime/memory/store_vector.go new file mode 100644 index 0000000..66929c7 --- /dev/null +++ b/internal/runtime/memory/store_vector.go @@ -0,0 +1,68 @@ +package memory + +import ( + "context" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory/vector" +) + +// 本文件是「持久语义索引的写入侧」：把记忆条目真正喂进向量索引。 +// +// 背景（审计 P0-3）：vec_items 表与 Index.Add 早已实现，但生产代码 +// 从未把任何条目写进索引 —— 索引恒空，而检索一见 vindex 非 nil 就 +// 只查它，语义检索因此在线上恒为 0 分，还遮住了可用的内存余弦回退。 +// 这直接触碰 design/03 §3「选增强则向量检索路径真实生效」、design/04 +// §7「配置后必须真实走语义路径（非空壳）」与 AGENTS §9 的红线。 +// +// 接线方式：装配方（cmd/cyrene/wire_runtime.go）把索引交给检索器 +// （memory.WithVectorIndex），NewRetriever 随即把它接到同一个 store 的 +// 写入路径（attachVectorIndex），因此「写记忆」与「写向量」不再可能 +// 只发生一半，也无需装配方多记一步。 + +// attachVectorIndex 把持久语义索引接到该 store 的写入路径：此后每条 +// 经 Add 落库的条目都会被向量化并 upsert 进索引。nil 索引是空操作 +// （保持「未配置索引」的语义）。它可被并发调用，last-writer-wins。 +func (s *Store) attachVectorIndex(idx vector.Index) { + if s == nil || idx == nil { + return + } + s.vecMu.Lock() + defer s.vecMu.Unlock() + s.vec = idx +} + +// vectorIndex 返回当前接入的持久语义索引，未接入时为 nil。 +func (s *Store) vectorIndex() vector.Index { + if s == nil { + return nil + } + s.vecMu.RLock() + defer s.vecMu.RUnlock() + return s.vec +} + +// indexItem 把已落库的 item 向量化并写入持久索引（写入侧闭环）。 +// 未接入索引、索引报错或向量数不符时都只记日志：条目本身已经 +// 在关系表里，语义增强的失败绝不使写入失败，检索侧另有内存余弦 +// 回退。索引文本取 item.Content —— 与内存语义回退所用的文本一致， +// 使「索引作答」与「回退作答」在同一份输入上可比。 +func (s *Store) indexItem(ctx context.Context, item MemoryItem) { + idx := s.vectorIndex() + if idx == nil { + return + } + vecs, err := idx.Embed(ctx, []string{item.Content}) + if err != nil { + s.log.Warn("memory: vector embed for index failed, item left unindexed", + "id", item.ID, "error", err) + return + } + if len(vecs) == 0 { + s.log.Warn("memory: vector embed returned no vector, item left unindexed", "id", item.ID) + return + } + if err := idx.Add(ctx, item.ID, vecs[0]); err != nil { + s.log.Warn("memory: vector index add failed, item left unindexed", + "id", item.ID, "error", err) + } +} diff --git a/internal/runtime/memory/vector_sync_test.go b/internal/runtime/memory/vector_sync_test.go new file mode 100644 index 0000000..3e8334e --- /dev/null +++ b/internal/runtime/memory/vector_sync_test.go @@ -0,0 +1,220 @@ +package memory + +import ( + "context" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/config" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory/vector" +) + +// countingIndex 包一层真实 vector.Index，数出写入侧与检索侧各用了几次： +// 它是「持久向量索引真的被喂了数据 / 真的被查询了」的观测点， +// 而不是替换掉真实后端的桩 —— 命中与得分仍由底层 SQLite 索引算出。 +type countingIndex struct { + vector.Index + adds int + searches int +} + +func (c *countingIndex) Add(ctx context.Context, id string, v vector.Vector) error { + c.adds++ + return c.Index.Add(ctx, id, v) +} + +func (c *countingIndex) Search(ctx context.Context, q vector.Vector, topK int) ([]vector.Hit, error) { + c.searches++ + return c.Index.Search(ctx, q, topK) +} + +// newTestVectorIndex 用**生产的构造路径** vector.New 建一个真实向量索引 +// （默认的零额外依赖 SQLite 后端 + n-gram embedder），只是把库路径换成 +// 隔离的内存库，使测试与线上装配走同一条代码路径。 +func newTestVectorIndex(t *testing.T, dim int) vector.Index { + t.Helper() + idx, err := vector.New( + config.DBConfig{SQLitePath: ":memory:"}, + embedding.NewNGram(dim), + testLogger(), + ) + if err != nil { + t.Fatalf("vector.New: %v", err) + } + t.Cleanup(func() { _ = idx.Close() }) + return idx +} + +// TestStore_Add_WritesVectorIndex_写入侧真生效 是审计 P0-3 的主断言： +// 检索器接入索引后，**每条经 Store.Add 落库的记忆都被向量化并写进 +// 索引**，因此对索引自身检索能命中该条目、且自相似度接近 1 —— +// 索引不再是恒空的空壳（design/03 §3「真正启用向量检索」）。 +func TestStore_Add_WritesVectorIndex_写入侧真生效(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + idx := &countingIndex{Index: newTestVectorIndex(t, 64)} + + // 装配顺序与 cmd/cyrene 一致：先写 store，再经 NewRetriever 把索引 + // 接到同一个 store 的写入路径。 + r := NewRetriever(store, WithEmbedder(embedding.NewNGram(64)), WithVectorIndex(idx)) + if r == nil { + t.Fatal("NewRetriever returned nil") + } + + stored := addItem(t, store, MemoryItem{ + PersonaID: "persona.a", Content: "她喜欢早上喝手冲咖啡", + Scope: ScopePersona, Kind: KindRelationship, + }) + if idx.adds == 0 { + t.Fatal("Store.Add did not write any vector: the persistent index is still an empty shell") + } + + // 索引里确实有这条向量：用同一内容的向量去检索，必须命中它本身。 + qvecs, err := idx.Embed(ctx, []string{stored.Content}) + if err != nil { + t.Fatalf("index embed: %v", err) + } + hits, err := idx.Search(ctx, qvecs[0], 5) + if err != nil { + t.Fatalf("index search: %v", err) + } + if len(hits) == 0 { + t.Fatal("index has no vector for the persisted item") + } + if hits[0].ID != stored.ID { + t.Fatalf("top index hit = %q, want the persisted item %q", hits[0].ID, stored.ID) + } + if hits[0].Score < 0.99 { + t.Fatalf("self-similarity = %v, want ~1 (the stored vector is not the item's own)", hits[0].Score) + } +} + +// TestRetriever_Semantic_UsesIndexThenFallsBack_语义路径先索引后回退 +// 覆盖 P0-3 的第二半：索引答不出候选命中时，语义分回退到内存余弦， +// 而不再因为「装了索引」就交出一片 0 分。 +// +// 三路对照都在同一个候选集上做：① 索引命中候选 → 只有索引给分的条目 +// 有语义分（索引作答）；② 索引只命中候选之外的 id（全局索引里别的 +// 角色的向量）→ 内存余弦接手（装上 embedder 时冷门条目仍有正分）； +// ③ 同样的空候选命中但没有 embedder → 没有任何语义来源，分值为空。 +func TestRetriever_Semantic_UsesIndexThenFallsBack_语义路径先索引后回退(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + // 内容为英文、查询为德语（无共同词元），关键词路径因此贡献 0： + // 语义分若有，只可能来自向量路径。 + item := addItem(t, store, MemoryItem{ + PersonaID: "persona.a", Content: "she is a morning person", + Scope: ScopePersona, Kind: KindRelationship, + }) + const query = "Kaffeegetränk" + + items, err := store.ListByPersona(ctx, "persona.a") + if err != nil { + t.Fatalf("ListByPersona: %v", err) + } + + // ① 索引认得候选 → 由索引作答。 + hitIdx := &countingIndex{Index: &mockVectorIndex{ + dim: 4, + searchFn: func(_ context.Context, _ vector.Vector, _ int) ([]vector.Hit, error) { + return []vector.Hit{{ID: item.ID, Score: 0.5}}, nil + }, + }} + fromIndex := NewRetriever(store, WithEmbedder(fixedEmbedder{dim: 4}), WithVectorIndex(hitIdx)) + got := fromIndex.semanticScores(ctx, items, query) + if len(got) != 1 || got[item.ID] <= 0 { + t.Fatalf("semantic scores from the index = %+v, want exactly the indexed item %q", got, item.ID) + } + if hitIdx.searches == 0 { + t.Fatal("the vector index was never queried: the semantic path is not wired") + } + + // ② 索引只命中候选之外的 id → 候选集内无命中 → 回退内存余弦。 + foreignIdx := &countingIndex{Index: &mockVectorIndex{ + dim: 4, + searchFn: func(_ context.Context, _ vector.Vector, _ int) ([]vector.Hit, error) { + return []vector.Hit{{ID: "someone-else-item", Score: 0.9}}, nil + }, + }} + withFallback := NewRetriever(store, WithEmbedder(fixedEmbedder{dim: 4}), WithVectorIndex(foreignIdx)) + fallback := withFallback.semanticScores(ctx, items, query) + if fallback[item.ID] <= 0 { + t.Fatalf("semantic scores = %+v, want a positive score for %q from the in-memory cosine fallback", + fallback, item.ID) + } + + // ③ 同样的空候选命中，但没有 embedder 可回退 → 没有语义分。 + bare := NewRetriever(store, WithVectorIndex(foreignIdx)) + if got := bare.semanticScores(ctx, items, query); len(got) != 0 { + t.Fatalf("semantic scores without an embedder = %+v, want empty", got) + } +} + +// TestRetriever_BackfillsPersistedItems_索引空时一次性回填 证明既有 +// 条目也进入语义路径：写入发生在接入索引**之前**（进程重启后读到的 +// 历史记忆就是这种状态），首次语义检索发现索引为空，于是把已落库 +// 的条目一次性向量化并写进索引 —— 之后它们真的能被索引检索到。 +func TestRetriever_BackfillsPersistedItems_索引空时一次性回填(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + + // 索引接入之前写入的两条记忆（模拟上一次运行留下的历史条目）。 + first := addItem(t, store, MemoryItem{ + PersonaID: "persona.a", Content: "她喜欢早上喝手冲咖啡", + Scope: ScopePersona, Kind: KindRelationship, + }) + addItem(t, store, MemoryItem{ + PersonaID: "persona.a", Content: "她周末去山里徒步", + Scope: ScopePersona, Kind: KindRelationship, + }) + + idx := &countingIndex{Index: newTestVectorIndex(t, 64)} + r := NewRetriever(store, WithEmbedder(embedding.NewNGram(64)), WithVectorIndex(idx)) + if _, err := r.Retrieve(ctx, "persona.a", "手冲咖啡", 10); err != nil { + t.Fatalf("Retrieve: %v", err) + } + if idx.adds < 2 { + t.Fatalf("backfill called Add %d times, want >= 2 (both persisted items)", idx.adds) + } + + qvecs, err := idx.Embed(ctx, []string{first.Content}) + if err != nil { + t.Fatalf("index embed: %v", err) + } + hits, err := idx.Search(ctx, qvecs[0], 5) + if err != nil { + t.Fatalf("index search: %v", err) + } + if len(hits) == 0 || hits[0].ID != first.ID { + t.Fatalf("index hits = %+v, want the backfilled item %q first", hits, first.ID) + } + + // 回填只做一次：再检索一轮不再新增 Add。 + before := idx.adds + if _, err := r.Retrieve(ctx, "persona.a", "徒步", 10); err != nil { + t.Fatalf("second Retrieve: %v", err) + } + if idx.adds != before { + t.Fatalf("backfill ran again: adds %d -> %d, want no further embedding", before, idx.adds) + } +} + +// fixedEmbedder 是「任何文本都得到同一个向量」的确定性 embedder：它使 +// 内存余弦路径对任意 query/内容都给出正相似度，从而让测试能判定 +// 「语义分是否由内存回退贡献」。 +type fixedEmbedder struct{ dim int } + +func (f fixedEmbedder) Embed(_ context.Context, texts []string) ([]embedding.Vector, error) { + out := make([]embedding.Vector, len(texts)) + for i := range texts { + v := make(embedding.Vector, f.dim) + v[0] = 1 + out[i] = v + } + return out, nil +} + +func (f fixedEmbedder) Dim() int { return f.dim } diff --git a/internal/runtime/memory/working.go b/internal/runtime/memory/working.go index 0256266..3df61c9 100644 --- a/internal/runtime/memory/working.go +++ b/internal/runtime/memory/working.go @@ -18,8 +18,77 @@ const ( // summaryLastTextMax 是朴素摘要中所含末条消息文本 // 的最大 rune 数，用于保持自摘要紧凑。 summaryLastTextMax = 80 + + // maxWorkingSessions 是工作记忆同时保留的**会话**窗口数上限。 + // 进程内只有一个 WorkingMemory（loop 的每会话滑动窗口），而会话数 + // 随渠道消息无界增长，因此超出上限时淘汰最久未用的会话窗口， + // 使内存有界（design/03 §0 层①）。 + maxWorkingSessions = 64 ) +// window 是一条会话窗口的可变状态：消息滑动窗口 + 摘要。 +// +// 它不持锁：所有方法都要求调用方已持有 WorkingMemory.mu +// （读锁或写锁），这正是窗口与摘要共用同一临界区的原因。 +type window struct { + messages []model.Message + summary string +} + +// add 把 msg 追加到窗口，并在窗口超出 max 时淘汰最旧的消息。 +func (w *window) add(msg model.Message, max int) { + w.messages = append(w.messages, msg) + w.trim(max) +} + +// replace 用 msgs 替换窗口内容，只保留最后 max 条（msgs 是该会话 +// 当前对话的完整快照时，窗口即「最近 max 条」的滑窗）。传入的消息 +// 逐条深拷贝，因此调用方后续修改自己的切片不会影响窗口。 +func (w *window) replace(msgs []model.Message, max int) { + out := make([]model.Message, 0, len(msgs)) + for _, m := range msgs { + out = append(out, cloneMessage(m)) + } + w.messages = out + w.trim(max) +} + +// trim 只保留窗口最后 max 条消息。 +func (w *window) trim(max int) { + if max <= 0 || len(w.messages) <= max { + return + } + overflow := len(w.messages) - max + copy(w.messages, w.messages[overflow:]) + w.messages = w.messages[:max] +} + +// snapshot 返回窗口消息的深拷贝，最旧的在前。返回的切片与每条 +// 消息的 content 部分都与内部状态脱离，因此调用方可以随意修改。 +func (w *window) snapshot() []model.Message { + out := make([]model.Message, len(w.messages)) + for i, m := range w.messages { + out[i] = cloneMessage(m) + } + return out +} + +// summarize 在已设置摘要时返回存储的摘要，否则返回由当前 +// 窗口推导的朴素结构化摘要。它从不修改状态。 +func (w *window) summarize() string { + if w.summary != "" { + return w.summary + } + return naiveSummary(w.messages) +} + +// sessionWindow 是一条会话的窗口及其最近使用序号，序号供 +// 超出上限时淘汰最久未用的会话（见 maxWorkingSessions）。 +type sessionWindow struct { + win window + lastUsed uint64 +} + // WorkingMemory 是每会话的短期记忆：当前对话消息的 // 滑动窗口加一份摘要（design §0 layer ①）。它是 // 代理推理所依据的即时上下文。 @@ -29,13 +98,25 @@ const ( // 朴素结构化摘要。LLM 生成的摘要推迟到 M4；管线 // 已就位，摘要器可调用 SetSummary。 // -// WorkingMemory 并发安全；Add/SetSummary 以写锁串行， -// Window/Summarize 在读锁下读取。 +// 两种使用方式并存： +// +// - 默认窗口（Add/Window/SetSummary/Summarize）：无会话键的 +// 单窗口路径，保持该类型最初的语义。 +// - 会话窗口（RecordSession/SessionWindow/SummarizeSession/ +// SetSessionSummary）：**按会话 id 隔离**的窗口，循环在回合 +// 落定时写入、在组装上下文时读取。进程内只有一个 +// WorkingMemory 实例（loop.Options.WorkingMemory），因此没有 +// 会话键隔离时，群聊/陌生人的窗口会混进主人的工作记忆 —— +// 这是隔离红线（design/03 §5），会话路径即为此而设。 +// +// WorkingMemory 并发安全；写路径以写锁串行， +// 读路径在读锁下读取。 type WorkingMemory struct { - mu sync.RWMutex - window []model.Message - max int - summary string + mu sync.RWMutex + def window + sessions map[string]*sessionWindow + max int + seq uint64 } // NewWorkingMemory 构建一个工作记忆，滑动窗口最多 @@ -47,32 +128,136 @@ func NewWorkingMemory(maxMessages int) *WorkingMemory { return &WorkingMemory{max: maxMessages} } -// Add 把 msg 追加到窗口，并在窗口超出上限时 -// 淘汰最旧的消息。 +// Add 把 msg 追加到**默认**窗口，并在窗口超出上限时 +// 淘汰最旧的消息。它写入的是无会话键的单窗口；按会话隔离的 +// 写入走 RecordSession。 func (w *WorkingMemory) Add(msg model.Message) { w.mu.Lock() defer w.mu.Unlock() - w.window = append(w.window, msg) - if len(w.window) > w.max { - overflow := len(w.window) - w.max - copy(w.window, w.window[overflow:]) - w.window = w.window[:w.max] - } + w.def.add(msg, w.max) } -// Window 返回当前消息窗口的深拷贝，最旧的在前。 +// Window 返回默认窗口当前消息的深拷贝，最旧的在前。 // 返回的切片与每条消息的 content 部分都与内部 // 状态脱离，因此调用方可以随意修改。 func (w *WorkingMemory) Window() []model.Message { w.mu.RLock() defer w.mu.RUnlock() - out := make([]model.Message, len(w.window)) - for i, m := range w.window { - out[i] = cloneMessage(m) + return w.def.snapshot() +} + +// SetSummary 记录一条显式摘要（例如 M4 中由 LLM 产出）到默认窗口。 +func (w *WorkingMemory) SetSummary(s string) { + w.mu.Lock() + defer w.mu.Unlock() + + w.def.summary = s +} + +// Summarize 在已设置摘要时返回存储的摘要，否则返回由默认窗口 +// 推导的朴素结构化摘要。它从不修改状态，因此 +// 重复调用返回与调用时窗口一致的值。 +func (w *WorkingMemory) Summarize() string { + w.mu.RLock() + defer w.mu.RUnlock() + + return w.def.summarize() +} + +// RecordSession 把会话 sessionID 的窗口置为 msgs 的最后 +// maxMessages 条（msgs 是该会话当前对话的完整快照，因此结果 +// 即「最近 N 条」的滑窗）。空会话 id 是空操作 —— 无键的写入无从 +// 隔离，绝不落到默认窗口，以免污染无会话键的读点。 +// +// 循环在回合落定时以该 turn 的完整对话调用它：会话恢复时，checkpoint +// 还原出的历史随同一个 turn 一起进来，窗口因此被回填，而不是 +// 从空白重启（design §1 durable execution）。 +func (w *WorkingMemory) RecordSession(sessionID string, msgs []model.Message) { + if strings.TrimSpace(sessionID) == "" { + return + } + w.mu.Lock() + defer w.mu.Unlock() + + sw := w.sessionLocked(sessionID) + sw.win.replace(msgs, w.max) +} + +// SessionWindow 返回会话 sessionID 窗口的深拷贝（最旧的在前）。 +// 该会话没有窗口时返回空切片。 +func (w *WorkingMemory) SessionWindow(sessionID string) []model.Message { + w.mu.RLock() + defer w.mu.RUnlock() + + sw, ok := w.sessions[sessionID] + if !ok { + return []model.Message{} + } + return sw.win.snapshot() +} + +// SetSessionSummary 为会话 sessionID 记录一条显式摘要（LLM 摘要器 +// 的落点，design §3 §6）。空会话 id 是空操作。 +func (w *WorkingMemory) SetSessionSummary(sessionID, s string) { + if strings.TrimSpace(sessionID) == "" { + return + } + w.mu.Lock() + defer w.mu.Unlock() + + w.sessionLocked(sessionID).win.summary = s +} + +// SummarizeSession 返回会话 sessionID 的摘要：已由 SetSessionSummary +// 显式设置时返回它，否则返回该会话窗口的朴素结构化摘要。 +// 该会话没有窗口（或窗口为空）时返回 ""，因此调用方可以据此 +// 跳过前导块。它从不修改状态。 +func (w *WorkingMemory) SummarizeSession(sessionID string) string { + w.mu.RLock() + defer w.mu.RUnlock() + + sw, ok := w.sessions[sessionID] + if !ok { + return "" + } + return sw.win.summarize() +} + +// sessionLocked 返回会话 sessionID 的窗口（必要时新建），并把它 +// 标为最近使用；新建会使保留的会话数超出上限时，先淘汰最久 +// 未用的那个，使会话窗口数有界。它要求持有 w.mu 的**写锁**。 +func (w *WorkingMemory) sessionLocked(sessionID string) *sessionWindow { + if w.sessions == nil { + w.sessions = make(map[string]*sessionWindow) + } + w.seq++ + sw, ok := w.sessions[sessionID] + if !ok { + if len(w.sessions) >= maxWorkingSessions { + w.evictOldestLocked() + } + sw = &sessionWindow{} + w.sessions[sessionID] = sw + } + sw.lastUsed = w.seq + return sw +} + +// evictOldestLocked 删除最久未使用的会话窗口。它要求持有 w.mu 的 +// 写锁；空 map 时是空操作。 +func (w *WorkingMemory) evictOldestLocked() { + oldestKey := "" + var oldest uint64 + for key, sw := range w.sessions { + if oldestKey == "" || sw.lastUsed < oldest { + oldestKey, oldest = key, sw.lastUsed + } + } + if oldestKey != "" { + delete(w.sessions, oldestKey) } - return out } // cloneMessage 返回 m 的副本，其 content 部分与源脱离。 @@ -85,41 +270,21 @@ func cloneMessage(m model.Message) model.Message { return m } -// SetSummary 记录一条显式摘要（例如 M4 中由 LLM 产出）。 -func (w *WorkingMemory) SetSummary(s string) { - w.mu.Lock() - defer w.mu.Unlock() - w.summary = s -} - -// Summarize 在已设置摘要时返回存储的摘要，否则返回由当前 -// 窗口推导的朴素结构化摘要。它从不修改状态，因此 -// 重复调用返回与调用时窗口一致的值。 -func (w *WorkingMemory) Summarize() string { - w.mu.RLock() - defer w.mu.RUnlock() - - if w.summary != "" { - return w.summary - } - return w.naiveSummaryLocked() -} - -// naiveSummaryLocked 构造 "{n} 条消息, {role} {count}, ...; 最近: {text}"。 -// 它要求持有 w.mu（读锁或写锁）。 -func (w *WorkingMemory) naiveSummaryLocked() string { - if len(w.window) == 0 { +// naiveSummary 构造 "{n} 条消息, {role} {count}, ...; 最近: {text}"。 +// 它是纯函数：不读取也不修改任何窗口状态。 +func naiveSummary(messages []model.Message) string { + if len(messages) == 0 { return "" } counts := make(map[model.Role]int) last := "" - for _, m := range w.window { + for _, m := range messages { counts[m.Role]++ last = messageText(m) } var b strings.Builder - b.WriteString(strconv.Itoa(len(w.window))) + b.WriteString(strconv.Itoa(len(messages))) b.WriteString(" 条消息") for _, role := range []model.Role{model.RoleSystem, model.RoleUser, model.RoleAssistant, model.RoleTool} { if c := counts[role]; c > 0 { diff --git a/internal/runtime/memory/working_session_test.go b/internal/runtime/memory/working_session_test.go new file mode 100644 index 0000000..8e2de19 --- /dev/null +++ b/internal/runtime/memory/working_session_test.go @@ -0,0 +1,107 @@ +package memory + +import ( + "strings" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// TestWorkingMemory_RecordSession_Sliding_滑窗 证明会话窗口与默认窗口一样 +// 是「最近 N 条」的滑窗，且只保留窗口内的最后 max 条。 +func TestWorkingMemory_RecordSession_Sliding_滑窗(t *testing.T) { + t.Parallel() + + w := NewWorkingMemory(3) + msgs := make([]model.Message, 0, 5) + for i := 0; i < 5; i++ { + msgs = append(msgs, model.TextMessage(model.RoleUser, string(rune('a'+i)))) + } + w.RecordSession("sess-1", msgs) + + win := w.SessionWindow("sess-1") + if len(win) != 3 { + t.Fatalf("会话窗口长度 = %d, want 3（滑窗）", len(win)) + } + for i, want := range []string{"c", "d", "e"} { + if got := win[i].Content[0].Text; got != want { + t.Errorf("窗口[%d] = %q, want %q", i, got, want) + } + } + // 摘要由该会话窗口推导，因此非空且带消息条数。 + summary := w.SummarizeSession("sess-1") + if !strings.Contains(summary, "3 条消息") { + t.Fatalf("SummarizeSession = %q, want 含 3 条消息", summary) + } + if w.Summarize() != "" { + t.Fatalf("默认窗口被会话写入污染: %q", w.Summarize()) + } + + // 返回的切片与内部状态脱离。 + win[0].Content[0].Text = "mutated" + if again := w.SessionWindow("sess-1"); again[0].Content[0].Text != "c" { + t.Fatalf("SessionWindow 返回了共享切片: %q", again[0].Content[0].Text) + } +} + +// TestWorkingMemory_SessionWindow_IsolatedPerSession_会话隔离 证明窗口 +// 按会话 id 隔离 —— 这是群聊/陌生人的工作记忆不混进主人会话的前提。 +func TestWorkingMemory_SessionWindow_IsolatedPerSession_会话隔离(t *testing.T) { + t.Parallel() + + w := NewWorkingMemory(10) + w.RecordSession("web:owner", []model.Message{model.TextMessage(model.RoleUser, "主人的话")}) + w.RecordSession("sess-group", []model.Message{model.TextMessage(model.RoleUser, "群里的话")}) + + owner := w.SessionWindow("web:owner") + if len(owner) != 1 || owner[0].Content[0].Text != "主人的话" { + t.Fatalf("owner 窗口 = %+v, want 只有主人的话（会话隔离）", owner) + } + group := w.SessionWindow("sess-group") + if len(group) != 1 || group[0].Content[0].Text != "群里的话" { + t.Fatalf("group 窗口 = %+v, want 只有群里的话", group) + } + if s := w.SummarizeSession("web:owner"); strings.Contains(s, "群里的话") { + t.Fatalf("owner 摘要混入了别的会话内容: %q", s) + } + if got := w.SessionWindow("never-seen"); len(got) != 0 { + t.Fatalf("未知会话窗口 = %+v, want 空", got) + } + if got := w.SummarizeSession("never-seen"); got != "" { + t.Fatalf("未知会话摘要 = %q, want 空", got) + } + + // 空会话 id 是空操作：无键的写入无从隔离，绝不落到默认窗口。 + w.RecordSession("", []model.Message{model.TextMessage(model.RoleUser, "无键写入")}) + if got := w.Summarize(); got != "" { + t.Fatalf("空会话 id 的写入污染了默认窗口: %q", got) + } +} + +// TestWorkingMemory_Sessions_Bounded_会话窗口有界 证明保留的会话窗口数 +// 有上限：会话数随渠道消息无界增长，最久未用的窗口被淘汰，内存不失控。 +func TestWorkingMemory_Sessions_Bounded_会话窗口有界(t *testing.T) { + t.Parallel() + + w := NewWorkingMemory(10) + for i := 0; i < maxWorkingSessions; i++ { + w.RecordSession(sessionName(i), []model.Message{model.TextMessage(model.RoleUser, "x")}) + } + // 超出上限后再写入一个新会话：最久未用的第一个会话被淘汰。 + w.RecordSession("overflow", []model.Message{model.TextMessage(model.RoleUser, "y")}) + + if got := len(w.SessionWindow(sessionName(0))); got != 0 { + t.Fatalf("最久未用的会话窗口仍在（%d 条）: 会话窗口数没有上限", got) + } + if got := len(w.SessionWindow("overflow")); got != 1 { + t.Fatalf("新会话窗口 = %d 条, want 1", got) + } + if got := len(w.SessionWindow(sessionName(maxWorkingSessions - 1))); got != 1 { + t.Fatalf("最近使用的会话窗口被误淘汰: %d 条", got) + } +} + +// sessionName 生成第 i 个会话 id。 +func sessionName(i int) string { + return "sess-" + string(rune('A'+i%26)) + string(rune('a'+i/26)) +}