Multi-Agent Video Systems

Why Multi-Agent?

Single agents hit limits:

• Context windows: Can't process hours of video in one call
• Specialization: No single model is best at everything
• Parallelism: Sequential processing is slow
• Reliability: One failure shouldn't crash everything

Multi-agent systems address these by distributing work across specialized agents.

Coordination Patterns

Pattern 1: Pipeline

Agents process sequentially, each adding to the result:

Video → [Enhancement] → [Detection] → [Tracking] → [Analysis] → Output

Good for: Dependent tasks, clear processing stages.

Pattern 2: Fan-Out / Fan-In

Multiple agents process the same video in parallel:

         ┌→ [Face Agent] ────┐
Video ──┼→ [OCR Agent] ─────┼→ [Aggregator] → Result
         └→ [Scene Agent] ───┘

Good for: Independent extraction tasks, speed.

Pattern 3: Supervisor

One agent coordinates others:

[Supervisor]
    ├── Assigns tasks to worker agents
    ├── Monitors progress
    ├── Handles failures
    └── Aggregates results

Good for: Dynamic task assignment, complex workflows.

Pattern 4: Debate/Consensus

Multiple agents analyze, then reach consensus:

[Agent A] → Opinion A ─┐
[Agent B] → Opinion B ─┼→ [Consensus] → Final Answer
[Agent C] → Opinion C ─┘

Good for: High-stakes decisions, reducing bias.

Implementation Example

import asyncio
from typing import Dict, List

class MultiAgentVideoProcessor:
    def __init__(self):
        self.agents = {
            "face": FaceAgent(),
            "ocr": OCRAgent(),
            "scene": SceneAgent(),
            "objects": ObjectAgent()
        }

    async def process_parallel(self, video_url: str) -> Dict:
        """Run all agents in parallel."""
        # Enhance once, use for all
        enhanced = await bettervideo.enhance_async(video_url)

        # Run agents in parallel
        tasks = {
            name: agent.process(enhanced)
            for name, agent in self.agents.items()
        }

        results = await asyncio.gather(*tasks.values())
        return dict(zip(tasks.keys(), results))

    async def process_pipeline(self, video_url: str) -> Dict:
        """Run agents in sequence, each building on previous."""
        enhanced = await bettervideo.enhance_async(video_url)

        # Stage 1: Detection
        objects = await self.agents["objects"].process(enhanced)

        # Stage 2: Track detected objects
        tracks = await self.agents["tracking"].track(objects)

        # Stage 3: Analyze with context
        analysis = await self.agents["analysis"].analyze(
            video=enhanced,
            objects=objects,
            tracks=tracks
        )

        return analysis

Agent Communication

Agents need to share information. Options:

Shared State

# All agents read/write to shared state
state = SharedState()
state.set("faces", face_results)
scene_results = scene_agent.process(state.get("faces"))

Message Passing

# Agents communicate via messages
await face_agent.send("scene_agent", {"faces": results})
message = await scene_agent.receive()

Blackboard

# Central blackboard that all agents can read/write
blackboard = Blackboard()
face_agent.post(blackboard, "faces", results)
scene_agent.read(blackboard, "faces")

Error Handling

• Retry: Individual agent failures retry independently
• Fallback: Alternative agent if primary fails
• Partial results: Return what succeeded, flag what failed
• Circuit breaker: Stop sending to failing agents

async def resilient_process(video: str) -> Dict:
    results = {}

    for name, agent in agents.items():
        try:
            results[name] = await agent.process(video)
        except AgentError as e:
            # Log failure, continue with others
            results[name] = {"error": str(e), "status": "failed"}

            # Try fallback if available
            if fallback := get_fallback(name):
                results[name] = await fallback.process(video)

    return results