heretek-memory-system

name	heretek-memory-system
description	Memory system development for Heretek Swarm's Cognee-backed memory. Use when implementing memory operations, working with knowledge graphs, or debugging memory issues. Covers Cognee API, access patterns, and intelligent prefetching.

Heretek Swarm Memory System

Architecture Overview

The memory system is built on Cognee (knowledge graph + vector memory engine) with access-pattern optimization and intelligent pre-fetching.

Core Components

backend/heretek_swarm/memory/
├── __init__.py           # Package exports (23 symbols)
├── access_patterns.py    # Tier classification, pattern analysis
├── cognee_reader.py      # Read-only Cognee client
├── cognee_writer.py      # Write-path Cognee client
├── eliza_memory.py       # Importance-decay memory manager
└── prefetcher.py         # LRU/LFU caches, pre-fetch scheduling

No Legacy Code

The following have been deleted and must not be reintroduced:

• base.py, persistent.py, tiering.py, versioned.py, compression.py, migration_strategies.py
• Legacy classes: DualTierMemory, PersistentMemory, TieringManager, VersionedMemory, CompressionManager

Cognee API

Reader Operations

from heretek_swarm.memory import CogneeMemoryReader

reader = CogneeMemoryReader()

# Search memories
results = await reader.search(
    query="search text",
    limit=10,
    agent="agent_name",
    min_importance=0.5
)

# Get specific memory
memory = await reader.get(memory_id="abc123")

# List memories with filters
memories = await reader.list(
    tags=["learning", "observation"],
    created_after=datetime(2024, 1, 1)
)

Writer Operations

from heretek_swarm.memory import CogneeMemoryWriter

writer = CogneeMemoryWriter()

# Add memory
memory_id = await writer.add(
    content="observation text",
    metadata={
        "agent": "explorer",
        "importance": 0.8,
        "tags": ["observation", "learning"],
        "context": {"task_id": "T001"}
    }
)

# Process into knowledge graph
await writer.cognify()

# Update memory
await writer.update(
    memory_id=memory_id,
    content="updated text",
    metadata={"importance": 0.9}
)

# Delete memory
await writer.delete(memory_id=memory_id)

Access Patterns

Tier Classification

from heretek_swarm.memory import AccessPattern, AccessTier

# Analyze access pattern
analyzer = AccessPatternAnalyzer()
pattern = analyzer.analyze(access_logs)

# Get tier recommendation
tier = pattern.recommended_tier  # HOT, WARM, COLD, ARCHIVE

# Apply tier
await writer.set_tier(memory_id, tier)

Pattern Types

• Sequential - Accesses in order
• Random - No pattern
• Temporal - Time-based access
• Frequency - Based on access count

Usage

from heretek_swarm.memory import AccessPatternAnalyzer, MemoryAccessProfile

# Create analyzer
analyzer = AccessPatternAnalyzer()

# Analyze profile
profile = MemoryAccessProfile(
    agent="explorer",
    time_window=timedelta(hours=24),
    access_count=150
)

# Get recommendations
recommendations = analyzer.recommend(profile)

Intelligent Prefetching

Cache Types

from heretek_swarm.memory import LRUCache, LFUCache

# LRU Cache (Least Recently Used)
lru = LRUCache(max_size=1000)
await lru.set("key", value)
value = await lru.get("key")

# LFU Cache (Least Frequently Used)
lfu = LFUCache(max_size=1000)
await lfu.set("key", value)
value = await lfu.get("key")

Prefetch Scheduling

from heretek_swarm.memory import (
    IntelligentPrefetcher,
    PreFetchScheduler,
    PreFetchRequest,
    PreFetchPriority
)

# Create prefetcher
prefetcher = IntelligentPrefetcher(
    reader=reader,
    cache=lru
)

# Schedule prefetch
request = PreFetchRequest(
    query="likely needed context",
    priority=PreFetchPriority.HIGH,
    agent="explorer",
    estimated_access_time=datetime.now() + timedelta(minutes=5)
)

scheduler = PreFetchScheduler(prefetcher)
await scheduler.schedule(request)

Prefetch Strategies

• Pattern-based - Based on historical access patterns
• Time-based - Scheduled prefetching
• Priority-based - High priority items first
• Predictive - ML-based prediction

Eliza Memory Manager

Importance-Decay Pattern

