Build Agentic Video RAG with Strands Agents and Amazon Aurora PostgreSQL – Local Infrastructure

from strands import tool

@tool
def video_embedding_local(
    video_path: str,
    user_id: str = "default_user",
    action: str = "process",
    query: Optional[str] = None,
    similarity_threshold: float = 0.8,
    frames_per_second: int = 1,
    region: str = None,
) -> Dict[str, Any]:
    """
    Simple video embedding processor following notebook 05 pattern exactly.

    Args:
        video_path: Path to video file (local)
        user_id: User identifier for data isolation
        action: Action to perform ('process', 'search', 'list')
        query: Search query for retrieval (when action is 'search')
        similarity_threshold: Threshold for frame similarity comparison (0.0-1.0)
        frames_per_second: Frames to extract per second
        region: AWS region

    Returns:
        Dictionary with processing results
    """
    try:
        # Complex video processing logic here...
        cluster_arn = os.getenv('AURORA_CLUSTER_ARN')
        secret_arn = os.getenv('AURORA_SECRET_ARN')
        database_name = os.getenv('AURORA_DATABASE_NAME', 'kbdata')

        aurora = AuroraPostgres(cluster_arn, database_name, secret_arn, region)

        if action == "search":
            return _search_videos(query, aurora, region)
        elif action == "list":
            return _list_videos(user_id, aurora)
        else:
            return _process_video(video_path, user_id, similarity_threshold, 
                                frames_per_second, s3_bucket, aurora, region)

    except Exception as e:
        return {
            "status": "error",
            "message": f"Error: {str(e)}",
            "error_type": type(e).__name__
        }

@tool
def s3_vector_memory(
    action: str,
    content: str = None,
    query: str = None,
    user_id: str = None,
    vector_bucket_name: str = None,
    index_name: str = None,
    top_k: int = 20,
    min_score: float = 0.1
) -> Dict:
    """
    AWS-native memory management using Amazon S3 Vectors.

    Actions:
    - store: Store new memory content
    - retrieve: Search and retrieve relevant memories
    - list: List all user memories
    """
    if not user_id:
        return {"status": "error", "message": "user_id is required for memory isolation"}

    try:
        config = {
            "bucket_name": vector_bucket_name or os.environ.get('VECTOR_BUCKET_NAME'),
            "index_name": index_name or os.environ.get('VECTOR_INDEX_NAME'),
            "region": region_name or os.environ.get('AWS_REGION', 'us-east-1'),
            "model_id": embedding_model or os.environ.get('EMBEDDING_MODEL')
        }

        bedrock = boto3.client("bedrock-runtime", region_name=config["region"])
        s3vectors = boto3.client("s3vectors", region_name=config["region"])

        if action == "store":
            return _store_memory(s3vectors, bedrock, config, content, user_id)
        elif action == "retrieve":
            return _retrieve_memories(s3vectors, bedrock, config, query, user_id, top_k, min_score)
        elif action == "list":
            return _list_memories(s3vectors, bedrock, config, user_id, top_k)

    except Exception as e:
        return {"status": "error", "message": str(e)}

@tool
def display_video_images(
    search_results: List[Dict[str, Any]],
    region: str = None,
    base_path: str = "images/"
) -> Dict[str, Any]:
    """
    Display images from video search results.

    Args:
        search_results: List of search results from video_embeddings_aws
        region: AWS region for S3 client
        base_path: Local path to save downloaded images
    """
    try:
        os.makedirs(base_path, exist_ok=True)
        s3_client = boto3.client('s3', region_name=region)

        displayed_count = 0
        text_count = 0

        for result in search_results:
            metadata = result.get('metadata', {})
            content_type = metadata.get('content_type', 'unknown')

            if content_type == "text":
                text_count += 1
                print(f"📝 Text Result: {result.get('content_preview', '')}")
            elif content_type == "image":
                displayed_count += 1
                # Download and display logic...

        return {
            "status": "success",
            "images_displayed": displayed_count,
            "text_results": text_count,
            "total_processed": len(search_results)
        }

    except Exception as e:
        return {"status": "error", "message": f"Failed to display images: {str(e)}"}

from strands import Agent
from strands.models import BedrockModel

model = BedrockModel(model_id="us.anthropic.claude-3-5-sonnet-20241022-v2:0")

VIDEO_SYSTEM_PROMPT = """You are a video processing AI assistant.

Available actions:
- process: Upload and process videos 
- search: Search video content using semantic similarity
- list: List all processed videos

Use video_embeddings_aws for all cloud video operations.
Use display_video_images to show search results.
"""

from strands import Agent
agent = Agent()  # Uses Claude 4 Sonnet by default

agent = Agent(model="anthropic.claude-sonnet-4-20250514-v1:0")

from strands.models import BedrockModel

model = BedrockModel(
    model_id="anthropic.claude-sonnet-4-20250514-v1:0",
    temperature=0.3,
    top_p=0.8
)
agent = Agent(model=model)

from strands.models.anthropic import AnthropicModel

model = AnthropicModel(
    model_id="claude-sonnet-4-20250514",
    max_tokens=1028,
    params={"temperature": 0.7}
)

video_agent = Agent(
    model=model, 
    tools=[video_embedding_local, display_video_images],
    system_prompt=VIDEO_SYSTEM_PROMPT
)

memory_agent = Agent(
    model=model, 
    tools=[video_embedding_local, s3_vector_memory, display_video_images],
    system_prompt=VIDEO_SYSTEM_PROMPT
)

response = video_agent(f"What is the video about in {VIDEO_PATH}?")

response = memory_agent(f"""I'm interested in learning about AI and database technologies. 
Store this preference for user {USER_ID}, then search the video in {VIDEO_PATH} for technical 
discussions about vector databases and embeddings.""")

response = memory_agent(f"What did user {USER_ID} ask before?")