498 lines
17 KiB
Go
498 lines
17 KiB
Go
// memory 包实现 Cyrene v2 记忆子系统，规格见
|
||
// DevDocs/design/03-记忆.md。
|
||
//
|
||
// 它对外暴露三层：
|
||
//
|
||
// - Store：基于 SQLite 的长期记忆存储。它按角色隔离持久化记忆条目
|
||
// （design §0 layer ②、§4），经 AddScoped/ListScope/ListProject/
|
||
// DeleteScoped 施加三作用域访问规则（design §3），并提供
|
||
// Add/Get/ListByPersona/Delete，外加一条手动的
|
||
// 「记住」路径（design §1）。每条条目可携带 project_id，链到
|
||
// 项目索引层级。
|
||
// - WorkingMemory：每会话的消息滑动窗口加一份摘要
|
||
// （design §0 layer ①）。
|
||
// - Retriever：相关性（关键词 + 可选语义）× 衰减检索，带
|
||
// 溯源与上下文注入（design §2）。它可选接收
|
||
// vector.Index 作为持久化语义路径，以及 projectindex.Store，使
|
||
// 项目作用域的命中带出其链接的项目条目。
|
||
//
|
||
// 提取（design §1）经 Extractor 接口暴露，有两个
|
||
// 实现：ManualExtractor（规则式，记录一条显式的「记住 X」
|
||
// 指令）与 LLMExtractor（驱动 model.Adapter 对
|
||
// 一轮对话去噪去重，经 Extract 同步运行或经 ExtractAsync 异步运行）。
|
||
//
|
||
// M4 里程碑在此实现：schema 携带 decay_weight 与
|
||
// project_id，检索器接入可选的 embedding.Embedder / vector.Index
|
||
// 与项目索引存储，存储层施加 user/persona/project
|
||
// 作用域规则。除既有包外不需要任何外部依赖。
|
||
package memory
|
||
|
||
import (
|
||
"context"
|
||
"crypto/rand"
|
||
"database/sql"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"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 正交，
|
||
// 两者都存于每条条目上（design §3 注）。
|
||
type Scope string
|
||
|
||
const (
|
||
// ScopeUser 是关于用户习惯的记忆，跨角色共享。
|
||
ScopeUser Scope = "user"
|
||
// ScopePersona 是角色的日常记忆；其他角色只读。
|
||
ScopePersona Scope = "persona"
|
||
// ScopeProject 是项目作用域记忆，人人可读可写。
|
||
ScopeProject Scope = "project"
|
||
)
|
||
|
||
// valid 报告 s 是否为已定义的作用域之一。
|
||
func (s Scope) valid() bool {
|
||
switch s {
|
||
case ScopeUser, ScopePersona, ScopeProject:
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
// Kind 分类一条长期记忆条目的内容类别。
|
||
type Kind string
|
||
|
||
const (
|
||
// KindRelationship 是关于「你和我」的记忆：偏好、习惯、
|
||
// 约定（design §0「关系记忆」）。
|
||
KindRelationship Kind = "relationship"
|
||
// KindWorldEvent 是关于客观事实与世界事件的记忆
|
||
// （design §0「世界/事件记忆」）。
|
||
KindWorldEvent Kind = "world_event"
|
||
)
|
||
|
||
// valid 报告 k 是否为已定义的 kind 之一。
|
||
func (k Kind) valid() bool {
|
||
switch k {
|
||
case KindRelationship, KindWorldEvent:
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
// MemoryItem 是一条长期记忆记录。零值不是合法
|
||
// 条目：PersonaID 与 Content 必填，Scope 与 Kind 由 Add 规范化到
|
||
// 其默认值。
|
||
type MemoryItem struct {
|
||
ID string
|
||
PersonaID string
|
||
RoleID string
|
||
ProjectID string
|
||
Scope Scope
|
||
Kind Kind
|
||
Content string
|
||
Tags []string
|
||
Confidence float64
|
||
SourceConversation string
|
||
CreatedAt time.Time
|
||
UpdatedAt time.Time
|
||
DecayWeight float64
|
||
}
|
||
|
||
// 区分 memory 包各类失败的哨兵错误。每个都在
|
||
// 调用点用 %w 包裹，调用方可用 errors.Is 判断。
|
||
var (
|
||
// ErrMissingPersona 在条目没有 PersonaID 时返回。
|
||
ErrMissingPersona = errors.New("memory: persona_id is required")
|
||
|
||
// ErrMissingContent 在条目没有内容时返回。
|
||
ErrMissingContent = errors.New("memory: content is required")
|
||
|
||
// ErrInvalidScope 在条目的 Scope 不是已定义作用域时返回。
|
||
ErrInvalidScope = errors.New("memory: invalid scope")
|
||
|
||
// ErrInvalidKind 在条目的 Kind 不是已定义 kind 时返回。
|
||
ErrInvalidKind = errors.New("memory: invalid kind")
|
||
|
||
// ErrNotFound 在条目 id 不存在时返回。
|
||
ErrNotFound = errors.New("memory: item not found")
|
||
|
||
// ErrNilDB 在 NewStore 收到 nil *sql.DB 时返回。
|
||
ErrNilDB = errors.New("memory: nil db")
|
||
)
|
||
|
||
// schema 默认值与时间窗口常量。
|
||
const (
|
||
defaultConfidence = 1.0
|
||
defaultDecayWeight = 1.0
|
||
)
|
||
|
||
// Store 是基于 *sql.DB 的长期记忆存储。
|
||
//
|
||
// 存储自己拥有 schema：NewStore 幂等地创建 memory_items 表及其
|
||
// 索引。并发交由 *sql.DB，按约定它可安全
|
||
// 并发使用，因此除语义索引的接入外，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
|
||
// 辅助函数即可服务 Get 与 ListByPersona。
|
||
type rowScanner interface {
|
||
Scan(dest ...any) error
|
||
}
|
||
|
||
// NewStore 在 db 上构建 Store，确保 memory_items schema 存在。
|
||
//
|
||
// db 必须是存活且已迁移的连接（迁移框架本身由
|
||
// internal/db 拥有）。nil db 返回 ErrNilDB。log 可为 nil，此时
|
||
// 使用丢弃型 logger。ctx 控制 schema 创建，调用方必须
|
||
// 遵守。任何旧的变宽时间戳都就地
|
||
// 规范化（见 normalizeMemoryTimestamps），使文本顺序与时间顺序一致。
|
||
func NewStore(ctx context.Context, db *sql.DB, log *slog.Logger) (*Store, error) {
|
||
if db == nil {
|
||
return nil, ErrNilDB
|
||
}
|
||
log = logger.WithSource(discardLogger(log), "memory")
|
||
for _, ddl := range []string{
|
||
createMemoryItemsTable,
|
||
createMemoryItemsPersonaIndex,
|
||
createMemoryItemsPersonaCreatedIndex,
|
||
} {
|
||
if _, err := db.ExecContext(ctx, ddl); err != nil {
|
||
return nil, fmt.Errorf("memory: apply schema: %w", err)
|
||
}
|
||
}
|
||
if err := ensureProjectIDColumn(ctx, db); err != nil {
|
||
return nil, err
|
||
}
|
||
// 记忆整理（design/23 §10）的列与两张新表：同样在这一处幂等前滚，
|
||
// 使「在任意 *sql.DB 上都能整理」与 memory_items 自身口径一致。
|
||
if err := ensureTidySchema(ctx, db); err != nil {
|
||
return nil, err
|
||
}
|
||
if err := normalizeMemoryTimestamps(ctx, db); err != nil {
|
||
return nil, err
|
||
}
|
||
return &Store{db: db, log: log}, nil
|
||
}
|
||
|
||
// 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
|
||
}
|
||
var err error
|
||
if item, err = applyTimestamps(item); err != nil {
|
||
return MemoryItem{}, err
|
||
}
|
||
tagsJSON, err := json.Marshal(item.Tags)
|
||
if err != nil {
|
||
return MemoryItem{}, fmt.Errorf("memory: encode tags: %w", err)
|
||
}
|
||
|
||
_, err = s.db.ExecContext(ctx, `INSERT INTO memory_items
|
||
(id, persona_id, role_id, project_id, scope, kind, content, tags, confidence,
|
||
source_conversation, created_at, updated_at, decay_weight)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||
item.ID, item.PersonaID, item.RoleID, item.ProjectID, string(item.Scope), string(item.Kind),
|
||
item.Content, string(tagsJSON), item.Confidence, item.SourceConversation,
|
||
formatTime(item.CreatedAt), formatTime(item.UpdatedAt), item.DecayWeight)
|
||
if err != nil {
|
||
return MemoryItem{}, fmt.Errorf("memory: insert item: %w", err)
|
||
}
|
||
s.indexItem(ctx, item)
|
||
return item, nil
|
||
}
|
||
|
||
// Get 返回给定 id 的条目。id 为空或
|
||
// 不存在时返回 ErrNotFound。
|
||
func (s *Store) Get(ctx context.Context, id string) (MemoryItem, error) {
|
||
if strings.TrimSpace(id) == "" {
|
||
return MemoryItem{}, ErrNotFound
|
||
}
|
||
row := s.db.QueryRowContext(ctx, `SELECT id, persona_id, role_id, project_id, scope, kind,
|
||
content, tags, confidence, source_conversation, created_at, updated_at,
|
||
decay_weight FROM memory_items WHERE id = ?`, id)
|
||
item, err := scanItem(row)
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return MemoryItem{}, ErrNotFound
|
||
}
|
||
if err != nil {
|
||
return MemoryItem{}, fmt.Errorf("memory: get item: %w", err)
|
||
}
|
||
return item, nil
|
||
}
|
||
|
||
// ListByPersona 返回 personaID 的全部条目，最新的在前。rowid 打破
|
||
// created_at 并列，使同一时钟刻内写入的条目保持稳定的
|
||
// 最新优先顺序。它是角色隔离边界：属于
|
||
// 其他角色的条目绝不返回，因此角色之间从不共享各自的角色日常
|
||
// 命名空间（design §4）。它只是检索候选集的*自有*
|
||
// 那一半 —— 检索器在排名前把它与共享的用户/项目作用域
|
||
// 取并集（design §3/§4，见 Retriever.visibleCandidates）。
|
||
func (s *Store) ListByPersona(ctx context.Context, personaID string) ([]MemoryItem, error) {
|
||
if strings.TrimSpace(personaID) == "" {
|
||
return nil, ErrMissingPersona
|
||
}
|
||
rows, err := s.db.QueryContext(ctx, `SELECT id, persona_id, role_id, project_id, scope, kind,
|
||
content, tags, confidence, source_conversation, created_at, updated_at,
|
||
decay_weight FROM memory_items WHERE persona_id = ?
|
||
ORDER BY created_at DESC, rowid DESC`, personaID)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("memory: list persona items: %w", err)
|
||
}
|
||
return scanItems(rows, "persona item")
|
||
}
|
||
|
||
// List 返回跨所有角色的每条记忆条目，最新的在前。rowid
|
||
// 打破 created_at 并列，使同一时钟刻内写入的条目保持
|
||
// 稳定的最新优先顺序。它是聚合/管理视图，枚举全部
|
||
// 已持久化记忆，不受角色隔离边界限制；必须
|
||
// 遵守隔离的调用方应改用 ListByPersona 或 ListScope。返回的
|
||
// 切片非 nil。
|
||
func (s *Store) List(ctx context.Context) ([]MemoryItem, error) {
|
||
rows, err := s.db.QueryContext(ctx, `SELECT id, persona_id, role_id, project_id, scope, kind,
|
||
content, tags, confidence, source_conversation, created_at, updated_at,
|
||
decay_weight FROM memory_items
|
||
ORDER BY created_at DESC, rowid DESC`)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("memory: list items: %w", err)
|
||
}
|
||
return scanItems(rows, "item")
|
||
}
|
||
|
||
// scanItems 把每一行读入 MemoryItem 切片，关闭 rows 并用
|
||
// 操作名包裹 scan/遍历失败。成功时总是返回
|
||
// 非 nil、可能为空的切片。
|
||
func scanItems(rows *sql.Rows, op string) ([]MemoryItem, error) {
|
||
defer func() { _ = rows.Close() }()
|
||
|
||
items := make([]MemoryItem, 0)
|
||
for rows.Next() {
|
||
item, err := scanItem(rows)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("memory: scan %s: %w", op, err)
|
||
}
|
||
items = append(items, item)
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return nil, fmt.Errorf("memory: iterate %s: %w", op, err)
|
||
}
|
||
return items, nil
|
||
}
|
||
|
||
// Delete 删除给定 id 的条目，id 为空或
|
||
// 没有删除任何行时返回 ErrNotFound。调用方只应传入经
|
||
// 角色作用域读取（ListByPersona/Get）得到的 id，使删除始终留在
|
||
// 调用方的角色边界内。
|
||
func (s *Store) Delete(ctx context.Context, id string) error {
|
||
if strings.TrimSpace(id) == "" {
|
||
return ErrNotFound
|
||
}
|
||
res, err := s.db.ExecContext(ctx, `DELETE FROM memory_items WHERE id = ?`, id)
|
||
if err != nil {
|
||
return fmt.Errorf("memory: delete item: %w", err)
|
||
}
|
||
n, err := res.RowsAffected()
|
||
if err != nil {
|
||
return fmt.Errorf("memory: delete row count: %w", err)
|
||
}
|
||
if n == 0 {
|
||
return ErrNotFound
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// RememberManual 为 content 文本写入一条长期记忆条目 —— design §1 中的
|
||
// 「记住 X」路径。条目默认取角色作用域，
|
||
// 并以满置信度记为关系记忆。返回已落库的条目。
|
||
func (s *Store) RememberManual(ctx context.Context, personaID, roleID, text string) (MemoryItem, error) {
|
||
item := MemoryItem{
|
||
PersonaID: personaID,
|
||
RoleID: roleID,
|
||
Scope: ScopePersona,
|
||
Kind: KindRelationship,
|
||
Content: strings.TrimSpace(text),
|
||
}
|
||
if item.Content == "" {
|
||
return MemoryItem{}, ErrMissingContent
|
||
}
|
||
return s.Add(ctx, item)
|
||
}
|
||
|
||
// normalize 填充默认的 Scope/Kind/Confidence/DecayWeight，然后校验。
|
||
// 它由 Add 调用，因此调用方可以省略取值，同时仍能
|
||
// 捕获显式非法值。
|
||
func (it *MemoryItem) normalize() error {
|
||
if it.Scope == "" {
|
||
it.Scope = ScopePersona
|
||
}
|
||
if it.Kind == "" {
|
||
it.Kind = KindRelationship
|
||
}
|
||
if it.Confidence == 0 {
|
||
it.Confidence = defaultConfidence
|
||
}
|
||
if it.DecayWeight == 0 {
|
||
it.DecayWeight = defaultDecayWeight
|
||
}
|
||
return it.validate()
|
||
}
|
||
|
||
// validate 检查必填与枚举字段。它是纯函数：不施加
|
||
// 默认值，因此也可用于拒绝调用方提供的值。
|
||
func (it *MemoryItem) validate() error {
|
||
if strings.TrimSpace(it.PersonaID) == "" {
|
||
return ErrMissingPersona
|
||
}
|
||
if strings.TrimSpace(it.Content) == "" {
|
||
return ErrMissingContent
|
||
}
|
||
if !it.Scope.valid() {
|
||
return ErrInvalidScope
|
||
}
|
||
if !it.Kind.valid() {
|
||
return ErrInvalidKind
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// applyTimestamps 分配 id 与缺失的 CreatedAt/UpdatedAt 值。
|
||
// 仅在无法生成新 id 时返回错误。
|
||
func applyTimestamps(item MemoryItem) (MemoryItem, error) {
|
||
now := time.Now().UTC()
|
||
if item.ID == "" {
|
||
id, err := newID()
|
||
if err != nil {
|
||
return MemoryItem{}, err
|
||
}
|
||
item.ID = id
|
||
}
|
||
if item.CreatedAt.IsZero() {
|
||
item.CreatedAt = now
|
||
}
|
||
if item.UpdatedAt.IsZero() {
|
||
item.UpdatedAt = now
|
||
}
|
||
return item, nil
|
||
}
|
||
|
||
// scanItem 读取一行 memory_items。tags 以 JSON 数组字符串存储；
|
||
// 时间戳以定宽 UTC 字符串（timeLayout）存储，因此文本
|
||
// 比较即可按时间排序。
|
||
func scanItem(sc rowScanner) (MemoryItem, error) {
|
||
var (
|
||
it MemoryItem
|
||
scope, kind string
|
||
tagsJSON string
|
||
createdAt, updatedAt string
|
||
)
|
||
if err := sc.Scan(&it.ID, &it.PersonaID, &it.RoleID, &it.ProjectID, &scope, &kind,
|
||
&it.Content, &tagsJSON, &it.Confidence, &it.SourceConversation,
|
||
&createdAt, &updatedAt, &it.DecayWeight); err != nil {
|
||
return MemoryItem{}, err
|
||
}
|
||
it.Scope = Scope(scope)
|
||
it.Kind = Kind(kind)
|
||
if err := json.Unmarshal([]byte(tagsJSON), &it.Tags); err != nil {
|
||
return MemoryItem{}, fmt.Errorf("memory: decode tags: %w", err)
|
||
}
|
||
if it.Tags == nil {
|
||
it.Tags = []string{}
|
||
}
|
||
|
||
var err error
|
||
if it.CreatedAt, err = parseTime(createdAt); err != nil {
|
||
return MemoryItem{}, err
|
||
}
|
||
if it.UpdatedAt, err = parseTime(updatedAt); err != nil {
|
||
return MemoryItem{}, err
|
||
}
|
||
return it, nil
|
||
}
|
||
|
||
// newID 返回一个 128 位密码学随机 id，十六进制编码。它避开
|
||
// UUID 依赖，使本包保持自包含。rand.Read 只在
|
||
// 平台熵源损坏时失败，因此错误向上传播。
|
||
func newID() (string, error) {
|
||
b := make([]byte, 16)
|
||
if _, err := rand.Read(b); err != nil {
|
||
return "", fmt.Errorf("memory: generate id: %w", err)
|
||
}
|
||
return hex.EncodeToString(b), nil
|
||
}
|
||
|
||
// discardLogger 在 log 非 nil 时返回 log，否则返回一个不写任何地方的 logger。
|
||
func discardLogger(log *slog.Logger) *slog.Logger {
|
||
if log != nil {
|
||
return log
|
||
}
|
||
return slog.New(slog.NewTextHandler(io.Discard, nil))
|
||
}
|
||
|
||
// schema DDL。internal/db 中的迁移框架是领域表的权威
|
||
// 机制，但这里刻意不用它：db 包的迁移测试断言精确的
|
||
// 迁移数量，而记忆存储必须自包含、
|
||
// 可在任意 *sql.DB 上使用。表与索引因此
|
||
// 由 NewStore 幂等创建。
|
||
const createMemoryItemsTable = `
|
||
CREATE TABLE IF NOT EXISTS memory_items (
|
||
id TEXT PRIMARY KEY,
|
||
persona_id TEXT NOT NULL,
|
||
role_id TEXT NOT NULL DEFAULT '',
|
||
project_id TEXT NOT NULL DEFAULT '',
|
||
scope TEXT NOT NULL,
|
||
kind TEXT NOT NULL,
|
||
content TEXT NOT NULL,
|
||
tags TEXT NOT NULL DEFAULT '[]',
|
||
confidence REAL NOT NULL DEFAULT 1.0,
|
||
source_conversation TEXT NOT NULL DEFAULT '',
|
||
created_at TEXT NOT NULL,
|
||
updated_at TEXT NOT NULL,
|
||
decay_weight REAL NOT NULL DEFAULT 1.0,
|
||
status TEXT NOT NULL DEFAULT 'active',
|
||
merged_into TEXT NOT NULL DEFAULT '',
|
||
merged_from TEXT NOT NULL DEFAULT '[]',
|
||
merge_count INTEGER NOT NULL DEFAULT 1,
|
||
occurred_at TEXT NOT NULL DEFAULT '',
|
||
valid_to TEXT NOT NULL DEFAULT '',
|
||
expires_at TEXT NOT NULL DEFAULT ''
|
||
)`
|
||
|
||
const createMemoryItemsPersonaIndex = `
|
||
CREATE INDEX IF NOT EXISTS idx_memory_items_persona_id
|
||
ON memory_items (persona_id)`
|
||
|
||
const createMemoryItemsPersonaCreatedIndex = `
|
||
CREATE INDEX IF NOT EXISTS idx_memory_items_persona_id_created_at
|
||
ON memory_items (persona_id, created_at)`
|