Short-term Memory

• 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);
  

• 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
  );
  

• 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
  );
  

Long-term Memory

• 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))
  

• 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
  

• 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)))
  

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

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

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)

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)