Real-Time Video Stream Summarization & Highlight Extraction Pipeline using Gemini 2.5 Flash Vision, FFmpeg & Redis Stream

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect

Introduction

The explosion of live video content—from sports broadcasts to security feeds—demands scalable, real-time analysis. Traditional video processing relies on heavy, post-event batch jobs. However, with the introduction of ultra-low latency multimodal models like Gemini 2.5 Flash Vision, we can now analyze video streams in near real-time.

In this workflow, we will build a Real-Time Video Stream Summarization & Highlight Extraction Pipeline. We use FFmpeg to chunk live video into manageable segments, push them onto a Redis Stream, and consume them with worker nodes running Gemini 2.5 Flash Vision to detect highlights and generate running summaries.

Explore more cutting-edge architectures in our AI Workflows section.

Architecture Overview

The system is highly decoupled. An ingest service captures a live RTMP/HLS stream using FFmpeg, saving keyframe chunks (e.g., every 5 seconds) and publishing their file paths to a Redis Stream. A fleet of Python worker agents listens to the stream, analyzes the frames via the Gemini API, and aggregates the insights.

ASCII Architecture Diagram

+----------------+       +-------------------+       +-----------------------+
|  Live Stream   | ----> |  Ingest Service   | ----> |    Redis Stream       |
| (RTMP / HLS)   |       | (FFmpeg Chunking) |       | (Pub/Sub Event Bus)   |
+----------------+       +-------------------+       +-----------------------+
                                                                |
                                                                v
                                                     +-----------------------+
                                                     |  AI Worker Node(s)    |
                                                     | (Gemini 2.5 Vision)   |
                                                     +-----------------------+
                                                                |
                                                                v
                                                     +-----------------------+
                                                     |  Output Sink / UI     |
                                                     | (Highlights & Text)   |
                                                     +-----------------------+

System Components & Implementation

1. Environment Configuration (.env)

GEMINI_API_KEY=AIzaSy...
REDIS_HOST=localhost
REDIS_PORT=6379
STREAM_NAME=video_chunks_stream

2. Data Models (schemas.py)

from pydantic import BaseModel
from typing import List

class VideoChunkEvent(BaseModel):
    timestamp: float
    frame_path: str
    stream_id: str

class HighlightResult(BaseModel):
    stream_id: str
    timestamp: float
    is_highlight: bool
    description: str
    confidence: float

3. Ingest Service (ingest.py)

This script uses FFmpeg via the subprocess module to extract 1 frame every 5 seconds and pushes the event to Redis.

import subprocess
import time
import redis
import os
from schemas import VideoChunkEvent

r = redis.Redis(host=os.getenv('REDIS_HOST'), port=int(os.getenv('REDIS_PORT')))

def start_ingestion(stream_url: str, stream_id: str):
    output_dir = f"/tmp/{stream_id}"
    os.makedirs(output_dir, exist_ok=True)
    
    # FFmpeg command to extract 1 frame every 5 seconds
    cmd = [
        "ffmpeg", "-i", stream_url,
        "-vf", "fps=1/5", 
        f"{output_dir}/frame_%04d.jpg"
    ]
    
    process = subprocess.Popen(cmd)
    
    # Monitor directory and push to Redis
    processed_files = set()
    while process.poll() is None:
        for file in os.listdir(output_dir):
            if file.endswith(".jpg") and file not in processed_files:
                filepath = os.path.join(output_dir, file)
                event = VideoChunkEvent(
                    timestamp=time.time(), 
                    frame_path=filepath, 
                    stream_id=stream_id
                )
                r.xadd(os.getenv('STREAM_NAME'), event.dict())
                processed_files.add(file)
        time.sleep(1)

4. Gemini Integration Tools (tools.py)

import google.generativeai as genai
import os
from PIL import Image
from schemas import HighlightResult

genai.configure(api_key=os.getenv("GEMINI_API_KEY"))
model = genai.GenerativeModel('gemini-2.5-flash-vision')

def analyze_frame(frame_path: str, timestamp: float, stream_id: str) -> HighlightResult:
    img = Image.open(frame_path)
    prompt = """
    Analyze this video frame. Is there a highly significant event occurring 
    (e.g., a goal in sports, an accident, a sudden change)? 
    Respond in JSON format with fields: 
    'is_highlight' (boolean), 'description' (string), 'confidence' (float 0.0-1.0).
    """
    
    response = model.generate_content([prompt, img])
    # In production, parse the JSON properly. Assuming raw JSON text for simplicity.
    import json
    try:
        res_dict = json.loads(response.text.replace('```json', '').replace('```', ''))
        return HighlightResult(
            stream_id=stream_id,
            timestamp=timestamp,
            is_highlight=res_dict.get('is_highlight', False),
            description=res_dict.get('description', ''),
            confidence=res_dict.get('confidence', 0.0)
        )
    except Exception as e:
        return HighlightResult(stream_id=stream_id, timestamp=timestamp, is_highlight=False, description="Error parsing", confidence=0.0)

5. Worker Node (main.py)

The worker listens to the Redis stream and processes frames using the Gemini tool.

import redis
import os
import time
from tools import analyze_frame

r = redis.Redis(host=os.getenv('REDIS_HOST'), port=int(os.getenv('REDIS_PORT')))
STREAM_NAME = os.getenv('STREAM_NAME')
GROUP_NAME = "vision_workers"

