Changeset 0.30.1 (#200)
Co-authored-by: Jeffrey Smith <jasafpro@gmail.com> Co-committed-by: Jeffrey Smith <jasafpro@gmail.com>
This commit is contained in:
157
server/memory/compactor.go
Normal file
157
server/memory/compactor.go
Normal file
@@ -0,0 +1,157 @@
|
||||
package memory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"strings"
|
||||
|
||||
"git.gobha.me/xcaliber/chat-switchboard/models"
|
||||
"git.gobha.me/xcaliber/chat-switchboard/providers"
|
||||
"git.gobha.me/xcaliber/chat-switchboard/roles"
|
||||
"git.gobha.me/xcaliber/chat-switchboard/store"
|
||||
)
|
||||
|
||||
// compactionThreshold is the minimum number of active memories before
|
||||
// compaction is considered worthwhile for a user.
|
||||
const compactionThreshold = 30
|
||||
|
||||
// compactionPrompt instructs the utility model to merge related memories.
|
||||
const compactionPrompt = `You are a memory compaction assistant. Given a set of user memories (key-value facts), merge related or redundant entries into fewer, more precise facts.
|
||||
|
||||
Rules:
|
||||
- Combine facts about the same topic into a single entry
|
||||
- Remove duplicates, keeping the most specific/recent version
|
||||
- Preserve all unique information — do not drop facts
|
||||
- Keep keys short (2-5 words, lowercase)
|
||||
- Keep values concise (1-2 sentences max)
|
||||
- Set confidence based on how well-supported the merged fact is
|
||||
|
||||
Respond ONLY with a JSON array of merged facts:
|
||||
[{"key": "...", "value": "...", "confidence": 0.0-1.0}]
|
||||
|
||||
If no merging is possible, return the original facts unchanged.`
|
||||
|
||||
// Compactor merges related memories to reduce count and improve quality.
|
||||
type Compactor struct {
|
||||
stores store.Stores
|
||||
roleResolver *roles.Resolver
|
||||
}
|
||||
|
||||
// NewCompactor creates a memory compaction service.
|
||||
func NewCompactor(stores store.Stores, rr *roles.Resolver) *Compactor {
|
||||
return &Compactor{stores: stores, roleResolver: rr}
|
||||
}
|
||||
|
||||
// CompactResult holds the outcome of a compaction run.
|
||||
type CompactResult struct {
|
||||
Before int `json:"before"`
|
||||
After int `json:"after"`
|
||||
Merged int `json:"merged"`
|
||||
Archived int `json:"archived"`
|
||||
}
|
||||
|
||||
// Compact loads active memories for a user, sends them to the utility model
|
||||
// for merging, then upserts merged entries and archives originals.
|
||||
func (c *Compactor) Compact(ctx context.Context, userID string, teamID *string) (*CompactResult, error) {
|
||||
if c.stores.Memories == nil {
|
||||
return nil, fmt.Errorf("memory store not available")
|
||||
}
|
||||
|
||||
// Load all active user-scope memories
|
||||
memories, err := c.stores.Memories.List(ctx, models.MemoryFilter{
|
||||
Scope: models.MemoryScopeUser,
|
||||
OwnerID: userID,
|
||||
Status: models.MemoryStatusActive,
|
||||
Limit: 500,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("list memories: %w", err)
|
||||
}
|
||||
|
||||
if len(memories) < compactionThreshold {
|
||||
return &CompactResult{Before: len(memories), After: len(memories)}, nil
|
||||
}
|
||||
|
||||
// Build memory text for the model
|
||||
var sb strings.Builder
|
||||
for i, m := range memories {
|
||||
sb.WriteString(fmt.Sprintf("%d. [%s] %s (confidence: %.2f)\n", i+1, m.Key, m.Value, m.Confidence))
|
||||
}
|
||||
|
||||
apiMessages := []providers.Message{
|
||||
{Role: "system", Content: compactionPrompt},
|
||||
{Role: "user", Content: fmt.Sprintf("Compact these %d memories:\n\n%s", len(memories), sb.String())},
|
||||
}
|
||||
|
||||
result, err := c.roleResolver.Complete(ctx, "utility", userID, teamID, apiMessages)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("compaction completion: %w", err)
|
||||
}
|
||||
|
||||
// Parse merged facts
|
||||
merged, err := parseExtractionResponse(result.Content)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("parse compaction response: %w", err)
|
||||
}
|
||||
|
||||
if len(merged) == 0 || len(merged) >= len(memories) {
|
||||
// Model returned nothing useful or didn't reduce — skip
|
||||
log.Printf("🧠 memory compaction: user %s — model returned %d facts (was %d), skipping",
|
||||
userID, len(merged), len(memories))
|
||||
return &CompactResult{Before: len(memories), After: len(memories)}, nil
|
||||
}
|
||||
|
||||
// Archive all original memories
|
||||
archived := 0
|
||||
for _, m := range memories {
|
||||
if err := c.stores.Memories.Archive(ctx, m.ID); err == nil {
|
||||
archived++
|
||||
}
|
||||
}
|
||||
|
||||
// Upsert merged memories
|
||||
for _, fact := range merged {
|
||||
mem := &models.Memory{
|
||||
Scope: models.MemoryScopeUser,
|
||||
OwnerID: userID,
|
||||
Key: fact.Key,
|
||||
Value: fact.Value,
|
||||
Confidence: fact.Confidence,
|
||||
Status: models.MemoryStatusActive,
|
||||
}
|
||||
if err := c.stores.Memories.Upsert(ctx, mem); err != nil {
|
||||
log.Printf("⚠ memory compaction upsert failed: key=%s: %v", fact.Key, err)
|
||||
}
|
||||
}
|
||||
|
||||
log.Printf("🧠 memory compaction: user %s — %d → %d memories (archived %d)",
|
||||
userID, len(memories), len(merged), archived)
|
||||
|
||||
return &CompactResult{
|
||||
Before: len(memories),
|
||||
After: len(merged),
|
||||
Merged: len(merged),
|
||||
Archived: archived,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// parseCompactionResponse parses the model's JSON array response.
|
||||
// Reuses the same extractedFact struct from extractor.go.
|
||||
func parseCompactionResponse(content string) ([]extractedFact, error) {
|
||||
content = strings.TrimSpace(content)
|
||||
// Strip markdown code fences if present
|
||||
if strings.HasPrefix(content, "```") {
|
||||
lines := strings.Split(content, "\n")
|
||||
if len(lines) > 2 {
|
||||
content = strings.Join(lines[1:len(lines)-1], "\n")
|
||||
}
|
||||
}
|
||||
|
||||
var facts []extractedFact
|
||||
if err := json.Unmarshal([]byte(content), &facts); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return facts, nil
|
||||
}
|
||||
@@ -95,45 +95,81 @@ func (s *Scanner) scan() {
|
||||
// If key is missing (err != nil or val == nil), extraction is enabled.
|
||||
}
|
||||
|
||||
// ── Extraction ──
|
||||
candidates, err := s.findCandidates(ctx)
|
||||
if err != nil {
|
||||
log.Printf("⚠ memory scanner: find candidates: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
if len(candidates) == 0 {
|
||||
if len(candidates) > 0 {
|
||||
sem := make(chan struct{}, s.cfg.Concurrency)
|
||||
var wg sync.WaitGroup
|
||||
|
||||
for _, c := range candidates {
|
||||
key := c.ChannelID + ":" + c.UserID
|
||||
if _, loaded := s.inFlight.LoadOrStore(key, true); loaded {
|
||||
continue
|
||||
}
|
||||
|
||||
sem <- struct{}{}
|
||||
wg.Add(1)
|
||||
|
||||
go func(cand candidate) {
|
||||
defer wg.Done()
|
||||
defer func() { <-sem }()
|
||||
defer s.inFlight.Delete(cand.ChannelID + ":" + cand.UserID)
|
||||
|
||||
extractCtx, cancel := context.WithTimeout(ctx, 2*time.Minute)
|
||||
defer cancel()
|
||||
|
||||
if err := s.extractor.Extract(extractCtx, cand.ChannelID, cand.UserID, cand.TeamID, cand.PersonaID); err != nil {
|
||||
log.Printf("⚠ memory extraction failed: channel=%s user=%s: %v",
|
||||
cand.ChannelID, cand.UserID, err)
|
||||
}
|
||||
}(c)
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
// ── v0.30.1: Compaction (decay + prune) ──
|
||||
// Runs after extraction on every tick. Gated by memory_compaction_enabled.
|
||||
s.runCompaction(ctx)
|
||||
}
|
||||
|
||||
// runCompaction applies confidence decay and pruning to stale memories.
|
||||
func (s *Scanner) runCompaction(ctx context.Context) {
|
||||
if s.stores.Memories == nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Semaphore for concurrency control
|
||||
sem := make(chan struct{}, s.cfg.Concurrency)
|
||||
var wg sync.WaitGroup
|
||||
|
||||
for _, c := range candidates {
|
||||
key := c.ChannelID + ":" + c.UserID
|
||||
if _, loaded := s.inFlight.LoadOrStore(key, true); loaded {
|
||||
continue // already in progress
|
||||
}
|
||||
|
||||
sem <- struct{}{} // acquire
|
||||
wg.Add(1)
|
||||
|
||||
go func(cand candidate) {
|
||||
defer wg.Done()
|
||||
defer func() { <-sem }() // release
|
||||
defer s.inFlight.Delete(cand.ChannelID + ":" + cand.UserID)
|
||||
|
||||
extractCtx, cancel := context.WithTimeout(ctx, 2*time.Minute)
|
||||
defer cancel()
|
||||
|
||||
if err := s.extractor.Extract(extractCtx, cand.ChannelID, cand.UserID, cand.TeamID, cand.PersonaID); err != nil {
|
||||
log.Printf("⚠ memory extraction failed: channel=%s user=%s: %v",
|
||||
cand.ChannelID, cand.UserID, err)
|
||||
// Check compaction kill switch (default: enabled)
|
||||
if s.stores.GlobalConfig != nil {
|
||||
val, err := s.stores.GlobalConfig.Get(ctx, "memory_compaction_enabled")
|
||||
if err == nil && val != nil {
|
||||
if enabled, ok := val["value"].(bool); ok && !enabled {
|
||||
return
|
||||
}
|
||||
}(c)
|
||||
}
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
// Decay: reduce confidence for memories not updated in 7 days
|
||||
olderThan := time.Now().Add(-7 * 24 * time.Hour).UTC().Format(time.RFC3339)
|
||||
decayed, err := s.stores.Memories.DecayConfidence(ctx, olderThan, 0.9)
|
||||
if err != nil {
|
||||
log.Printf("⚠ memory compaction: decay failed: %v", err)
|
||||
} else if decayed > 0 {
|
||||
log.Printf("🧠 memory compaction: decayed %d memories", decayed)
|
||||
}
|
||||
|
||||
// Prune: archive memories with confidence below 0.15
|
||||
pruned, err := s.stores.Memories.Prune(ctx, 0.15)
|
||||
if err != nil {
|
||||
log.Printf("⚠ memory compaction: prune failed: %v", err)
|
||||
} else if pruned > 0 {
|
||||
log.Printf("🧠 memory compaction: pruned %d memories", pruned)
|
||||
}
|
||||
}
|
||||
|
||||
// findCandidates queries for channels with enough new activity since last extraction.
|
||||
|
||||
Reference in New Issue
Block a user