feat: add long-term memory systems with fused retrieval and memory explorer
Phase 5 (ADR-0003). pkg/concepts fixes the epistemic vocabulary (observation, fact, memory, claim, hypothesis, belief, prediction, assumption, inference, opinion), memory kinds and relation names. internal/memory stores episodic, semantic, procedural and autobiographical memories in one mind-scoped table with trust-derived credibility, importance, salience and decaying activation. Retrieval fuses full-text rank (any-term by default), optional pgvector similarity, recency, activation, importance and credibility, and returns the component scores so every recall can be explained. Decay lowers activation and never deletes; retraction and supersession keep rows for provenance. Relationships are constrained to one mind at insert time. Embeddings live in a conditional pgvector table keyed by model so vector spaces are never mixed; without an embedder the system degrades to text and graph retrieval. Attention gates encoding: the attend handler turns admitted observations into episodic memories (operator changes become autobiographical ones) with memory_created events; the retrieve process proposes recalled memories as candidates and admitted recalls are reinforced with memory_retrieved events carrying their scoring components. Memory explorer UI (search with component breakdown, timeline, filters, operator knowledge entry), memory detail with provenance, relationships, citing events and an inaccurate-memory correction; API; noemactl memory commands; decay and embed-missing background jobs.
This commit is contained in:
Andy Dixon
@@ -0,0 +1,476 @@
|
||||
// Package memory implements long-term memory: episodic, semantic, procedural
|
||||
// and autobiographical records with explicit epistemic status, credibility,
|
||||
// importance and decaying activation, plus deterministic fused retrieval
|
||||
// (ADR-0003). Nothing here is deleted by decay.
|
||||
package memory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"math"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"github.com/andydixon/noema/internal/embeddings"
|
||||
"github.com/andydixon/noema/internal/storage"
|
||||
"github.com/andydixon/noema/pkg/concepts"
|
||||
"github.com/andydixon/noema/pkg/events"
|
||||
"github.com/andydixon/noema/pkg/ids"
|
||||
)
|
||||
|
||||
// Status of a memory record. Retracted and superseded memories stay
|
||||
// retrievable for provenance but are excluded from normal recall.
|
||||
type Status string
|
||||
|
||||
// Statuses.
|
||||
const (
|
||||
StatusActive Status = "active"
|
||||
StatusRetracted Status = "retracted"
|
||||
StatusSuperseded Status = "superseded"
|
||||
StatusArchived Status = "archived"
|
||||
)
|
||||
|
||||
// Memory is one long-term record.
|
||||
type Memory struct {
|
||||
ID ids.ID `json:"id"`
|
||||
MindID ids.ID `json:"mind_id"`
|
||||
Kind concepts.MemoryKind `json:"kind"`
|
||||
Epistemic concepts.Epistemic `json:"epistemic"`
|
||||
Status Status `json:"status"`
|
||||
Content string `json:"content"`
|
||||
Detail map[string]any `json:"detail"`
|
||||
Source string `json:"source"`
|
||||
Trust events.Trust `json:"trust"`
|
||||
Credibility float64 `json:"credibility"`
|
||||
Confidence float64 `json:"confidence"`
|
||||
Importance float64 `json:"importance"`
|
||||
Salience float64 `json:"salience"`
|
||||
Activation float64 `json:"activation"`
|
||||
Valence *float64 `json:"valence,omitempty"`
|
||||
Arousal *float64 `json:"arousal,omitempty"`
|
||||
OccurredAt time.Time `json:"occurred_at"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
RetrievalCount int `json:"retrieval_count"`
|
||||
LastRetrievedAt *time.Time `json:"last_retrieved_at,omitempty"`
|
||||
OriginEvent ids.ID `json:"origin_event,omitempty"`
|
||||
Supersedes ids.ID `json:"supersedes,omitempty"`
|
||||
}
|
||||
|
||||
// New describes a memory to store.
|
||||
type New struct {
|
||||
MindID ids.ID
|
||||
Kind concepts.MemoryKind
|
||||
Epistemic concepts.Epistemic
|
||||
Content string
|
||||
Detail map[string]any
|
||||
Source string
|
||||
Trust events.Trust
|
||||
Confidence float64
|
||||
Importance float64
|
||||
Salience float64
|
||||
Valence *float64
|
||||
Arousal *float64
|
||||
OccurredAt time.Time
|
||||
OriginEvent ids.ID
|
||||
Supersedes ids.ID
|
||||
}
|
||||
|
||||
// Relationship is an edge between two memories.
|
||||
type Relationship struct {
|
||||
ID ids.ID `json:"id"`
|
||||
FromMemory ids.ID `json:"from_memory"`
|
||||
ToMemory ids.ID `json:"to_memory"`
|
||||
Relation concepts.Relation `json:"relation"`
|
||||
Confidence float64 `json:"confidence"`
|
||||
Source string `json:"source"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
}
|
||||
|
||||
// ErrInvalid marks bad input.
|
||||
var ErrInvalid = errors.New("invalid memory")
|
||||
|
||||
// Store persists memories.
|
||||
type Store struct {
|
||||
db *pgxpool.Pool
|
||||
embedder embeddings.Embedder
|
||||
vectors bool // memory_embeddings exists (pgvector available)
|
||||
}
|
||||
|
||||
// NewStore creates a store and detects vector support.
|
||||
func NewStore(ctx context.Context, db *pgxpool.Pool, embedder embeddings.Embedder) *Store {
|
||||
s := &Store{db: db, embedder: embedder}
|
||||
if embedder == nil {
|
||||
s.embedder = embeddings.None{}
|
||||
}
|
||||
db.QueryRow(ctx, `SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'memory_embeddings')`).Scan(&s.vectors) //nolint:errcheck
|
||||
return s
|
||||
}
|
||||
|
||||
// VectorsEnabled reports whether pgvector storage is present.
|
||||
func (s *Store) VectorsEnabled() bool { return s.vectors }
|
||||
|
||||
// SemanticEnabled reports whether semantic (embedding) retrieval can run.
|
||||
func (s *Store) SemanticEnabled() bool { return s.vectors && s.embedder.Model() != "" }
|
||||
|
||||
// SetEmbedder swaps the embedder (providers are configured at runtime).
|
||||
func (s *Store) SetEmbedder(e embeddings.Embedder) {
|
||||
if e == nil {
|
||||
e = embeddings.None{}
|
||||
}
|
||||
s.embedder = e
|
||||
}
|
||||
|
||||
func (n *New) validate() error {
|
||||
if n.MindID.IsZero() {
|
||||
return fmt.Errorf("%w: mind id required", ErrInvalid)
|
||||
}
|
||||
if !n.Kind.Valid() {
|
||||
return fmt.Errorf("%w: unknown kind %q", ErrInvalid, n.Kind)
|
||||
}
|
||||
if !n.Epistemic.Valid() {
|
||||
return fmt.Errorf("%w: unknown epistemic kind %q", ErrInvalid, n.Epistemic)
|
||||
}
|
||||
n.Content = strings.TrimSpace(n.Content)
|
||||
if n.Content == "" || len(n.Content) > 20000 {
|
||||
return fmt.Errorf("%w: content required (max 20000 chars)", ErrInvalid)
|
||||
}
|
||||
if !n.Trust.Valid() {
|
||||
n.Trust = events.TrustLow
|
||||
}
|
||||
n.Confidence, n.Importance, n.Salience = clamp(n.Confidence), clamp(n.Importance), clamp(n.Salience)
|
||||
if n.OccurredAt.IsZero() {
|
||||
n.OccurredAt = time.Now()
|
||||
}
|
||||
if n.Detail == nil {
|
||||
n.Detail = map[string]any{}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func clamp(v float64) float64 {
|
||||
if math.IsNaN(v) || v < 0 {
|
||||
return 0
|
||||
}
|
||||
if v > 1 {
|
||||
return 1
|
||||
}
|
||||
return v
|
||||
}
|
||||
|
||||
// Create stores a memory using q (a pool or transaction). Embedding happens
|
||||
// best-effort after the row exists so a slow provider never blocks encoding.
|
||||
func (s *Store) Create(ctx context.Context, q storage.Querier, n New) (Memory, error) {
|
||||
if err := n.validate(); err != nil {
|
||||
return Memory{}, err
|
||||
}
|
||||
m := Memory{ID: ids.New(), MindID: n.MindID, Kind: n.Kind, Epistemic: n.Epistemic, Status: StatusActive, Content: n.Content, Detail: n.Detail, Source: n.Source,
|
||||
Trust: n.Trust, Credibility: n.Trust.Credibility(), Confidence: n.Confidence, Importance: n.Importance, Salience: n.Salience, Activation: 1,
|
||||
Valence: n.Valence, Arousal: n.Arousal, OccurredAt: n.OccurredAt, CreatedAt: time.Now(), UpdatedAt: time.Now(), OriginEvent: n.OriginEvent, Supersedes: n.Supersedes}
|
||||
detail, _ := json.Marshal(n.Detail)
|
||||
_, err := q.Exec(ctx, `INSERT INTO memories (id, mind_id, kind, epistemic, content, detail, source, trust, credibility, confidence, importance, salience, valence, arousal, occurred_at, origin_event, supersedes)
|
||||
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17)`,
|
||||
m.ID, m.MindID, string(m.Kind), string(m.Epistemic), m.Content, detail, m.Source, string(m.Trust), m.Credibility, m.Confidence, m.Importance, m.Salience, m.Valence, m.Arousal, m.OccurredAt, nullID(m.OriginEvent), nullID(m.Supersedes))
|
||||
if err != nil {
|
||||
return Memory{}, storage.Translate(err)
|
||||
}
|
||||
if !n.Supersedes.IsZero() {
|
||||
q.Exec(ctx, `UPDATE memories SET status='superseded', updated_at=now() WHERE id=$1 AND mind_id=$2`, n.Supersedes, n.MindID) //nolint:errcheck
|
||||
}
|
||||
return m, nil
|
||||
}
|
||||
|
||||
// Embed computes and stores the embedding for a memory if a provider exists.
|
||||
// Safe to call from a background task; errors are returned for logging only.
|
||||
func (s *Store) Embed(ctx context.Context, m Memory) error {
|
||||
if !s.SemanticEnabled() {
|
||||
return nil
|
||||
}
|
||||
vecs, err := s.embedder.Embed(ctx, []string{m.Content})
|
||||
if err != nil || len(vecs) != 1 {
|
||||
return fmt.Errorf("embed: %w", err)
|
||||
}
|
||||
_, err = s.db.Exec(ctx, `INSERT INTO memory_embeddings (memory_id, model, dims, embedding) VALUES ($1,$2,$3,$4::vector)
|
||||
ON CONFLICT (memory_id, model) DO UPDATE SET embedding=excluded.embedding, dims=excluded.dims, created_at=now()`, m.ID, s.embedder.Model(), len(vecs[0]), embeddings.Format(vecs[0]))
|
||||
return err
|
||||
}
|
||||
|
||||
// EmbedMissing embeds up to limit memories lacking a vector for the current model.
|
||||
func (s *Store) EmbedMissing(ctx context.Context, limit int) (int, error) {
|
||||
if !s.SemanticEnabled() {
|
||||
return 0, nil
|
||||
}
|
||||
rows, err := s.db.Query(ctx, `SELECT `+cols+` FROM memories m WHERE status='active' AND NOT EXISTS (SELECT 1 FROM memory_embeddings e WHERE e.memory_id=m.id AND e.model=$1) ORDER BY created_at DESC LIMIT $2`, s.embedder.Model(), limit)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
var list []Memory
|
||||
for rows.Next() {
|
||||
m, err := scan(rows)
|
||||
if err != nil {
|
||||
rows.Close()
|
||||
return 0, err
|
||||
}
|
||||
list = append(list, m)
|
||||
}
|
||||
rows.Close()
|
||||
n := 0
|
||||
for _, m := range list {
|
||||
if err := s.Embed(ctx, m); err != nil {
|
||||
return n, err
|
||||
}
|
||||
n++
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
|
||||
func nullID(id ids.ID) *string {
|
||||
if id.IsZero() {
|
||||
return nil
|
||||
}
|
||||
s := string(id)
|
||||
return &s
|
||||
}
|
||||
|
||||
const cols = `m.id, m.mind_id, m.kind, m.epistemic, m.status, m.content, m.detail, m.source, m.trust, m.credibility, m.confidence, m.importance, m.salience, m.activation, m.valence, m.arousal, m.occurred_at, m.created_at, m.updated_at, m.retrieval_count, m.last_retrieved_at, m.origin_event, m.supersedes`
|
||||
|
||||
func scan(row pgx.Row) (Memory, error) {
|
||||
var m Memory
|
||||
var kind, epi, status, trust string
|
||||
var detail []byte
|
||||
var origin, supersedes *string
|
||||
err := row.Scan(&m.ID, &m.MindID, &kind, &epi, &status, &m.Content, &detail, &m.Source, &trust, &m.Credibility, &m.Confidence, &m.Importance, &m.Salience, &m.Activation,
|
||||
&m.Valence, &m.Arousal, &m.OccurredAt, &m.CreatedAt, &m.UpdatedAt, &m.RetrievalCount, &m.LastRetrievedAt, &origin, &supersedes)
|
||||
if err != nil {
|
||||
return m, storage.Translate(err)
|
||||
}
|
||||
m.Kind, m.Epistemic, m.Status, m.Trust = concepts.MemoryKind(kind), concepts.Epistemic(epi), Status(status), events.Trust(trust)
|
||||
json.Unmarshal(detail, &m.Detail) //nolint:errcheck
|
||||
if m.Detail == nil {
|
||||
m.Detail = map[string]any{}
|
||||
}
|
||||
if origin != nil {
|
||||
m.OriginEvent = ids.ID(*origin)
|
||||
}
|
||||
if supersedes != nil {
|
||||
m.Supersedes = ids.ID(*supersedes)
|
||||
}
|
||||
return m, nil
|
||||
}
|
||||
|
||||
// Get fetches a memory scoped to a mind (cross-mind reads are impossible by construction).
|
||||
func (s *Store) Get(ctx context.Context, mind, id ids.ID) (Memory, error) {
|
||||
return scan(s.db.QueryRow(ctx, `SELECT `+cols+` FROM memories m WHERE m.id=$1 AND m.mind_id=$2`, id, mind))
|
||||
}
|
||||
|
||||
// Filter selects memories for listing.
|
||||
type Filter struct {
|
||||
Kinds []concepts.MemoryKind
|
||||
Epistemic []concepts.Epistemic
|
||||
Statuses []Status // default: active
|
||||
MinConfidence float64
|
||||
MinImportance float64
|
||||
Source string
|
||||
Since, Until time.Time
|
||||
Search string // plain text search (websearch syntax)
|
||||
Before time.Time
|
||||
Limit int
|
||||
}
|
||||
|
||||
// List returns memories newest first (by occurred_at) with keyset paging on Before.
|
||||
func (s *Store) List(ctx context.Context, mind ids.ID, f Filter) ([]Memory, error) {
|
||||
where := []string{"m.mind_id = $1"}
|
||||
args := []any{mind}
|
||||
arg := func(v any) string { args = append(args, v); return fmt.Sprintf("$%d", len(args)) }
|
||||
if len(f.Kinds) > 0 {
|
||||
where = append(where, "m.kind = ANY("+arg(strs(f.Kinds))+")")
|
||||
}
|
||||
if len(f.Epistemic) > 0 {
|
||||
where = append(where, "m.epistemic = ANY("+arg(strs(f.Epistemic))+")")
|
||||
}
|
||||
if len(f.Statuses) > 0 {
|
||||
where = append(where, "m.status = ANY("+arg(strs(f.Statuses))+")")
|
||||
} else {
|
||||
where = append(where, "m.status = 'active'")
|
||||
}
|
||||
if f.MinConfidence > 0 {
|
||||
where = append(where, "m.confidence >= "+arg(f.MinConfidence))
|
||||
}
|
||||
if f.MinImportance > 0 {
|
||||
where = append(where, "m.importance >= "+arg(f.MinImportance))
|
||||
}
|
||||
if f.Source != "" {
|
||||
where = append(where, "m.source = "+arg(f.Source))
|
||||
}
|
||||
if !f.Since.IsZero() {
|
||||
where = append(where, "m.occurred_at >= "+arg(f.Since))
|
||||
}
|
||||
if !f.Until.IsZero() {
|
||||
where = append(where, "m.occurred_at <= "+arg(f.Until))
|
||||
}
|
||||
if !f.Before.IsZero() {
|
||||
where = append(where, "m.occurred_at < "+arg(f.Before))
|
||||
}
|
||||
if q := strings.TrimSpace(f.Search); q != "" {
|
||||
where = append(where, "m.search @@ websearch_to_tsquery('english', "+arg(q)+")")
|
||||
}
|
||||
limit := f.Limit
|
||||
if limit <= 0 || limit > 500 {
|
||||
limit = 100
|
||||
}
|
||||
rows, err := s.db.Query(ctx, `SELECT `+cols+` FROM memories m WHERE `+strings.Join(where, " AND ")+` ORDER BY m.occurred_at DESC, m.id DESC LIMIT `+arg(limit), args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []Memory
|
||||
for rows.Next() {
|
||||
m, err := scan(rows)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, m)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func strs[T ~string](xs []T) []string {
|
||||
out := make([]string, len(xs))
|
||||
for i, x := range xs {
|
||||
out[i] = string(x)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// SetStatus changes status (retract, archive). It never deletes.
|
||||
func (s *Store) SetStatus(ctx context.Context, mind, id ids.ID, st Status) error {
|
||||
tag, err := s.db.Exec(ctx, `UPDATE memories SET status=$3, updated_at=now() WHERE id=$1 AND mind_id=$2`, id, mind, string(st))
|
||||
if err == nil && tag.RowsAffected() == 0 {
|
||||
return storage.ErrNotFound
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// UpdateConfidence sets a new confidence (used by consolidation and corrections).
|
||||
func (s *Store) UpdateConfidence(ctx context.Context, q storage.Querier, mind, id ids.ID, confidence float64) error {
|
||||
tag, err := q.Exec(ctx, `UPDATE memories SET confidence=$3, updated_at=now() WHERE id=$1 AND mind_id=$2`, id, mind, clamp(confidence))
|
||||
if err == nil && tag.RowsAffected() == 0 {
|
||||
return storage.ErrNotFound
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// Reinforce records a retrieval: count, timestamp and activation towards 1.
|
||||
func (s *Store) Reinforce(ctx context.Context, q storage.Querier, mind, id ids.ID) error {
|
||||
_, err := q.Exec(ctx, `UPDATE memories SET retrieval_count = retrieval_count + 1, last_retrieved_at = now(), activation = LEAST(1, activation + (1 - activation) * 0.5), updated_at = now() WHERE id=$1 AND mind_id=$2`, id, mind)
|
||||
return err
|
||||
}
|
||||
|
||||
// Decay lowers activation of memories not retrieved recently. Importance slows
|
||||
// decay; nothing is deleted. Returns the number of rows touched.
|
||||
func (s *Store) Decay(ctx context.Context, mind ids.ID, halfLife time.Duration) (int64, error) {
|
||||
if halfLife <= 0 {
|
||||
halfLife = 14 * 24 * time.Hour
|
||||
}
|
||||
// activation *= 0.5 ^ (age / (halfLife * (1 + importance)))
|
||||
tag, err := s.db.Exec(ctx, `UPDATE memories SET activation = GREATEST(0.01, activation * power(0.5, EXTRACT(EPOCH FROM (now() - coalesce(last_retrieved_at, created_at))) / ($2 * (1 + importance)))), updated_at = now()
|
||||
WHERE mind_id=$1 AND status='active' AND activation > 0.01 AND coalesce(last_retrieved_at, created_at) < now() - interval '1 hour'`, mind, halfLife.Seconds())
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return tag.RowsAffected(), nil
|
||||
}
|
||||
|
||||
// Relate adds an edge between two memories of the same mind.
|
||||
func (s *Store) Relate(ctx context.Context, q storage.Querier, mind, from, to ids.ID, rel concepts.Relation, confidence float64, source string) (Relationship, error) {
|
||||
if !rel.Valid() {
|
||||
return Relationship{}, fmt.Errorf("%w: unknown relation %q", ErrInvalid, rel)
|
||||
}
|
||||
if from == to {
|
||||
return Relationship{}, fmt.Errorf("%w: self relation", ErrInvalid)
|
||||
}
|
||||
r := Relationship{ID: ids.New(), FromMemory: from, ToMemory: to, Relation: rel, Confidence: clamp(confidence), Source: source, CreatedAt: time.Now()}
|
||||
// Both endpoints must belong to the mind; the subquery enforces it.
|
||||
tag, err := q.Exec(ctx, `INSERT INTO memory_relationships (id, mind_id, from_memory, to_memory, relation, confidence, source)
|
||||
SELECT $1, $2, $3, $4, $5, $6, $7 WHERE (SELECT count(*) FROM memories WHERE id IN ($3,$4) AND mind_id=$2) = 2
|
||||
ON CONFLICT (from_memory, to_memory, relation) DO UPDATE SET confidence = GREATEST(memory_relationships.confidence, excluded.confidence)`,
|
||||
r.ID, mind, from, to, string(rel), r.Confidence, source)
|
||||
if err != nil {
|
||||
return Relationship{}, err
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return Relationship{}, fmt.Errorf("%w: memories not found in mind", ErrInvalid)
|
||||
}
|
||||
return r, nil
|
||||
}
|
||||
|
||||
// Related lists edges touching a memory, with the other endpoint loaded.
|
||||
type Related struct {
|
||||
Relationship
|
||||
Other Memory `json:"other"`
|
||||
Direction string `json:"direction"` // out | in
|
||||
}
|
||||
|
||||
// Relationships returns edges in both directions.
|
||||
func (s *Store) Relationships(ctx context.Context, mind, id ids.ID) ([]Related, error) {
|
||||
rows, err := s.db.Query(ctx, `SELECT r.id, r.from_memory, r.to_memory, r.relation, r.confidence, r.source, r.created_at, CASE WHEN r.from_memory=$1 THEN 'out' ELSE 'in' END, `+cols+`
|
||||
FROM memory_relationships r JOIN memories m ON m.id = CASE WHEN r.from_memory=$1 THEN r.to_memory ELSE r.from_memory END
|
||||
WHERE r.mind_id=$2 AND (r.from_memory=$1 OR r.to_memory=$1) ORDER BY r.created_at`, id, mind)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []Related
|
||||
for rows.Next() {
|
||||
var r Related
|
||||
var rel string
|
||||
var kind, epi, status, trust string
|
||||
var detail []byte
|
||||
var origin, supersedes *string
|
||||
m := &r.Other
|
||||
if err := rows.Scan(&r.ID, &r.FromMemory, &r.ToMemory, &rel, &r.Confidence, &r.Source, &r.CreatedAt, &r.Direction,
|
||||
&m.ID, &m.MindID, &kind, &epi, &status, &m.Content, &detail, &m.Source, &trust, &m.Credibility, &m.Confidence, &m.Importance, &m.Salience, &m.Activation,
|
||||
&m.Valence, &m.Arousal, &m.OccurredAt, &m.CreatedAt, &m.UpdatedAt, &m.RetrievalCount, &m.LastRetrievedAt, &origin, &supersedes); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
r.Relation = concepts.Relation(rel)
|
||||
m.Kind, m.Epistemic, m.Status, m.Trust = concepts.MemoryKind(kind), concepts.Epistemic(epi), Status(status), events.Trust(trust)
|
||||
json.Unmarshal(detail, &m.Detail) //nolint:errcheck
|
||||
if origin != nil {
|
||||
m.OriginEvent = ids.ID(*origin)
|
||||
}
|
||||
out = append(out, r)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// Stats summarises a mind's memory for overview pages.
|
||||
type Stats struct {
|
||||
Total int64 `json:"total"`
|
||||
ByKind map[string]int64 `json:"by_kind"`
|
||||
}
|
||||
|
||||
// Stats counts active memories by kind.
|
||||
func (s *Store) Stats(ctx context.Context, mind ids.ID) (Stats, error) {
|
||||
st := Stats{ByKind: map[string]int64{}}
|
||||
rows, err := s.db.Query(ctx, `SELECT kind, count(*) FROM memories WHERE mind_id=$1 AND status='active' GROUP BY kind`, mind)
|
||||
if err != nil {
|
||||
return st, err
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var k string
|
||||
var n int64
|
||||
if err := rows.Scan(&k, &n); err != nil {
|
||||
return st, err
|
||||
}
|
||||
st.ByKind[k] = n
|
||||
st.Total += n
|
||||
}
|
||||
return st, rows.Err()
|
||||
}
|
||||
@@ -0,0 +1,108 @@
|
||||
package memory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/andydixon/noema/internal/minds"
|
||||
"github.com/andydixon/noema/internal/storage"
|
||||
"github.com/andydixon/noema/migrations"
|
||||
"github.com/andydixon/noema/pkg/concepts"
|
||||
"github.com/andydixon/noema/pkg/events"
|
||||
)
|
||||
|
||||
func TestMemoryLifecycleRetrievalAndIsolation(t *testing.T) {
|
||||
pool := storage.TestPool(t, migrations.FS)
|
||||
ctx := context.Background()
|
||||
ms := minds.NewStore(pool)
|
||||
a, _ := ms.Create(ctx, minds.CreateParams{Slug: "a"})
|
||||
b, _ := ms.Create(ctx, minds.CreateParams{Slug: "b"})
|
||||
s := NewStore(ctx, pool, nil)
|
||||
if !s.VectorsEnabled() {
|
||||
t.Log("pgvector not available in this database; vector path skipped")
|
||||
}
|
||||
|
||||
mk := func(mind minds.Mind, kind concepts.MemoryKind, content string, importance float64, when time.Time) Memory {
|
||||
m, err := s.Create(ctx, pool, New{MindID: mind.ID, Kind: kind, Epistemic: concepts.Observation, Content: content, Source: "test", Trust: events.TrustUser, Confidence: 0.9, Importance: importance, OccurredAt: when})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return m
|
||||
}
|
||||
now := time.Now()
|
||||
e1 := mk(a, concepts.Episodic, "Andy asked for the atlas server to be checked after a latency spike", 0.7, now.Add(-48*time.Hour))
|
||||
e2 := mk(a, concepts.Episodic, "The atlas disk showed SMART CRC errors during the incident", 0.8, now.Add(-24*time.Hour))
|
||||
mk(a, concepts.Semantic, "Andy prefers Go for deployed services", 0.6, now.Add(-30*24*time.Hour))
|
||||
mk(b, concepts.Episodic, "atlas latency spike discussed with a different operator", 0.9, now)
|
||||
|
||||
if _, err := s.Create(ctx, pool, New{MindID: a.ID, Kind: "weird", Epistemic: concepts.Claim, Content: "x"}); err == nil {
|
||||
t.Fatal("invalid kind accepted")
|
||||
}
|
||||
|
||||
got, err := s.Retrieve(ctx, a.ID, Query{Text: "atlas latency", Limit: 5})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(got) != 2 || got[0].Memory.ID != e1.ID {
|
||||
t.Fatalf("retrieval wrong: %+v", got)
|
||||
}
|
||||
if got[0].Components["text"] == 0 || got[0].Components["recency"] == 0 || got[0].Score <= 0 || got[0].Score > 1 {
|
||||
t.Fatalf("components missing: %+v", got[0])
|
||||
}
|
||||
for _, r := range got {
|
||||
if r.Memory.MindID != a.ID {
|
||||
t.Fatal("cross-mind leakage in retrieval")
|
||||
}
|
||||
}
|
||||
// Mind b sees only its own memory, and cannot Get a's memory.
|
||||
gotB, _ := s.Retrieve(ctx, b.ID, Query{Text: "atlas", Limit: 5})
|
||||
if len(gotB) != 1 || gotB[0].Memory.MindID != b.ID {
|
||||
t.Fatalf("isolation broken: %+v", gotB)
|
||||
}
|
||||
if _, err := s.Get(ctx, b.ID, e1.ID); err == nil {
|
||||
t.Fatal("cross-mind Get succeeded")
|
||||
}
|
||||
// Reinforcement and decay never delete.
|
||||
if err := s.Reinforce(ctx, pool, a.ID, e2.ID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
m, _ := s.Get(ctx, a.ID, e2.ID)
|
||||
if m.RetrievalCount != 1 || m.LastRetrievedAt == nil {
|
||||
t.Fatalf("reinforce wrong: %+v", m)
|
||||
}
|
||||
pool.Exec(ctx, `UPDATE memories SET created_at = now() - interval '60 days', last_retrieved_at = NULL WHERE id=$1`, e1.ID) //nolint:errcheck
|
||||
if _, err := s.Decay(ctx, a.ID, 14*24*time.Hour); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
m1, _ := s.Get(ctx, a.ID, e1.ID)
|
||||
if m1.Activation >= 1 || m1.Activation < 0.01 || m1.Status != StatusActive {
|
||||
t.Fatalf("decay wrong: %+v", m1)
|
||||
}
|
||||
// Relationships are mind-scoped.
|
||||
if _, err := s.Relate(ctx, pool, a.ID, e1.ID, e2.ID, concepts.RelatedTo, 0.8, "test"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := s.Relate(ctx, pool, a.ID, e1.ID, gotB[0].Memory.ID, concepts.RelatedTo, 0.8, "test"); err == nil {
|
||||
t.Fatal("cross-mind relationship accepted")
|
||||
}
|
||||
rel, _ := s.Relationships(ctx, a.ID, e2.ID)
|
||||
if len(rel) != 1 || rel[0].Direction != "in" || rel[0].Other.ID != e1.ID {
|
||||
t.Fatalf("relationships wrong: %+v", rel)
|
||||
}
|
||||
// Retraction keeps the row.
|
||||
if err := s.SetStatus(ctx, a.ID, e2.ID, StatusRetracted); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
list, _ := s.List(ctx, a.ID, Filter{})
|
||||
if len(list) != 2 {
|
||||
t.Fatalf("retracted memory still listed as active: %d", len(list))
|
||||
}
|
||||
if _, err := s.Get(ctx, a.ID, e2.ID); err != nil {
|
||||
t.Fatal("retracted memory must remain readable")
|
||||
}
|
||||
st, _ := s.Stats(ctx, a.ID)
|
||||
if st.Total != 2 || st.ByKind["semantic"] != 1 {
|
||||
t.Fatalf("stats wrong: %+v", st)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,242 @@
|
||||
package memory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"math"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/andydixon/noema/internal/embeddings"
|
||||
"github.com/andydixon/noema/pkg/concepts"
|
||||
"github.com/andydixon/noema/pkg/events"
|
||||
"github.com/andydixon/noema/pkg/ids"
|
||||
)
|
||||
|
||||
// Retrieval is one recalled memory with the arithmetic behind its score.
|
||||
type Retrieval struct {
|
||||
Memory Memory `json:"memory"`
|
||||
Score float64 `json:"score"`
|
||||
Components map[string]float64 `json:"components"` // text, vector, recency, activation, importance, credibility
|
||||
}
|
||||
|
||||
// Query describes what to recall.
|
||||
type Query struct {
|
||||
Text string
|
||||
Kinds []concepts.MemoryKind
|
||||
Limit int
|
||||
Now time.Time
|
||||
Semantic bool // use embeddings when available
|
||||
Strict bool // websearch syntax with AND semantics; default matches any term
|
||||
}
|
||||
|
||||
// Weights for fusion. When vectors are unavailable the vector weight is
|
||||
// redistributed to text so totals stay comparable.
|
||||
var weights = map[string]float64{"text": 0.40, "vector": 0.25, "recency": 0.10, "activation": 0.10, "importance": 0.10, "credibility": 0.05}
|
||||
|
||||
// Retrieve performs fused retrieval: full-text rank, optional vector
|
||||
// similarity, recency, activation, importance and credibility, each recorded.
|
||||
func (s *Store) Retrieve(ctx context.Context, mind ids.ID, q Query) ([]Retrieval, error) {
|
||||
if q.Limit <= 0 || q.Limit > 50 {
|
||||
q.Limit = 10
|
||||
}
|
||||
if q.Now.IsZero() {
|
||||
q.Now = time.Now()
|
||||
}
|
||||
text := strings.TrimSpace(q.Text)
|
||||
cands := map[ids.ID]*Retrieval{}
|
||||
add := func(m Memory) *Retrieval {
|
||||
r, ok := cands[m.ID]
|
||||
if !ok {
|
||||
r = &Retrieval{Memory: m, Components: map[string]float64{}}
|
||||
cands[m.ID] = r
|
||||
}
|
||||
return r
|
||||
}
|
||||
kindFilter := "TRUE"
|
||||
args := []any{mind}
|
||||
if len(q.Kinds) > 0 {
|
||||
kindFilter = "m.kind = ANY($2)"
|
||||
args = append(args, strs(q.Kinds))
|
||||
}
|
||||
// Full-text candidates.
|
||||
if text != "" {
|
||||
tsq := "websearch_to_tsquery('english', $%d)"
|
||||
queryText := text
|
||||
if !q.Strict {
|
||||
tsq = "to_tsquery('english', $%d)"
|
||||
queryText = anyTermsQuery(text)
|
||||
}
|
||||
targs := append(append([]any{}, args...), queryText)
|
||||
tsq = strings.ReplaceAll(tsq, "%d", itoa(len(targs)))
|
||||
rows, err := s.db.Query(ctx, `SELECT `+cols+`, ts_rank_cd(m.search, `+tsq+`) AS rank
|
||||
FROM memories m WHERE m.mind_id=$1 AND m.status='active' AND `+kindFilter+` AND m.search @@ `+tsq+`
|
||||
ORDER BY rank DESC LIMIT 50`, targs...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var maxRank float64
|
||||
type ranked struct {
|
||||
m Memory
|
||||
rank float64
|
||||
}
|
||||
var list []ranked
|
||||
for rows.Next() {
|
||||
var m Memory
|
||||
var rank float64
|
||||
var kind, epi, status, trust string
|
||||
var detail []byte
|
||||
var origin, supersedes *string
|
||||
if err := rows.Scan(&m.ID, &m.MindID, &kind, &epi, &status, &m.Content, &detail, &m.Source, &trust, &m.Credibility, &m.Confidence, &m.Importance, &m.Salience, &m.Activation,
|
||||
&m.Valence, &m.Arousal, &m.OccurredAt, &m.CreatedAt, &m.UpdatedAt, &m.RetrievalCount, &m.LastRetrievedAt, &origin, &supersedes, &rank); err != nil {
|
||||
rows.Close()
|
||||
return nil, err
|
||||
}
|
||||
m.Kind, m.Epistemic, m.Status, m.Trust = concepts.MemoryKind(kind), concepts.Epistemic(epi), Status(status), events.Trust(trust)
|
||||
if origin != nil {
|
||||
m.OriginEvent = ids.ID(*origin)
|
||||
}
|
||||
list = append(list, ranked{m, rank})
|
||||
if rank > maxRank {
|
||||
maxRank = rank
|
||||
}
|
||||
}
|
||||
rows.Close()
|
||||
for _, r := range list {
|
||||
c := add(r.m)
|
||||
if maxRank > 0 {
|
||||
c.Components["text"] = round4(r.rank / maxRank)
|
||||
}
|
||||
}
|
||||
}
|
||||
// Vector candidates.
|
||||
vectorUsed := false
|
||||
if text != "" && q.Semantic && s.SemanticEnabled() {
|
||||
if vecs, err := s.embedder.Embed(ctx, []string{text}); err == nil && len(vecs) == 1 {
|
||||
vargs := append(append([]any{}, args...), s.embedder.Model(), embeddings.Format(vecs[0]))
|
||||
n := len(vargs)
|
||||
rows, err := s.db.Query(ctx, `SELECT `+cols+`, 1 - (e.embedding <=> $`+itoa(n)+`::vector) AS sim
|
||||
FROM memories m JOIN memory_embeddings e ON e.memory_id = m.id AND e.model = $`+itoa(n-1)+`
|
||||
WHERE m.mind_id=$1 AND m.status='active' AND `+kindFilter+` ORDER BY e.embedding <=> $`+itoa(n)+`::vector LIMIT 50`, vargs...)
|
||||
if err == nil {
|
||||
vectorUsed = true
|
||||
for rows.Next() {
|
||||
var m Memory
|
||||
var sim float64
|
||||
var kind, epi, status, trust string
|
||||
var detail []byte
|
||||
var origin, supersedes *string
|
||||
if err := rows.Scan(&m.ID, &m.MindID, &kind, &epi, &status, &m.Content, &detail, &m.Source, &trust, &m.Credibility, &m.Confidence, &m.Importance, &m.Salience, &m.Activation,
|
||||
&m.Valence, &m.Arousal, &m.OccurredAt, &m.CreatedAt, &m.UpdatedAt, &m.RetrievalCount, &m.LastRetrievedAt, &origin, &supersedes, &sim); err != nil {
|
||||
break
|
||||
}
|
||||
m.Kind, m.Epistemic, m.Status, m.Trust = concepts.MemoryKind(kind), concepts.Epistemic(epi), Status(status), events.Trust(trust)
|
||||
if origin != nil {
|
||||
m.OriginEvent = ids.ID(*origin)
|
||||
}
|
||||
c := add(m)
|
||||
c.Components["vector"] = round4(math.Max(0, sim))
|
||||
}
|
||||
rows.Close()
|
||||
}
|
||||
}
|
||||
}
|
||||
// No text: recall by importance/activation only (e.g. idle association).
|
||||
if text == "" {
|
||||
list, err := s.List(ctx, mind, Filter{Kinds: q.Kinds, Limit: 50})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, m := range list {
|
||||
add(m)
|
||||
}
|
||||
}
|
||||
// Fuse.
|
||||
w := map[string]float64{}
|
||||
for k, v := range weights {
|
||||
w[k] = v
|
||||
}
|
||||
if !vectorUsed {
|
||||
w["text"] += w["vector"]
|
||||
w["vector"] = 0
|
||||
}
|
||||
if text == "" {
|
||||
w["text"], w["vector"] = 0, 0
|
||||
}
|
||||
var sum float64
|
||||
for _, v := range w {
|
||||
sum += v
|
||||
}
|
||||
out := make([]Retrieval, 0, len(cands))
|
||||
for _, r := range cands {
|
||||
m := r.Memory
|
||||
age := q.Now.Sub(m.OccurredAt).Hours() / 24
|
||||
r.Components["recency"] = round4(math.Exp(-age / 30))
|
||||
r.Components["activation"] = round4(m.Activation)
|
||||
r.Components["importance"] = round4(m.Importance)
|
||||
r.Components["credibility"] = round4(m.Credibility)
|
||||
var total float64
|
||||
for k, wk := range w {
|
||||
total += r.Components[k] * wk
|
||||
}
|
||||
r.Score = round4(total / sum)
|
||||
out = append(out, *r)
|
||||
}
|
||||
sort.SliceStable(out, func(i, j int) bool {
|
||||
if out[i].Score != out[j].Score {
|
||||
return out[i].Score > out[j].Score
|
||||
}
|
||||
return out[i].Memory.ID < out[j].Memory.ID
|
||||
})
|
||||
if len(out) > q.Limit {
|
||||
out = out[:q.Limit]
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// anyTermsQuery builds an OR tsquery from the alphanumeric words of text so
|
||||
// recall favours breadth; ranking then orders by how many terms matched.
|
||||
func anyTermsQuery(text string) string {
|
||||
var terms []string
|
||||
seen := map[string]bool{}
|
||||
var cur strings.Builder
|
||||
flush := func() {
|
||||
w := cur.String()
|
||||
cur.Reset()
|
||||
if len(w) >= 2 && !seen[w] {
|
||||
seen[w] = true
|
||||
terms = append(terms, w)
|
||||
}
|
||||
}
|
||||
for _, r := range strings.ToLower(text) {
|
||||
if r >= 'a' && r <= 'z' || r >= '0' && r <= '9' {
|
||||
cur.WriteRune(r)
|
||||
} else {
|
||||
flush()
|
||||
}
|
||||
}
|
||||
flush()
|
||||
if len(terms) > 32 {
|
||||
terms = terms[:32]
|
||||
}
|
||||
if len(terms) == 0 {
|
||||
return "zzzznomatch"
|
||||
}
|
||||
return strings.Join(terms, " | ")
|
||||
}
|
||||
|
||||
func itoa(n int) string {
|
||||
if n == 0 {
|
||||
return "0"
|
||||
}
|
||||
var b [20]byte
|
||||
i := len(b)
|
||||
for n > 0 {
|
||||
i--
|
||||
b[i] = byte('0' + n%10)
|
||||
n /= 10
|
||||
}
|
||||
return string(b[i:])
|
||||
}
|
||||
|
||||
func round4(v float64) float64 { return math.Round(v*10000) / 10000 }
|
||||
Reference in New Issue
Block a user