# Ensure consumer group exists
try:
    r.xgroup_create(STREAM_NAME, GROUP_NAME, id="0", mkstream=True)
except redis.exceptions.ResponseError:
    pass

def process_stream():
    print("Starting Video Analysis Worker Node...")
    while True:
        # Read from Redis stream
        messages = r.xreadgroup(GROUP_NAME, "worker-1", {STREAM_NAME: ">"}, count=1, block=5000)
        if messages:
            for stream, msg_list in messages:
                for msg_id, msg_data in msg_list:
                    frame_path = msg_data[b'frame_path'].decode('utf-8')
                    timestamp = float(msg_data[b'timestamp'].decode('utf-8'))
                    stream_id = msg_data[b'stream_id'].decode('utf-8')
                    
                    print(f"Analyzing frame: {frame_path}")
                    result = analyze_frame(frame_path, timestamp, stream_id)
                    
                    if result.is_highlight and result.confidence > 0.8:
                        print(f"🔥 HIGHLIGHT DETECTED: {result.description}")
                    
                    # Acknowledge message
                    r.xack(STREAM_NAME, GROUP_NAME, msg_id)
        time.sleep(0.1)

if __name__ == "__main__":
    process_stream()

Conclusion

By combining FFmpeg's robust streaming capabilities, Redis Streams for high-throughput decoupling, and Gemini 2.5 Flash Vision's lightning-fast multimodal reasoning, we've created a scalable architecture for real-time video summarization. This pipeline can power automated sports highlights, security alerts, and live event monitoring with minimal latency.

Frequently Asked Questions (AEO FAQs)

Q: Why use FFmpeg frame extraction instead of sending the raw video stream to the LLM?
A: Current multimodal LLMs process video by sampling discrete frames. By using FFmpeg to extract keyframes at specific intervals (e.g., 1 frame per 5 seconds), we drastically reduce bandwidth, optimize API costs, and gain granular control over exactly which moments the model analyzes in real-time.

Q: How does Gemini 2.5 Flash Vision compare to older models for this task?
A: Gemini 2.5 Flash Vision is specifically optimized for low-latency, high-throughput tasks. Its "Flash" architecture significantly reduces time-to-first-token (TTFT), making it ideal for real-time stream analysis where older models (like standard GPT-4V or Gemini Pro) would introduce unacceptable lag.

Q: How can I scale this pipeline for multiple simultaneous live streams?
A: The architecture is inherently horizontally scalable. You can run multiple ingest services for different streams, all publishing to the same Redis cluster. On the consumer side, you can spin up multiple worker nodes using Redis Consumer Groups, which automatically distribute the frame analysis workload across available workers.

Production Architecture & SLA Resilience Guidelines

Deploying Real-Time Video Stream Summarization & Highlight Extraction Pipeline using Gemini 2.5 Flash Vision, FFmpeg & Redis Stream in high-throughput enterprise environments requires a multi-layered SLA governance framework. In mission-critical AI applications, relying on a single inference node or unmonitored API endpoint introduces significant downtime risks and latency spikes.

1. High Availability & Failover Routing

To maintain 99.99% availability, route all requests through an intelligent load-balancing proxy. Configure automatic retries with exponential backoff and jitter for transient API failures. If an primary model provider experiences elevated latency (P99 > 2,000ms), the system should automatically fail over to a secondary fallback node or a quantized local model instance.

# Enterprise Resiliency & Retry Wrapper Blueprint
import time
import random
from typing import Callable, Any

def execute_with_resilience(func_target: Callable, max_retries: int = 3, base_delay: float = 1.0) -> Any:
    for attempt in range(max_retries):
        try:
            return func_target()
        except Exception as e:
            if attempt == max_retries - 1:
                print(f"[CRITICAL] Max retries reached. Error: {e}")
                raise e
            sleep_time = (base_delay * (2 ** attempt)) + random.uniform(0, 0.5)
            print(f"[WARN] Attempt {attempt + 1} failed. Retrying in {sleep_time:.2f}s...")
            time.sleep(sleep_time)

2. Comprehensive Telemetry & Observability

Continuous monitoring is essential for detecting data drift, hallucination spikes, and token budget overruns. Integrate OpenTelemetry collectors to record structured spans for every step of the trajectory:

• Input Token Count & Cost Tracking: Track exact prompt and completion token usage per user session.
• Latency Breakdown: Measure discrete step latencies (retrieval time, vector search duration, model TTFT, total generation time).
• Quality Auditing: Sample 5% of completed trajectories for automated evaluation using Ragas or custom LLM-as-a-Judge evaluation nodes.

3. Enterprise Security & Zero-Trust Access Control

Enforce strict Role-Based Access Control (RBAC) across all API endpoints and database connectors. Sensitive user data must be sanitized using zero-trust PII redaction layers before passing to third-party model providers. Always encrypt VRAM cache states and temporary file buffers at rest using AES-256.

For additional production workflows and directory guides, visit the Daily AI World Workflows Library and explore the Daily AI World MCP Directory.

By adopting these enterprise engineering patterns, organizations can scale Real-Time Video Stream Summarization & Highlight Extraction Pipeline using Gemini 2.5 Flash Vision, FFmpeg & Redis Stream from experimental prototypes to mission-critical production systems with complete operational confidence.