from heretek_swarm.memory import MemoryManager, create_memory_manager

# Create manager
manager = create_memory_manager(
    decay_rate=0.1,
    importance_threshold=0.3
)

# Add memory with importance
memory_id = await manager.remember(
    content="important observation",
    importance=0.9,
    agent="explorer"
)

# Recall memories (importance-weighted)
memories = await manager.recall(
    query="relevant context",
    limit=10,
    min_importance=0.5
)

# Forget low-importance memories
forgotten = await manager.forget(
    threshold=0.2,
    older_than=timedelta(days=30)
)

Memory Types

• Episodic - Events and experiences
• Semantic - Knowledge and facts
• Procedural - Skills and procedures
• Emotional - Emotional associations

Testing Memory Operations

Unit Tests

import pytest
from heretek_swarm.memory import CogneeMemoryReader, CogneeMemoryWriter

@pytest.mark.asyncio
async def test_memory_roundtrip():
    writer = CogneeMemoryWriter()
    reader = CogneeMemoryReader()
    
    # Write
    memory_id = await writer.add(
        content="test memory",
        metadata={"importance": 0.8}
    )
    
    # Read
    memory = await reader.get(memory_id)
    assert memory.content == "test memory"
    assert memory.metadata["importance"] == 0.8

@pytest.mark.asyncio
async def test_search():
    writer = CogneeMemoryWriter()
    reader = CogneeMemoryReader()
    
    # Add test data
    await writer.add("Python programming", tags=["coding"])
    await writer.add("JavaScript programming", tags=["coding"])
    await writer.cognify()
    
    # Search
    results = await reader.search("programming", limit=2)
    assert len(results) == 2

Integration Tests

@pytest.mark.asyncio
async def test_memory_with_agent():
    agent = ExplorerAgent(config)
    writer = CogneeMemoryWriter()
    
    # Agent observes
    observation = await agent.observe(context)
    
    # Store in memory
    await writer.add(
        content=observation,
        metadata={"agent": "explorer", "importance": 0.7}
    )
    
    # Later, agent recalls
    reader = CogneeMemoryReader()
    memories = await reader.search("observation", agent="explorer")
    assert len(memories) > 0

Debugging Memory Issues

Common Problems

1. Cognee connection failed

   • Check if Cognee service is running
   • Verify connection string
   • Check network connectivity

2. Memory not found

   • Verify memory_id exists
   • Check access permissions
   • Ensure memory wasn't deleted

3. Search returns empty

   • Run cognify() after adding memories
   • Check search query
   • Verify memory tags

4. Prefetch not working

   • Check cache size limits
   • Verify prefetch schedule
   • Monitor access patterns

Debug Commands

# Check memory count
python -c "from heretek_swarm.memory import CogneeMemoryReader; import asyncio; r = CogneeMemoryReader(); print(asyncio.run(r.count()))"

# List recent memories
python -c "from heretek_swarm.memory import CogneeMemoryReader; import asyncio; r = CogneeMemoryReader(); print(asyncio.run(r.list(limit=10)))"

# Check cache stats
python -c "from heretek_swarm.memory import LRUCache; c = LRUCache(100); print(c.stats())"

Performance Optimization

Batch Operations

# Batch add
memories = [
    {"content": f"memory {i}", "metadata": {"importance": 0.5}}
    for i in range(100)
]
await writer.add_batch(memories)

# Batch search
queries = ["query1", "query2", "query3"]
results = await reader.search_batch(queries, limit=10)

Caching Strategy

1. Hot data - Keep in LRU cache
2. Warm data - Prefetch on access
3. Cold data - Load on demand
4. Archive - Compress and store

Indexing

• Use tags for fast filtering
• Create custom indexes for common queries
• Monitor query performance

Gotchas

1. Always run cognify() after adding memories - they're not searchable until processed
2. No legacy imports - Only use Cognee-backed types
3. Async operations - All memory operations are async
4. Importance decay - Memories fade over time unless reinforced
5. Cache invalidation - Manual invalidation may be needed
6. Batch limits - Respect Cognee API rate limits

Best Practices

1. Use meaningful tags for memories
2. Set appropriate importance levels
3. Implement memory cleanup schedules
4. Monitor memory usage and performance
5. Use negative assertions to prevent legacy code creep
6. Test both read and write paths
7. Document memory schemas