1.1.3.1. Short-term Memory

1. Context Windows
   

   CREATE TABLE context_windows (
       id BIGSERIAL PRIMARY KEY,
       created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP,
       expires_at TIMESTAMPTZ,
       context_data JSONB,
       embedding vector(1536),
       active BOOLEAN DEFAULT true
   );
   
   CREATE INDEX idx_context_active 
       ON context_windows(active);
   CREATE INDEX idx_context_embedding 
       ON context_windows USING ivfflat (embedding vector_cosine_ops);
   

2. Active Task State
   

   CREATE TABLE active_tasks (
       id BIGSERIAL PRIMARY KEY,
       context_window_id BIGINT REFERENCES context_windows(id),
       task_data JSONB,
       state task_state,
       progress FLOAT,
       metadata JSONB
   );
   

3. Recent Interactions
   

   CREATE TABLE interactions (
       id BIGSERIAL PRIMARY KEY,
       task_id BIGINT REFERENCES active_tasks(id),
       interaction_type TEXT,
       content TEXT,
       embedding vector(1536),
       created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
   );
   

1.1.3.2. Long-term Memory

1. Vector-based Pattern Storage
   

   class PatternStorage:
       def store_pattern(self, pattern: Dict, embedding: List[float]):
           """Store a pattern with its vector embedding"""
           return self.db.execute("""
               INSERT INTO task_patterns 
                   (pattern_data, pattern_embedding, metadata)
               VALUES (%s, %s, %s)
               RETURNING id
           """, (json.dumps(pattern), embedding, self.get_metadata()))
   
       def find_similar(self, embedding: List[float], limit: int = 5):
           """Find similar patterns using vector similarity"""
           return self.db.execute("""
               SELECT id, pattern_data, metadata,
                      1 - (pattern_embedding <=> %s) as similarity
               FROM task_patterns
               ORDER BY pattern_embedding <=> %s
               LIMIT %s
           """, (embedding, embedding, limit))
   

2. Success/Failure Tracking
   

   class OutcomeTracker:
       def record_outcome(self, pattern_id: int, 
                         outcome: OutcomeType, 
                         context: Dict):
           """Record task outcome and update pattern statistics"""
           with self.db.transaction():
               # Record specific outcome
               outcome_id = self.store_outcome(pattern_id, outcome, context)
               
               # Update pattern statistics
               self.update_pattern_stats(pattern_id, outcome)
               
               # Store learned improvements
               if outcome == OutcomeType.FAILURE:
                   self.store_error_pattern(pattern_id, context)
               
               return outcome_id
   

3. Solution Paths
   

   class SolutionPathManager:
       def record_path(self, task_id: int, steps: List[Dict]):
           """Record successful solution path"""
           path_data = {
               'task_id': task_id,
               'steps': steps,
               'metadata': self.extract_metadata(steps)
           }
           return self.db.execute("""
               INSERT INTO solution_paths 
                   (task_id, path_data, path_embedding)
               VALUES (%s, %s, %s)
               RETURNING id
           """, (task_id, json.dumps(path_data), 
                 self.embed_path(steps)))
   

1.1.4.1. Memory Integration

class MemoryIntegration:
    def __init__(self):
        self.stm = ShortTermMemory()
        self.ltm = LongTermMemory()
        self.vector_db = VectorStore()
        
    async def process_task(self, task: Dict):
        # Check short-term memory for recent context
        context = await self.stm.get_recent_context(task)
        
        # Find similar patterns in long-term memory
        patterns = await self.ltm.find_similar_patterns(task)
        
        # Get relevant vector embeddings
        embeddings = await self.vector_db.get_relevant(task)
        
        # Combine and process
        result = await self.process_with_memory(
            task, context, patterns, embeddings)
        
        # Update memories
        await self.update_memories(task, result)
        
        return result

1.1.4.2. Pattern Recognition

class PatternRecognition:
    def identify_patterns(self, task_history: List[Dict]):
        # Extract pattern features
        features = self.extract_features(task_history)
        
        # Generate embeddings
        embeddings = self.generate_embeddings(features)
        
        # Find clusters
        clusters = self.cluster_patterns(embeddings)
        
        # Extract pattern templates
        templates = self.extract_templates(clusters)
        
        return templates

1.1.5.1. Basic Memory Operations

from memory_system import AgentMemorySystem, MemoryConfig

config = MemoryConfig(
    dsn="postgresql://user:pass@localhost:5432/agent_memory",
    context_window_expiry=3600,
    embedding_dimension=1536
)

memory = AgentMemorySystem(config)

# Store context
context_id = await memory.create_context_window(
    {"user": "user123", "session": "abc123"},
    embedding)

# Start task
task_id = await memory.start_task(
    context_id,
    {"type": "file_operation", "action": "read"})

# Record interaction
await memory.record_interaction(
    task_id,
    "command",
    "read file.txt",
    embedding)

1.1.5.2. Advanced Pattern Usage

# Find similar patterns
patterns = await memory.find_similar_patterns(embedding)

# Record outcome
if patterns:
    await memory.record_outcome(
        patterns[0]['id'],
        OutcomeType.SUCCESS,
        {"steps": ["open", "read", "close"]},
        embedding)

# Update pattern statistics
await memory.update_pattern_stats(
    pattern_id,
    OutcomeType.SUCCESS)