AskaEth
0ce498754a
- 04.4.2/1.9 读侧：Retrieve 候选集并集 = 本人 persona 域 ∪ user 共享域 ∪ project 共享域（他角色 persona 域仍排除），群聊/陌生人取受限收窄
- 写侧：群聊/陌生人整段跳过记忆提取（可查不可写），persona 域仅属主可写
- 04.4.3：source_conversation 不再由 LLM 自由生成（prompt 不索要、候选结构不含该键），落库时无条件盖真实 {session_id,message_id}，编解码集中在 memory/source.go
- 新增跨角色种子用例（原用例即使关掉并集也会通过）+ 成对用例「同群聊上下文：读得到共享域、写不进 store」
- 同步 store.go/retrieve.go 文档注释，与 design/03 §3/§4 口径一致
324 lines
13 KiB
Go
324 lines
13 KiB
Go
package loop
|
||
|
||
import (
|
||
"context"
|
||
"strings"
|
||
"time"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||
)
|
||
|
||
// memoryExtractTimeout 限制单次后台提取。它只管辖
|
||
// 脱离出来的提取 context，因此挂起的模型调用不会让一个 goroutine
|
||
// 泄露到进程生命周期结束。
|
||
const memoryExtractTimeout = 60 * time.Second
|
||
|
||
// asyncExtractor 是循环脱离到后台 goroutine 的提取器能力：
|
||
// memory.LLMExtractor 经 ExtractAsync 实现它（design/03 §1
|
||
// LLM 异步自动提取），返回一个恰好投递一个结果的 channel。
|
||
// 确定性提取器（memory.ManualExtractor）不实现它，留在
|
||
// 同步路径上。
|
||
type asyncExtractor interface {
|
||
ExtractAsync(ctx context.Context, conversation []model.Message) (<-chan memory.AsyncResult, error)
|
||
}
|
||
|
||
// injectMemory 检索该角色的相关长期记忆并把它们作为前导
|
||
// 系统消息前置，使模型能把回答建立在召回出的上下文之上
|
||
// （design §6 / §2）。它经 assembleContext 每轮重新运行，因此
|
||
// 检索反映最新对话而非首轮快照。未配置
|
||
// retriever/persona、检索失败、或没找到相关内容时，
|
||
// 它原样返回输入消息。
|
||
//
|
||
// ctx 上的会话环境为检索设门：owner 全权限环境保留完整
|
||
// 召回（行为不变；候选集已并入 user/project 共享域，
|
||
// design/03 §4）；受限环境（群聊 / 陌生人，
|
||
// design/02 §3, design/11 §2, design/03 §5）是只读隔离 ——
|
||
// 只注入非敏感的共享记忆（用户习惯 + 项目），主人私有的
|
||
// 角色日常记忆绝不泄露给群聊或陌生人。
|
||
func (l *Loop) injectMemory(ctx context.Context, messages []model.Message) []model.Message {
|
||
block := l.memoryBlock(ctx, messages)
|
||
if block == "" {
|
||
return messages
|
||
}
|
||
out := make([]model.Message, 0, len(messages)+1)
|
||
out = append(out, model.TextMessage(model.RoleSystem, block))
|
||
out = append(out, messages...)
|
||
return out
|
||
}
|
||
|
||
// memoryBlock 渲染该轮的记忆块；无可注入内容时返回 ""（无
|
||
// retriever/persona、检索出错、或没有相关记忆）。它
|
||
// 从 injectMemory 中拆出，因为组装会在记忆块与对话之间
|
||
// 放置其他前导块（design/22 §7-5 的角色上下文），
|
||
// 因此它需要块本身而不是已前置好的切片。
|
||
func (l *Loop) memoryBlock(ctx context.Context, messages []model.Message) string {
|
||
personaID := l.personaFor(ctx)
|
||
if l.retriever == nil || personaID == "" {
|
||
return ""
|
||
}
|
||
env := sessionEnvFrom(ctx)
|
||
block, err := l.retrieveMemoryBlock(ctx, env, personaID, queryFromMessages(messages))
|
||
if err != nil {
|
||
l.log.Warn("memory inject", "error", err, "env", env.ActorEnv)
|
||
return ""
|
||
}
|
||
return block
|
||
}
|
||
|
||
// retrieveMemoryBlock 返回 env 与该 turn 角色的记忆注入块。
|
||
// owner 全权限环境保持既有的 Inject 行为（候选集已并入
|
||
// user/project 共享域）。受限环境（群聊 / 陌生人）只注入
|
||
// 非敏感的共享域记忆 —— 用户习惯（ScopeUser）与项目记忆
|
||
// （ScopeProject，人人可读）—— 并把该块标记为群聊隔离，
|
||
// 使调用方能区分隔离读取与主人的完整读取；主人私有的
|
||
// 角色日常记忆（ScopePersona）属敏感项，绝不注入
|
||
// （design/03 §5「不泄露主人敏感信息」+ 主人 2026-10-03
|
||
// 裁决「用户习惯记忆与不敏感数据：可查、不可写」）。
|
||
func (l *Loop) retrieveMemoryBlock(ctx context.Context, env SessionEnv, personaID, query string) (string, error) {
|
||
if !env.Restricted() {
|
||
return l.retriever.Inject(ctx, personaID, query, 0)
|
||
}
|
||
results, err := l.retriever.Retrieve(ctx, personaID, query, 0)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
shared := make([]memory.Result, 0, len(results))
|
||
for _, r := range results {
|
||
if sharedScope(r.Item.Scope) {
|
||
shared = append(shared, r)
|
||
}
|
||
}
|
||
if len(shared) == 0 {
|
||
return "", nil
|
||
}
|
||
return formatMemoryBlock(shared, "群聊隔离"), nil
|
||
}
|
||
|
||
// sharedScope 报告 scope 是否为群聊/陌生人可查的非敏感共享域：
|
||
// 用户习惯（user，全局共享）与项目记忆（project，人人可读）。
|
||
// 角色日常（persona）是主人私有的敏感项，不在其列
|
||
// （design/03 §3/§5 的敏感度划分即作用域本身）。
|
||
func sharedScope(s memory.Scope) bool {
|
||
return s == memory.ScopeUser || s == memory.ScopeProject
|
||
}
|
||
|
||
// formatMemoryBlock 由检索结果构建前导系统块。非空
|
||
// marker 作为该块的前缀，使隔离的群聊读取可与
|
||
// 主人的完整读取区分开。
|
||
func formatMemoryBlock(results []memory.Result, marker string) string {
|
||
var b strings.Builder
|
||
if marker != "" {
|
||
b.WriteString(marker)
|
||
b.WriteString("：")
|
||
}
|
||
b.WriteString("相关记忆（按相关度 × 衰减排序）：")
|
||
for _, res := range results {
|
||
b.WriteByte('\n')
|
||
b.WriteString("- [")
|
||
b.WriteString(string(res.Item.Kind))
|
||
b.WriteString("] ")
|
||
b.WriteString(res.Item.Content)
|
||
}
|
||
return b.String()
|
||
}
|
||
|
||
// assembleContext 为一轮构建模型可见的消息列表（design
|
||
// §1.3 每轮 assemble，循环中与 OpenClaw 的 transformContext 等价之物）：它
|
||
// 重新注入该角色的相关长期记忆 —— 检索每轮都从
|
||
// 当前对话重新运行，不从首轮缓存，因此
|
||
// 支持跨轮刷新 —— 在存在时并入每会话的工作记忆摘要，
|
||
// 渲染对话的角色/发言人归属（design/22
|
||
// §7-5），最后应用可选的 TransformContext 钩子，使调用方能
|
||
// 投影/过滤模型实际所见。什么都没配置时
|
||
// 它原样返回输入。
|
||
//
|
||
// 既有顺序被精确保留：前导块依次是每轮的
|
||
// 时间、工作记忆摘要与记忆块；角色
|
||
// 声明紧接其后（就在它所描述的对话之前），而发言人
|
||
// 边界说明被插入对话本身。没有发言人标签的对话
|
||
// 完全不渲染归属，因此每条单角色路径都保持多角色支持
|
||
// 之前的提示词。
|
||
func (l *Loop) assembleContext(ctx context.Context, messages []model.Message, round int) []model.Message {
|
||
var leading []model.Message
|
||
// 每轮「当前时间」，使模型自然知道墙上时钟（而非首轮
|
||
// 快照）。以已配置的时区为条件，使测试与无时区的运行
|
||
// 保持其精确的前导消息数。它位于工作记忆 /
|
||
// 记忆块之前，且仍可被 transformContext 投影。
|
||
if l.timeLocation != nil {
|
||
leading = append(leading, l.currentTimeBlock())
|
||
}
|
||
if l.working != nil {
|
||
if s := l.working.Summarize(); s != "" {
|
||
leading = append(leading, model.TextMessage(model.RoleSystem, "工作记忆: "+s))
|
||
}
|
||
}
|
||
if block := l.memoryBlock(ctx, messages); block != "" {
|
||
leading = append(leading, model.TextMessage(model.RoleSystem, block))
|
||
}
|
||
decl, ok, view := l.attributeConversation(ctx, messages)
|
||
if ok {
|
||
leading = append(leading, decl)
|
||
}
|
||
out := append(leading, view...)
|
||
if l.transformContext != nil {
|
||
out = l.transformContext(ctx, out, round)
|
||
}
|
||
return out
|
||
}
|
||
|
||
// compactIfNeeded 在估算上下文超出配置的 MaxContextTokens
|
||
// 时压缩对话。压缩被关闭（MaxContextTokens 非正）、
|
||
// compactor 为 nil、估算在预算内、或 compactor 返回错误时
|
||
// 它是空操作 —— 压缩失败会被记录日志并保持对话
|
||
// 原样，而不是让 run 失败。
|
||
func (l *Loop) compactIfNeeded(ctx context.Context, messages []model.Message) []model.Message {
|
||
if l.maxContextTokens <= 0 || l.compactor == nil {
|
||
return messages
|
||
}
|
||
compacted, err := l.compactor.Compact(ctx, messages, l.maxContextTokens)
|
||
if err != nil {
|
||
l.log.Warn("context compact", "error", err, "tokens", l.maxContextTokens)
|
||
return messages
|
||
}
|
||
if compacted == nil {
|
||
return messages
|
||
}
|
||
return compacted
|
||
}
|
||
|
||
// extractMemory 在当前对话上运行提取钩子（design §6）
|
||
// 并持久化导出的条目。错误会被记录日志，绝不致命，因此
|
||
// 一次糟糕的提取无法中止 run。
|
||
//
|
||
// 在受限上下文（群聊 / 陌生人，!MemWritable）中提取被
|
||
// 完全跳过：该对话可以只读地读取共享记忆，但绝不
|
||
// 持久化任何可能泄露主人敏感信息的东西（design/02
|
||
// §3, design/11 §2）。
|
||
func (l *Loop) extractMemory(ctx context.Context, messages []model.Message, step int) {
|
||
env := sessionEnvFrom(ctx)
|
||
if env.Restricted() {
|
||
l.log.Debug("memory extract skipped: read-only context", "step", step, "env", env.ActorEnv)
|
||
return
|
||
}
|
||
if l.extractor == nil || l.memory == nil {
|
||
return
|
||
}
|
||
items, err := l.extractor.Extract(ctx, messages)
|
||
if err != nil {
|
||
l.log.Warn("memory extract", "step", step, "error", err)
|
||
return
|
||
}
|
||
l.persistMemories(ctx, items, step)
|
||
}
|
||
|
||
// extractMemoryAsync 在 turn 边界触发提取钩子而不阻塞
|
||
// 回复（design/03 §1: 每轮对话后后台异步抽取）。配置的
|
||
// 提取器能异步提取（LLM extractor）时，对话被交给
|
||
// 后台 goroutine，本调用立即返回，因此慢速
|
||
// 提取模型绝不延迟响应。确定性提取器
|
||
// （ManualExtractor）不实现 asyncExtractor，就地运行，这
|
||
// 使其廉价的 "记住 X" 路径保持可观测且不变。
|
||
//
|
||
// 与 extractMemory 相同的门仍然适用：受限对话绝不
|
||
// 被提取（只读隔离，design/02 §3 / design/11 §2），缺少
|
||
// extractor 或 store 时为空操作。提取失败与 store 失败会
|
||
// 带循环的来源记录日志，绝不向上传播。
|
||
func (l *Loop) extractMemoryAsync(ctx context.Context, messages []model.Message, step int) {
|
||
env := sessionEnvFrom(ctx)
|
||
if env.Restricted() {
|
||
l.log.Debug("memory extract skipped: read-only context", "step", step, "env", env.ActorEnv)
|
||
return
|
||
}
|
||
if l.extractor == nil || l.memory == nil {
|
||
return
|
||
}
|
||
async, ok := l.extractor.(asyncExtractor)
|
||
if !ok {
|
||
l.extractMemory(ctx, messages, step)
|
||
return
|
||
}
|
||
// 该 turn 的 context 随请求消亡，而提取刻意比它活得
|
||
// 更久，因此 goroutine 拿到一份脱离的 context 副本（值 ——
|
||
// 其中包括会话环境 —— 被保留），并处于自己的超时之下。
|
||
dctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), memoryExtractTimeout)
|
||
// 随着该 turn 继续，调用方会不断向 messages 追加，因此
|
||
// goroutine 读取一份私有快照，而不是在共享切片上竞争。
|
||
snapshot := append([]model.Message(nil), messages...)
|
||
results, err := async.ExtractAsync(dctx, snapshot)
|
||
if err != nil {
|
||
cancel()
|
||
l.log.Warn("memory extract dispatch", "step", step, "error", err)
|
||
return
|
||
}
|
||
go func() {
|
||
defer cancel()
|
||
res, ok := <-results
|
||
if !ok {
|
||
return
|
||
}
|
||
if res.Err != nil {
|
||
l.log.Warn("memory extract", "step", step, "error", res.Err)
|
||
return
|
||
}
|
||
l.persistMemories(dctx, res.Items, step)
|
||
}()
|
||
}
|
||
|
||
// persistMemories 把提取出的条目写入记忆 store，为每个条目
|
||
// 打上产出该 turn 的角色。该赋值刻意无条件：
|
||
// extractor 刻意不感知角色（它不再持有一个在构造期
|
||
// 内嵌的人格，cmd/cyrene/extractor.go），否则多角色
|
||
// 进程会把每条记忆都归到一个角色下。角色为空（完全未配置
|
||
// persona）时条目保持 extractor 产出时的样子。
|
||
//
|
||
// 溯源同样在此盖真实 id（裁决 04.4.3，design/03 §2）：
|
||
// source_conversation 由本持久化点从运行上下文取 Run 的
|
||
// session id 与 gateway 挂载的 turn id 结构化编码
|
||
// （memory.FormatSourceConversation），无条件覆盖 extractor
|
||
// 产出值 —— 模型自由文本绝无机会落库为来源。store
|
||
// 失败会被记录日志，绝不致命，因此一条被拒的条目无法中止一个 turn。
|
||
func (l *Loop) persistMemories(ctx context.Context, items []memory.MemoryItem, step int) {
|
||
personaID := l.personaFor(ctx)
|
||
source := memory.FormatSourceConversation(loopSession(ctx), loopTurn(ctx))
|
||
for _, it := range items {
|
||
if personaID != "" {
|
||
it.PersonaID = personaID
|
||
}
|
||
it.SourceConversation = source
|
||
if _, err := l.memory.Add(ctx, it); err != nil {
|
||
l.log.Warn("memory store", "step", step, "error", err)
|
||
}
|
||
}
|
||
}
|
||
|
||
// queryFromMessages 从对话导出检索查询：最后一个用户
|
||
// turn 的文本，回退到最后一条消息的文本，再回退到 ""。
|
||
func queryFromMessages(messages []model.Message) string {
|
||
for i := len(messages) - 1; i >= 0; i-- {
|
||
if messages[i].Role == model.RoleUser {
|
||
return messageText(messages[i])
|
||
}
|
||
}
|
||
if n := len(messages); n > 0 {
|
||
return messageText(messages[n-1])
|
||
}
|
||
return ""
|
||
}
|
||
|
||
// messageText 返回一条消息各文本部件的拼接纯文本载荷，
|
||
// 这是用于构建检索查询的共享视图。
|
||
func messageText(m model.Message) string {
|
||
var b strings.Builder
|
||
for _, part := range m.Content {
|
||
if part.Type != model.ContentTypeText {
|
||
continue
|
||
}
|
||
if b.Len() > 0 {
|
||
b.WriteByte(' ')
|
||
}
|
||
b.WriteString(part.Text)
|
||
}
|
||
return b.String()
|
||
}
|