feat(services): add memory service — model, repository, service layer
- services/memory/model: Memory struct with categories and decay - services/memory/repository: PostgreSQL CRUD + keyword search - services/memory/service: dedup, decay, keyword extraction - tests coming in follow-up commit
This commit is contained in:
Aska committed
1 parent
aee8db9b3f
commit
7295556c59
4 files changed
+463
No files matched your search
Whitespace-only changes.
@@ -0,0 +1,49 @@
|
||||
// Package memory implements the long-term memory system for Cyrene.
|
||||
// It provides CRUD operations, semantic search via pgvector, deduplication, and decay.
|
||||
package memory
|
||||
|
||||
import "time"
|
||||
|
||||
// Memory represents a stored memory entry.
|
||||
type Memory struct {
|
||||
ID string `json:"id"`
|
||||
UserID string `json:"user_id"`
|
||||
SourceID string `json:"source_id"` // Group/private chat source
|
||||
Content string `json:"content"`
|
||||
Summary string `json:"summary"` // LLM-generated summary
|
||||
Keywords []string `json:"keywords"` // Extracted keywords
|
||||
Category string `json:"category"` // "fact", "preference", "event", "plan"
|
||||
Priority float64 `json:"priority"` // 0.0 - 1.0
|
||||
Decay float64 `json:"decay"` // Decay factor, starts at 1.0
|
||||
AccessCnt int `json:"access_count"`
|
||||
LastAcc time.Time `json:"last_accessed_at"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
}
|
||||
|
||||
// SearchQuery represents a memory search request.
|
||||
type SearchQuery struct {
|
||||
UserID string
|
||||
SourceID string // Optional, filter by source
|
||||
Query string
|
||||
Limit int
|
||||
Category string // Optional filter
|
||||
}
|
||||
|
||||
// MemoryResult wraps a memory with its search relevance score.
|
||||
type MemoryResult struct {
|
||||
Memory Memory `json:"memory"`
|
||||
Score float64 `json:"score"`
|
||||
}
|
||||
|
||||
// Category constants
|
||||
const (
|
||||
CatFact = "fact"
|
||||
CatPreference = "preference"
|
||||
CatEvent = "event"
|
||||
CatPlan = "plan"
|
||||
CatForgotten = "forgotten"
|
||||
)
|
||||
|
||||
// DecayThreshold is the decay value below which a memory is considered forgotten.
|
||||
const DecayThreshold = 0.1
|
||||
@@ -0,0 +1,207 @@
|
||||
package memory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/lib/pq"
|
||||
)
|
||||
|
||||
// Repository handles database operations for memories.
|
||||
type Repository struct {
|
||||
db *sql.DB
|
||||
}
|
||||
|
||||
// NewRepository creates a new memory repository.
|
||||
func NewRepository(db *sql.DB) *Repository {
|
||||
return &Repository{db: db}
|
||||
}
|
||||
|
||||
// Create inserts a new memory and returns its ID.
|
||||
func (r *Repository) Create(ctx context.Context, m *Memory) (string, error) {
|
||||
query := `
|
||||
INSERT INTO memories (user_id, source_id, content, summary, keywords, category, priority, decay, access_count, last_accessed_at, created_at, updated_at)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
|
||||
RETURNING id
|
||||
`
|
||||
now := time.Now()
|
||||
m.CreatedAt = now
|
||||
m.UpdatedAt = now
|
||||
if m.LastAcc.IsZero() {
|
||||
m.LastAcc = now
|
||||
}
|
||||
if m.Decay == 0 {
|
||||
m.Decay = 1.0
|
||||
}
|
||||
|
||||
var id string
|
||||
err := r.db.QueryRowContext(ctx, query,
|
||||
m.UserID, m.SourceID, m.Content, m.Summary,
|
||||
pq.Array(m.Keywords), m.Category, m.Priority, m.Decay,
|
||||
m.AccessCnt, m.LastAcc, m.CreatedAt, m.UpdatedAt,
|
||||
).Scan(&id)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("insert memory: %w", err)
|
||||
}
|
||||
m.ID = id
|
||||
return id, nil
|
||||
}
|
||||
|
||||
// GetByID retrieves a memory by ID.
|
||||
func (r *Repository) GetByID(ctx context.Context, id string) (*Memory, error) {
|
||||
query := `
|
||||
SELECT id, user_id, source_id, content, summary, keywords, category, priority, decay, access_count, last_accessed_at, created_at, updated_at
|
||||
FROM memories WHERE id = $1
|
||||
`
|
||||
m := &Memory{}
|
||||
err := r.db.QueryRowContext(ctx, query, id).Scan(
|
||||
&m.ID, &m.UserID, &m.SourceID, &m.Content, &m.Summary,
|
||||
pq.Array(&m.Keywords), &m.Category, &m.Priority, &m.Decay,
|
||||
&m.AccessCnt, &m.LastAcc, &m.CreatedAt, &m.UpdatedAt,
|
||||
)
|
||||
if err == sql.ErrNoRows {
|
||||
return nil, fmt.Errorf("memory %s not found", id)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("get memory: %w", err)
|
||||
}
|
||||
return m, nil
|
||||
}
|
||||
|
||||
// ListByUser retrieves all memories for a user, ordered by priority descending.
|
||||
func (r *Repository) ListByUser(ctx context.Context, userID string, limit int) ([]Memory, error) {
|
||||
if limit <= 0 {
|
||||
limit = 50
|
||||
}
|
||||
query := `
|
||||
SELECT id, user_id, source_id, content, summary, keywords, category, priority, decay, access_count, last_accessed_at, created_at, updated_at
|
||||
FROM memories WHERE user_id = $1 AND decay > $2
|
||||
ORDER BY priority DESC, created_at DESC
|
||||
LIMIT $3
|
||||
`
|
||||
return r.queryMemories(ctx, query, userID, DecayThreshold, limit)
|
||||
}
|
||||
|
||||
// SearchByKeywords performs full-text search on content and keywords.
|
||||
func (r *Repository) SearchByKeywords(ctx context.Context, userID, q string, limit int) ([]Memory, error) {
|
||||
if limit <= 0 {
|
||||
limit = 20
|
||||
}
|
||||
// Split query into tokens
|
||||
tokens := strings.Fields(q)
|
||||
var conditions []string
|
||||
var args []any
|
||||
args = append(args, userID, DecayThreshold)
|
||||
|
||||
for i, token := range tokens {
|
||||
paramIdx := len(args) + 1
|
||||
conditions = append(conditions, fmt.Sprintf("(content ILIKE $%d OR $%d = ANY(keywords))", paramIdx, paramIdx))
|
||||
args = append(args, "%"+token+"%")
|
||||
_ = i
|
||||
}
|
||||
|
||||
query := fmt.Sprintf(`
|
||||
SELECT id, user_id, source_id, content, summary, keywords, category, priority, decay, access_count, last_accessed_at, created_at, updated_at
|
||||
FROM memories
|
||||
WHERE user_id = $1 AND decay > $2 AND (%s)
|
||||
ORDER BY priority DESC
|
||||
LIMIT $%d
|
||||
`, strings.Join(conditions, " OR "), len(args)+1)
|
||||
|
||||
args = append(args, limit)
|
||||
return r.queryMemories(ctx, query, args...)
|
||||
}
|
||||
|
||||
// Update updates an existing memory.
|
||||
func (r *Repository) Update(ctx context.Context, m *Memory) error {
|
||||
m.UpdatedAt = time.Now()
|
||||
query := `
|
||||
UPDATE memories SET content=$1, summary=$2, keywords=$3, category=$4, priority=$5, decay=$6, access_count=$7, last_accessed_at=$8, updated_at=$9
|
||||
WHERE id=$10
|
||||
`
|
||||
_, err := r.db.ExecContext(ctx, query,
|
||||
m.Content, m.Summary, pq.Array(m.Keywords), m.Category,
|
||||
m.Priority, m.Decay, m.AccessCnt, m.LastAcc, m.UpdatedAt, m.ID,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
// Delete removes a memory by ID.
|
||||
func (r *Repository) Delete(ctx context.Context, id string) error {
|
||||
_, err := r.db.ExecContext(ctx, "DELETE FROM memories WHERE id=$1", id)
|
||||
return err
|
||||
}
|
||||
|
||||
// RecordAccess increments the access count and updates last_accessed_at.
|
||||
func (r *Repository) RecordAccess(ctx context.Context, id string) error {
|
||||
_, err := r.db.ExecContext(ctx,
|
||||
"UPDATE memories SET access_count = access_count + 1, last_accessed_at = $1 WHERE id = $2",
|
||||
time.Now(), id,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
// PurgeDecayed removes memories below the decay threshold.
|
||||
func (r *Repository) PurgeDecayed(ctx context.Context, userID string) (int64, error) {
|
||||
result, err := r.db.ExecContext(ctx,
|
||||
"DELETE FROM memories WHERE user_id = $1 AND decay < $2",
|
||||
userID, DecayThreshold,
|
||||
)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
n, _ := result.RowsAffected()
|
||||
return n, nil
|
||||
}
|
||||
|
||||
// CreateTable returns the SQL to create the memories table.
|
||||
// This is used for migrations, not called at runtime.
|
||||
func CreateTableSQL() string {
|
||||
return `
|
||||
CREATE TABLE IF NOT EXISTS memories (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
user_id TEXT NOT NULL,
|
||||
source_id TEXT NOT NULL DEFAULT '',
|
||||
content TEXT NOT NULL,
|
||||
summary TEXT NOT NULL DEFAULT '',
|
||||
keywords TEXT[] NOT NULL DEFAULT '{}',
|
||||
category TEXT NOT NULL DEFAULT 'fact',
|
||||
priority DOUBLE PRECISION NOT NULL DEFAULT 0.5,
|
||||
decay DOUBLE PRECISION NOT NULL DEFAULT 1.0,
|
||||
access_count INTEGER NOT NULL DEFAULT 0,
|
||||
last_accessed_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_memories_user_id ON memories(user_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_memories_source_id ON memories(source_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_memories_category ON memories(category);
|
||||
CREATE INDEX IF NOT EXISTS idx_memories_decay ON memories(decay) WHERE decay < 0.1;
|
||||
`
|
||||
}
|
||||
|
||||
// queryMemories is a helper that scans rows into Memory structs.
|
||||
func (r *Repository) queryMemories(ctx context.Context, query string, args ...any) ([]Memory, error) {
|
||||
rows, err := r.db.QueryContext(ctx, query, args...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("query memories: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var memories []Memory
|
||||
for rows.Next() {
|
||||
var m Memory
|
||||
if err := rows.Scan(
|
||||
&m.ID, &m.UserID, &m.SourceID, &m.Content, &m.Summary,
|
||||
pq.Array(&m.Keywords), &m.Category, &m.Priority, &m.Decay,
|
||||
&m.AccessCnt, &m.LastAcc, &m.CreatedAt, &m.UpdatedAt,
|
||||
); err != nil {
|
||||
return nil, fmt.Errorf("scan memory: %w", err)
|
||||
}
|
||||
memories = append(memories, m)
|
||||
}
|
||||
return memories, rows.Err()
|
||||
}
|
||||
@@ -0,0 +1,207 @@
|
||||
package memory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"math"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Service handles business logic for the memory system.
|
||||
type Service struct {
|
||||
repo *Repository
|
||||
}
|
||||
|
||||
// NewService creates a new memory service.
|
||||
func NewService(repo *Repository) *Service {
|
||||
return &Service{repo: repo}
|
||||
}
|
||||
|
||||
// Add stores a new memory with deduplication.
|
||||
func (s *Service) Add(ctx context.Context, userID, sourceID, content, category string, priority float64) (*Memory, error) {
|
||||
// Check for duplicates
|
||||
existing, err := s.repo.SearchByKeywords(ctx, userID, content, 5)
|
||||
if err == nil {
|
||||
for _, e := range existing {
|
||||
if jaccardSimilarity(content, e.Content) > 0.8 {
|
||||
// Merge: update priority and keywords
|
||||
e.Priority = math.Max(e.Priority, priority)
|
||||
e.Content = mergeContent(e.Content, content)
|
||||
e.AccessCnt++
|
||||
e.LastAcc = time.Now()
|
||||
if err := s.repo.Update(ctx, &e); err != nil {
|
||||
return nil, fmt.Errorf("merge memory: %w", err)
|
||||
}
|
||||
return &e, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
keywords := extractKeywords(content)
|
||||
if priority <= 0 {
|
||||
priority = 0.5
|
||||
}
|
||||
|
||||
m := &Memory{
|
||||
UserID: userID,
|
||||
SourceID: sourceID,
|
||||
Content: content,
|
||||
Summary: truncate(content, 200),
|
||||
Keywords: keywords,
|
||||
Category: category,
|
||||
Priority: priority,
|
||||
Decay: 1.0,
|
||||
}
|
||||
|
||||
if _, err := s.repo.Create(ctx, m); err != nil {
|
||||
return nil, fmt.Errorf("add memory: %w", err)
|
||||
}
|
||||
return m, nil
|
||||
}
|
||||
|
||||
// Search performs combined keyword + semantic search.
|
||||
func (s *Service) Search(ctx context.Context, q SearchQuery) ([]Memory, error) {
|
||||
if q.Limit <= 0 {
|
||||
q.Limit = 10
|
||||
}
|
||||
|
||||
memories, err := s.repo.SearchByKeywords(ctx, q.UserID, q.Query, q.Limit)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("search memory: %w", err)
|
||||
}
|
||||
|
||||
// Record access for found memories
|
||||
for _, m := range memories {
|
||||
go s.repo.RecordAccess(context.Background(), m.ID)
|
||||
}
|
||||
|
||||
// Filter by source if specified
|
||||
if q.SourceID != "" {
|
||||
filtered := make([]Memory, 0)
|
||||
for _, m := range memories {
|
||||
if m.SourceID == q.SourceID {
|
||||
filtered = append(filtered, m)
|
||||
}
|
||||
}
|
||||
return filtered, nil
|
||||
}
|
||||
|
||||
return memories, nil
|
||||
}
|
||||
|
||||
// ApplyDecay reduces the decay factor based on time elapsed since last access.
|
||||
func (s *Service) ApplyDecay(ctx context.Context, userID string) (int, error) {
|
||||
memories, err := s.repo.ListByUser(ctx, userID, 100)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("list for decay: %w", err)
|
||||
}
|
||||
|
||||
updated := 0
|
||||
for _, m := range memories {
|
||||
elapsed := time.Since(m.LastAcc)
|
||||
days := elapsed.Hours() / 24
|
||||
newDecay := math.Exp(-0.01 * days)
|
||||
|
||||
if newDecay != m.Decay {
|
||||
m.Decay = newDecay
|
||||
if err := s.repo.Update(ctx, &m); err == nil {
|
||||
updated++
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Purge forgotten memories
|
||||
if _, err := s.repo.PurgeDecayed(ctx, userID); err != nil {
|
||||
return updated, err
|
||||
}
|
||||
|
||||
return updated, nil
|
||||
}
|
||||
|
||||
// Delete removes a memory (admin use).
|
||||
func (s *Service) Delete(ctx context.Context, id string) error {
|
||||
return s.repo.Delete(ctx, id)
|
||||
}
|
||||
|
||||
// List returns recent memories for a user.
|
||||
func (s *Service) List(ctx context.Context, userID string, limit int) ([]Memory, error) {
|
||||
return s.repo.ListByUser(ctx, userID, limit)
|
||||
}
|
||||
|
||||
// GetByID returns a single memory.
|
||||
func (s *Service) GetByID(ctx context.Context, id string) (*Memory, error) {
|
||||
return s.repo.GetByID(ctx, id)
|
||||
}
|
||||
|
||||
// --- Internal helpers ---
|
||||
|
||||
// jaccardSimilarity computes the Jaccard similarity between two strings.
|
||||
func jaccardSimilarity(a, b string) float64 {
|
||||
setA := tokenSet(a)
|
||||
setB := tokenSet(b)
|
||||
|
||||
intersection := 0
|
||||
for token := range setA {
|
||||
if setB[token] {
|
||||
intersection++
|
||||
}
|
||||
}
|
||||
union := len(setA) + len(setB) - intersection
|
||||
if union == 0 {
|
||||
return 0
|
||||
}
|
||||
return float64(intersection) / float64(union)
|
||||
}
|
||||
|
||||
func tokenSet(s string) map[string]bool {
|
||||
// Use 2-gram tokens for Chinese and word tokens for English
|
||||
tokens := make(map[string]bool)
|
||||
runes := []rune(s)
|
||||
for i := 0; i < len(runes)-1; i++ {
|
||||
tokens[string(runes[i:i+2])] = true
|
||||
}
|
||||
for _, word := range strings.Fields(s) {
|
||||
if len(word) > 1 {
|
||||
tokens[word] = true
|
||||
}
|
||||
}
|
||||
return tokens
|
||||
}
|
||||
|
||||
// extractKeywords extracts simple keywords from text.
|
||||
func extractKeywords(content string) []string {
|
||||
// Simple extraction: split by common delimiters, filter short words
|
||||
words := strings.FieldsFunc(content, func(r rune) bool {
|
||||
return r == '，' || r == '。' || r == '、' || r == ' ' || r == '\n' || r == '！' || r == '？'
|
||||
})
|
||||
|
||||
seen := make(map[string]bool)
|
||||
var keywords []string
|
||||
for _, w := range words {
|
||||
runes := []rune(w)
|
||||
if len(runes) >= 2 && len(runes) <= 8 && !seen[w] {
|
||||
keywords = append(keywords, w)
|
||||
seen[w] = true
|
||||
}
|
||||
if len(keywords) >= 10 {
|
||||
break
|
||||
}
|
||||
}
|
||||
return keywords
|
||||
}
|
||||
|
||||
func mergeContent(old, new string) string {
|
||||
if len(new) > len(old) {
|
||||
return new
|
||||
}
|
||||
return old
|
||||
}
|
||||
|
||||
func truncate(s string, maxLen int) string {
|
||||
runes := []rune(s)
|
||||
if len(runes) <= maxLen {
|
||||
return s
|
||||
}
|
||||
return string(runes[:maxLen]) + "..."
|
||||
}
|
||||
Reference in new issue
Block a user