Video Memory for AI Agents

The Memory Problem

Without memory, agents face limitations:

• Context limits: Can't process hour-long videos in one prompt
• Repeated work: Re-analyze video for each question
• No cross-video reasoning: Can't compare incidents across recordings
• No history: Can't remember what was discussed before

Video memory solves this by extracting and storing video content in queryable form.

Memory Architecture

┌─────────────────────────────────────────────────┐
│                  Agent Query                     │
│          "What time did the car arrive?"         │
└───────────────────────┬─────────────────────────┘
                        ↓
┌─────────────────────────────────────────────────┐
│              Memory Retrieval                    │
│    Search vector store + temporal index          │
└───────────────────────┬─────────────────────────┘
                        ↓
┌─────────────────────────────────────────────────┐
│            Relevant Context                      │
│  Frame 847 (10:32:15): "Blue sedan enters lot"   │
│  Frame 923 (10:33:42): "Sedan parks in spot 12"  │
└───────────────────────┬─────────────────────────┘
                        ↓
┌─────────────────────────────────────────────────┐
│              Agent Response                      │
│     "The car arrived at 10:32:15 AM"            │
└─────────────────────────────────────────────────┘

Building Video Memory

Step 1: Extract Content

from typing import List, Dict

def extract_video_content(video_url: str) -> List[Dict]:
    """Extract searchable content from video."""
    # Enhance first for better extraction
    enhanced = bettervideo.enhance(video_url)

    # Extract frames at key intervals
    frames = extract_keyframes(enhanced)

    content = []
    for frame in frames:
        # Describe frame with vision model
        description = vision_model.describe(frame.image)

        # Extract structured data
        objects = detector.detect(frame.image)
        text = ocr.extract(frame.image)
        faces = face_detector.detect(frame.image)

        content.append({
            "timestamp": frame.timestamp,
            "description": description,
            "objects": objects,
            "text": text,
            "faces": len(faces),
            "embedding": embed_model.embed(description)
        })

    return content

Step 2: Store in Vector Database

import chromadb

def store_video_memory(video_id: str, content: List[Dict]):
    """Store extracted content in vector database."""
    client = chromadb.Client()
    collection = client.get_or_create_collection("video_memory")

    for item in content:
        collection.add(
            ids=[f"{video_id}_{item['timestamp']}"],
            embeddings=[item['embedding']],
            metadatas=[{
                "video_id": video_id,
                "timestamp": item['timestamp'],
                "objects": str(item['objects']),
                "text": item['text'],
                "faces": item['faces']
            }],
            documents=[item['description']]
        )

Step 3: Query Memory

def query_video_memory(question: str, video_id: str = None) -> List[Dict]:
    """Retrieve relevant video content for a question."""
    collection = client.get_collection("video_memory")

    # Embed the question
    query_embedding = embed_model.embed(question)

    # Search
    where_filter = {"video_id": video_id} if video_id else None
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=10,
        where=where_filter
    )

    return results

Temporal Indexing

Video is temporal — time matters. Add temporal indexing to your memory:

# Store with temporal metadata
{
    "timestamp": "10:32:15",
    "frame_number": 847,
    "relative_time": 632.5,  # Seconds from start
    "events_before": ["Person A entered"],
    "events_after": ["Vehicle stopped"]
}

# Query with temporal constraints
def query_timerange(start_time: float, end_time: float):
    return collection.query(
        where={
            "$and": [
                {"relative_time": {"$gte": start_time}},
                {"relative_time": {"$lte": end_time}}
            ]
        }
    )

Memory-Augmented Agent

class VideoMemoryAgent:
    def __init__(self, video_ids: List[str]):
        self.video_ids = video_ids
        self.memory = VideoMemory()

    def answer(self, question: str) -> str:
        # 1. Retrieve relevant memory
        context = self.memory.query(question, self.video_ids)

        # 2. Build prompt with retrieved context
        prompt = f"""
You are analyzing video recordings. Use the following retrieved content to answer.

Retrieved Video Content:
{self.format_context(context)}

Question: {question}

Answer based only on the retrieved content. If the answer isn't in the content, say so.
"""

        # 3. Generate answer
        return llm.complete(prompt)

    def format_context(self, context):
        return "\n".join([
            f"[{c['timestamp']}] {c['description']}"
            for c in context
        ])