Implementing Memory on AWS

123456789101112
import { DynamoDBClient } from '@aws-sdk/client-dynamodb';
import { DynamoDBDocumentClient, PutCommand, QueryCommand } from '@aws-sdk/lib-dynamodb';

interface SessionMemory {
  tenantId: string;
  sessionId: string;
  messageId: string;
  timestamp: number;
  role: 'user' | 'assistant' | 'system';
  content: string;
  metadata: Record<string, any>;
  ttl: number;

123456789101112
import boto3
from datetime import datetime, timedelta
import json

class AgentMemoryStore:
    def __init__(self, table_name: str):
        self.dynamodb = boto3.resource('dynamodb')
        self.table = self.dynamodb.Table(table_name)
    
    def save_turn(self, session_id: str, turn_number: int, 
                  user_input: str, agent_response: str, 
                  tool_calls: list = None):

123456789101112
import redis
import json
from datetime import datetime, timedelta
from typing import Optional, List, Dict
import hashlib

class AgentWorkingMemory:
    def __init__(self, redis_url: str, default_ttl: int = 1800):
        self.redis = redis.from_url(redis_url, decode_responses=True)
        self.default_ttl = default_ttl  # 30 minutes
    
    def get_or_create_context(self, session_id: str) -> dict:

123456789101112
import boto3
import json
import hashlib
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from opensearchpy import OpenSearch, RequestsHttpConnection
import redis

class AgentMemoryManager:
    def __init__(self, config: Dict):
        self.dynamodb = boto3.resource('dynamodb')
        self.s3 = boto3.client('s3')

123456789101112
import boto3
from datetime import datetime, timedelta
from typing import List, Dict
import anthropic

class MemoryConsolidator:
    def __init__(self, memory_manager, claude_client):
        self.memory = memory_manager
        self.claude = claude_client
        
    def calculate_importance_score(self, memory: Dict) -> float:
        """Calculate memory importance based on multiple factors"""