Implementing Conversational AI Part 3

import boto3
from botocore.exceptions import ClientError
from typing import List, Dict

# AWS clients
bedrock = boto3.client("bedrock-runtime", region_name="us-east-1")
dynamodb = boto3.resource("dynamodb", region_name="us-east-1")

# DynamoDB table that stores conversation history per session
table = dynamodb.Table("ChatSessions")

MAX_MESSAGES = 10  # Keep only the most recent messages for context control

def load_messages(session_id: str) -> List[Dict]:
    """Load conversation history for a session from DynamoDB."""
    try:
        response = table.get_item(Key={"sessionId": session_id})
        item = response.get("Item")
        if item and "messages" in item:
            return item["messages"]
        return []
    except ClientError as e:
        print(f"Error loading session {session_id}: {e}")
        return []

def save_messages(session_id: str, messages: List[Dict]) -> None:
    """Save conversation history for a session to DynamoDB (overwrites existing history)."""
    try:
        table.put_item(
            Item={
                "sessionId": session_id,
                "messages": messages
            }
        )
    except ClientError as e:
        print(f"Error saving session {session_id}: {e}")
        raise

def trim_messages(messages: List[Dict], max_messages: int = MAX_MESSAGES) -> List[Dict]:
    """Keep only the most recent messages to avoid context window bloat."""
    return messages[-max_messages:]

# Example usage
if __name__ == "__main__":
    # Example flow: load, append a new user message, trim, save
    session_id = "user123"
    history = load_messages(session_id)
    history.append({"role": "user", "content": "Hello"})
    history = trim_messages(history)
    save_messages(session_id, history)

    # A different session uses a different session_id
    other_session = "user456"
    other_history = [{"role": "user", "content": "Hi there"}]
    save_messages(other_session, other_history)