Building Vision-Enabled AI Agents

Architecture Decision

Choose your architecture based on use case:

Pattern	Best For	Complexity
Preprocessing + VLM	Simple analysis, Q&A	Low
Agent with tools	Complex reasoning, dynamic	Medium
Multi-agent	High volume, parallel	High
Memory-augmented	Long videos, repeated queries	Medium

For your first agent, start with Preprocessing + VLM. It's simple and effective.

Step 1: Set Up Enhancement

# Install BetterVideo client
pip install bettervideo

# Initialize
import bettervideo
client = bettervideo.Client(api_key="your_key")

# Enhancement function
def enhance_video(video_url: str) -> str:
    job = client.submit(
        video_url=video_url,
        resolution="1080p"
    )
    result = job.wait()
    return result.download_url

Step 2: Frame Extraction

import cv2
import base64
from typing import List

def extract_frames(video_url: str, interval: float = 2.0) -> List[str]:
    """Extract frames at regular intervals, return as base64."""
    cap = cv2.VideoCapture(video_url)
    fps = cap.get(cv2.CAP_PROP_FPS)
    frame_interval = int(fps * interval)

    frames = []
    frame_count = 0

    while True:
        ret, frame = cap.read()
        if not ret:
            break

        if frame_count % frame_interval == 0:
            # Convert to base64 for API
            _, buffer = cv2.imencode('.jpg', frame)
            b64 = base64.b64encode(buffer).decode('utf-8')
            frames.append(b64)

        frame_count += 1

    cap.release()
    return frames

Step 3: Vision Model Integration

from openai import OpenAI

client = OpenAI()

def analyze_frame(frame_b64: str, question: str) -> str:
    """Analyze a single frame with GPT-4V."""
    response = client.chat.completions.create(
        model="gpt-4-vision-preview",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": question},
                {"type": "image_url", "image_url": {
                    "url": f"data:image/jpeg;base64,{frame_b64}"
                }}
            ]
        }],
        max_tokens=500
    )
    return response.choices[0].message.content

def analyze_video(frames: List[str], question: str) -> str:
    """Analyze multiple frames and synthesize answer."""
    # Analyze each frame
    frame_analyses = [
        f"Frame {i}: {analyze_frame(f, 'Describe what you see.')}"
        for i, f in enumerate(frames)
    ]

    # Synthesize with LLM
    synthesis = client.chat.completions.create(
        model="gpt-4",
        messages=[{
            "role": "system",
            "content": "You are analyzing video frames to answer a question."
        }, {
            "role": "user",
            "content": f"""
Frame analyses:
{chr(10).join(frame_analyses)}

Question: {question}

Answer based on the frame analyses.
"""
        }]
    )
    return synthesis.choices[0].message.content

Step 4: Complete Agent

class VideoAgent:
    def __init__(self):
        self.bv_client = bettervideo.Client(api_key="...")
        self.openai = OpenAI()

    def process(self, video_url: str, question: str) -> str:
        # 1. Enhance video
        print("Enhancing video...")
        enhanced_url = enhance_video(video_url)

        # 2. Extract frames
        print("Extracting frames...")
        frames = extract_frames(enhanced_url, interval=2.0)
        print(f"Extracted {len(frames)} frames")

        # 3. Analyze
        print("Analyzing...")
        answer = analyze_video(frames, question)

        return answer

# Usage
agent = VideoAgent()
answer = agent.process(
    video_url="https://example.com/video.mp4",
    question="What happened in this video?"
)
print(answer)

Step 5: Add Error Handling

class VideoAgent:
    def process(self, video_url: str, question: str) -> str:
        try:
            # Enhancement with retry
            for attempt in range(3):
                try:
                    enhanced_url = enhance_video(video_url)
                    break
                except bettervideo.RateLimitError:
                    time.sleep(2 ** attempt)
            else:
                raise Exception("Enhancement failed after retries")

            # Frame extraction with validation
            frames = extract_frames(enhanced_url)
            if not frames:
                raise ValueError("No frames extracted")

            # Limit frames to control cost
            if len(frames) > 30:
                frames = frames[::len(frames)//30][:30]

            # Analyze with timeout
            answer = analyze_video(frames, question)

            return answer

        except Exception as e:
            logger.error(f"Agent error: {e}")
            return f"Unable to analyze video: {str(e)}"

Deployment Considerations

• API keys: Store securely (environment variables, secrets manager)
• Async processing: Use queues for long-running jobs
• Monitoring: Track latency, errors, and costs
• Scaling: Run multiple workers for parallel processing
• Caching: Cache enhanced videos to avoid re-processing

# Example with Celery for async processing
from celery import Celery

app = Celery('video_agent')

@app.task
def process_video_task(video_url: str, question: str) -> str:
    agent = VideoAgent()
    return agent.process(video_url, question)

# Submit async
result = process_video_task.delay(video_url, question)
answer = result.get(timeout=300)