Module 7: Use Cases

4. Video: Frames + LLM

Next →

Description

Multimodal LLMs don't process video directly (yet). The standard technique is to extract representative frames from the video, send them as images to a Vision API, and combine the individual analyses into a coherent summary. In this capsule you build a complete video analysis pipeline: frame extraction with OpenCV, smart key-frame selection, analysis with the Vision API, and synthesis with an LLM.

Why it matters: Video is the modality richest in information and the hardest to process. Educational platforms analyze recorded classes, security systems review cameras, marketing teams analyze content, and e-commerce platforms process video reviews. The frames → Vision → LLM pattern is the foundation of all these cases.

Connection with the module: This pipeline is another destination of the Use Case Selector (capsule 08). When the router detects a video file, it applies the frame extraction and analysis you build here. The batch and rate limit patterns from capsule 06 are critical for processing many frames.

Visual Pipeline

┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│    Video     │────▶│   Extract    │────▶│   Select     │
│   (.mp4)     │     │   frames     │     │  key frames  │
└──────────────┘     └──────────────┘     └──────┬───────┘
                                                  │
                                                  ▼
┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│   Final      │◀────│   Synthesize │◀────│  Vision API  │
│   summary    │     │   with LLM   │     │  per frame   │
└──────────────┘     └──────────────┘     └──────────────┘

Stages:

  1. Load video — Open with OpenCV
  2. Extract frames — By fixed interval or scene-change detection
  3. Select key frames — Filter out duplicates and irrelevant frames
  4. Analyze with Vision — Send each frame to GPT-4o for a description
  5. Synthesize — LLM combines the descriptions into a temporal summary

Step 1: Video Info

Before extracting frames, get the video's metadata:

import cv2
from pathlib import Path

def get_video_info(video_path: str) -> dict:
    cap = cv2.VideoCapture(video_path)

    if not cap.isOpened():
        return {"error": f"Can't open: {video_path}"}

    fps = cap.get(cv2.CAP_PROP_FPS)
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    duration = total_frames / fps if fps > 0 else 0

    cap.release()

    return {
        "path": video_path,
        "fps": round(fps, 2),
        "total_frames": total_frames,
        "width": width,
        "height": height,
        "duration_seconds": round(duration, 2),
        "duration_formatted": format_duration(duration),
        "size_mb": round(Path(video_path).stat().st_size / (1024 * 1024), 2)
    }


def format_duration(seconds: float) -> str:
    minutes = int(seconds // 60)
    secs = int(seconds % 60)
    return f"{minutes}:{secs:02d}"

Step 2: Frame Extraction by Interval

Basic extraction

import os

def extract_frames_by_interval(
    video_path: str,
    interval_seconds: float = 5.0,
    output_dir: str = "/tmp/video_frames",
    max_frames: int = 50
) -> list[dict]:
    os.makedirs(output_dir, exist_ok=True)

    cap = cv2.VideoCapture(video_path)
    fps = cap.get(cv2.CAP_PROP_FPS)
    interval_frames = int(fps * interval_seconds)
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))

    frames = []
    frame_id = 0

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

        if frame_id % interval_frames == 0:
            timestamp = frame_id / fps
            frame_path = os.path.join(output_dir, f"frame_{frame_id:06d}.jpg")
            cv2.imwrite(frame_path, frame)

            frames.append({
                "path": frame_path,
                "frame_id": frame_id,
                "timestamp": round(timestamp, 2),
                "timestamp_formatted": format_duration(timestamp)
            })

            if len(frames) >= max_frames:
                break

        frame_id += 1

    cap.release()
    return frames

Optimal interval calculation

def calculate_optimal_interval(
    duration_seconds: float,
    target_frames: int = 20,
    min_interval: float = 2.0,
    max_interval: float = 30.0
) -> float:
    if duration_seconds <= 0 or target_frames <= 0:
        return min_interval

    interval = duration_seconds / target_frames
    return max(min_interval, min(interval, max_interval))

Step 3: Scene-Change Detection

Instead of extracting at a fixed interval (which can capture redundant frames), detect significant visual changes:

import numpy as np

def extract_frames_by_scene_change(
    video_path: str,
    threshold: float = 30.0,
    min_gap_seconds: float = 2.0,
    output_dir: str = "/tmp/video_frames",
    max_frames: int = 30
) -> list[dict]:
    os.makedirs(output_dir, exist_ok=True)

    cap = cv2.VideoCapture(video_path)
    fps = cap.get(cv2.CAP_PROP_FPS)
    min_gap_frames = int(fps * min_gap_seconds)

    frames = []
    prev_gray = None
    frame_id = 0
    last_captured = -min_gap_frames

    ret, first_frame = cap.read()
    if ret:
        frame_path = os.path.join(output_dir, f"frame_{0:06d}.jpg")
        cv2.imwrite(frame_path, first_frame)
        frames.append({
            "path": frame_path,
            "frame_id": 0,
            "timestamp": 0.0,
            "timestamp_formatted": "0:00",
            "trigger": "first_frame"
        })
        prev_gray = cv2.cvtColor(first_frame, cv2.COLOR_BGR2GRAY)
        last_captured = 0

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

        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)

        if prev_gray is not None and (frame_id - last_captured) >= min_gap_frames:
            diff = cv2.absdiff(prev_gray, gray)
            mean_diff = diff.mean()

            if mean_diff > threshold:
                timestamp = frame_id / fps
                frame_path = os.path.join(output_dir, f"frame_{frame_id:06d}.jpg")
                cv2.imwrite(frame_path, frame)

                frames.append({
                    "path": frame_path,
                    "frame_id": frame_id,
                    "timestamp": round(timestamp, 2),
                    "timestamp_formatted": format_duration(timestamp),
                    "trigger": "scene_change",
                    "diff_score": round(mean_diff, 2)
                })
                last_captured = frame_id

                if len(frames) >= max_frames:
                    break

        prev_gray = gray

    cap.release()
    return frames

Hybrid strategy

def extract_frames_hybrid(
    video_path: str,
    interval_seconds: float = 10.0,
    scene_threshold: float = 30.0,
    max_frames: int = 25,
    output_dir: str = "/tmp/video_frames"
) -> list[dict]:
    interval_frames = extract_frames_by_interval(
        video_path,
        interval_seconds=interval_seconds,
        output_dir=output_dir + "/interval",
        max_frames=max_frames
    )

    scene_frames = extract_frames_by_scene_change(
        video_path,
        threshold=scene_threshold,
        output_dir=output_dir + "/scene",
        max_frames=max_frames
    )

    all_frames = interval_frames + scene_frames
    all_frames.sort(key=lambda f: f["timestamp"])

    deduplicated = []
    last_ts = -5.0
    for frame in all_frames:
        if frame["timestamp"] - last_ts >= 2.0:
            deduplicated.append(frame)
            last_ts = frame["timestamp"]

    return deduplicated[:max_frames]

Step 4: Frame Analysis with the Vision API

from openai import OpenAI
import base64

client = OpenAI()

def analyze_single_frame(frame_path: str, context: str = "") -> dict:
    with open(frame_path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode()

    prompt = "Describe this video scene in 2-3 sentences. Include: visible people, actions, objects, on-screen text, setting."
    if context:
        prompt += f"\nAdditional context: {context}"

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": prompt},
                {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}
            ]
        }],
        max_tokens=200,
        temperature=0.2
    )

    return {
        "description": response.choices[0].message.content,
        "tokens_used": response.usage.total_tokens
    }


def analyze_frames_batch(frames: list[dict], context: str = "") -> list[dict]:
    analyzed = []
    total_tokens = 0

    for i, frame in enumerate(frames):
        try:
            result = analyze_single_frame(frame["path"], context)
            analyzed.append({
                **frame,
                "description": result["description"],
                "status": "success"
            })
            total_tokens += result["tokens_used"]
        except Exception as e:
            analyzed.append({
                **frame,
                "description": None,
                "status": "error",
                "error": str(e)
            })

    return analyzed

Multi-frame analysis in a single request

To reduce calls, send multiple frames in one request:

def analyze_frame_group(frames: list[dict], max_per_request: int = 5) -> str:
    content = [{
        "type": "text",
        "text": (
            f"These are {len(frames)} frames from a video, in chronological order. "
            "For each frame, describe the scene in 1-2 sentences. "
            "Format: 'Frame N (MM:SS): description'"
        )
    }]

    for frame in frames[:max_per_request]:
        with open(frame["path"], "rb") as f:
            b64 = base64.b64encode(f.read()).decode()
        content.append({
            "type": "image_url",
            "image_url": {"url": f"data:image/jpeg;base64,{b64}"}
        })

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": content}],
        max_tokens=500
    )
    return response.choices[0].message.content


def analyze_all_frames_grouped(frames: list[dict], group_size: int = 5) -> list[str]:
    descriptions = []
    for i in range(0, len(frames), group_size):
        group = frames[i:i + group_size]
        group_desc = analyze_frame_group(group)
        descriptions.append(group_desc)
    return descriptions

Step 5: Synthesis with an LLM

def synthesize_video_summary(
    video_info: dict,
    frame_descriptions: list[dict]
) -> dict:
    successful = [f for f in frame_descriptions if f.get("status") == "success"]

    timeline = []
    for f in successful:
        timeline.append(f"{f['timestamp_formatted']}: {f['description']}")

    timeline_text = "\n".join(timeline)

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": (
                f"Video info: duration {video_info['duration_formatted']}, "
                f"{video_info['total_frames']} total frames.\n\n"
                f"Analysis of {len(successful)} key frames:\n\n"
                f"{timeline_text}\n\n"
                "Generate:\n"
                "1. General summary (3-5 sentences)\n"
                "2. Main sections with timestamps\n"
                "3. Key topics or actions detected\n"
                "4. Relevant observations"
            )
        }],
        max_tokens=600,
        temperature=0.3
    )

    return {
        "summary": response.choices[0].message.content,
        "frames_analyzed": len(successful),
        "frames_failed": len(frame_descriptions) - len(successful),
        "video_duration": video_info["duration_formatted"]
    }

Complete Pipeline

def analyze_video_pipeline(
    video_path: str,
    method: str = "hybrid",
    target_frames: int = 20,
    context: str = ""
) -> dict:
    video_info = get_video_info(video_path)
    if "error" in video_info:
        return video_info

    interval = calculate_optimal_interval(
        video_info["duration_seconds"],
        target_frames=target_frames
    )

    if method == "interval":
        frames = extract_frames_by_interval(
            video_path, interval_seconds=interval, max_frames=target_frames
        )
    elif method == "scene":
        frames = extract_frames_by_scene_change(
            video_path, max_frames=target_frames
        )
    else:
        frames = extract_frames_hybrid(
            video_path,
            interval_seconds=interval,
            max_frames=target_frames
        )

    analyzed = analyze_frames_batch(frames, context=context)

    summary = synthesize_video_summary(video_info, analyzed)
    summary["video_info"] = video_info
    summary["method"] = method
    summary["frames_extracted"] = len(frames)

    return summary

Usage:

result = analyze_video_pipeline(
    "recorded_class.mp4",
    method="hybrid",
    target_frames=15,
    context="Video of a university class about machine learning"
)
print(result["summary"])

Detecting Specific Actions

To search for specific moments in the video:

def detect_action_in_video(
    video_path: str,
    action_description: str,
    check_interval: float = 3.0,
    max_frames: int = 40
) -> list[dict]:
    frames = extract_frames_by_interval(
        video_path,
        interval_seconds=check_interval,
        max_frames=max_frames
    )

    detections = []

    for frame in frames:
        with open(frame["path"], "rb") as f:
            b64 = base64.b64encode(f.read()).decode()

        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": (
                            f"Does this image show: '{action_description}'?\n"
                            "Reply in JSON: {\"detected\": true/false, \"confidence\": 0.0-1.0, \"details\": \"...\"}"
                        )
                    },
                    {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}
                ]
            }],
            max_tokens=100,
            temperature=0,
            response_format={"type": "json_object"}
        )

        import json
        result = json.loads(response.choices[0].message.content)

        if result.get("detected"):
            detections.append({
                **frame,
                "confidence": result["confidence"],
                "details": result["details"]
            })

    return detections

Usage:

moments = detect_action_in_video(
    "presentation.mp4",
    "a person showing a chart or diagram"
)
for m in moments:
    print(f"{m['timestamp_formatted']}: {m['details']} (conf: {m['confidence']})")

Troubleshooting

Problem 1: Too many frames, high cost

Symptom: A 1-hour video generates 720 frames at 1 frame/5s.

Solution: Calculate the optimal interval before extracting:

info = get_video_info("long_video.mp4")
interval = calculate_optimal_interval(info["duration_seconds"], target_frames=20)

For long videos (>30 min), use scene detection to capture only significant changes.

Problem 2: Blurry or redundant frames

Symptom: Many frames are nearly identical or blurry.

Solution:

def filter_blurry_frames(frames: list[dict], blur_threshold: float = 100.0) -> list[dict]:
    filtered = []
    for frame in frames:
        img = cv2.imread(frame["path"])
        gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
        laplacian_var = cv2.Laplacian(gray, cv2.CV_64F).var()

        if laplacian_var >= blur_threshold:
            frame["sharpness"] = round(laplacian_var, 2)
            filtered.append(frame)

    return filtered

Problem 3: OpenCV doesn't open the video

Symptom: cap.isOpened() returns False.

Cause: Unsupported codec or ffmpeg not installed.

Solution:

# macOS
brew install ffmpeg

# Linux
sudo apt-get install ffmpeg libavcodec-extra

Verify:

def verify_video_support(video_path: str) -> dict:
    cap = cv2.VideoCapture(video_path)
    if not cap.isOpened():
        return {
            "supported": False,
            "suggestion": "Install ffmpeg: brew install ffmpeg (macOS) or apt-get install ffmpeg (Linux)"
        }
    cap.release()
    return {"supported": True}

Problem 4: Image-per-request limit

Symptom: Error when sending more than 10 images in one request to OpenAI.

Solution: Split into groups (already implemented in analyze_frame_group with max_per_request=5). To combine results from multiple groups:

def combine_group_descriptions(group_results: list[str]) -> str:
    combined = "\n\n".join(group_results)
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"Combine these frame descriptions into a coherent chronological summary:\n\n{combined}"
        }],
        max_tokens=500
    )
    return response.choices[0].message.content

Problem 5: Videos with relevant audio

Symptom: The frame analysis loses the audio context (narration, dialogue).

Solution: Combine with transcription (see capsule 05 — Multi-Modality Combinations):

def analyze_video_with_audio(video_path: str) -> dict:
    video_summary = analyze_video_pipeline(video_path, target_frames=15)

    audio_path = extract_audio_from_video(video_path)
    transcript = transcribe(audio_path)

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": (
                f"Visual analysis of the video:\n{video_summary['summary']}\n\n"
                f"Audio transcript:\n{transcript[:4000]}\n\n"
                "Generate a complete summary that integrates the visual with the spoken."
            )
        }],
        max_tokens=600
    )

    return {
        "integrated_summary": response.choices[0].message.content,
        "visual_summary": video_summary["summary"],
        "transcript_length": len(transcript)
    }

Exercises

Exercise 1: Key timestamp extractor

Create a function that analyzes a video and returns a list of timestamps where important changes occur, with a description of each change.

See solution 
def extract_key_timestamps(video_path: str, max_events: int = 10) -> list[dict]:
    frames = extract_frames_by_scene_change(video_path, max_frames=max_events * 2)

    analyzed = analyze_frames_batch(frames)
    successful = [f for f in analyzed if f.get("status") == "success"]

    events = []
    for f in successful:
        events.append({
            "timestamp": f["timestamp_formatted"],
            "seconds": f["timestamp"],
            "description": f["description"],
            "trigger": f.get("trigger", "interval")
        })

    return events[:max_events]

Exercise 2: Compare two video segments

Extract frames from two time ranges of the same video and compare what changed.

See solution 
def compare_video_segments(
    video_path: str,
    segment_a: tuple[float, float],
    segment_b: tuple[float, float],
    frames_per_segment: int = 5
) -> dict:
    cap = cv2.VideoCapture(video_path)
    fps = cap.get(cv2.CAP_PROP_FPS)

    def extract_segment(start_sec, end_sec):
        interval = (end_sec - start_sec) / frames_per_segment
        frames = []
        for i in range(frames_per_segment):
            ts = start_sec + i * interval
            cap.set(cv2.CAP_PROP_POS_FRAMES, int(ts * fps))
            ret, frame = cap.read()
            if ret:
                path = f"/tmp/segment_{start_sec}_{i}.jpg"
                cv2.imwrite(path, frame)
                frames.append({"path": path, "timestamp": round(ts, 2)})
        return frames

    frames_a = extract_segment(*segment_a)
    frames_b = extract_segment(*segment_b)
    cap.release()

    desc_a = analyze_frames_batch(frames_a)
    desc_b = analyze_frames_batch(frames_b)

    text_a = "\n".join(f["description"] for f in desc_a if f.get("description"))
    text_b = "\n".join(f["description"] for f in desc_b if f.get("description"))

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": (
                f"Segment A ({segment_a[0]}s - {segment_a[1]}s):\n{text_a}\n\n"
                f"Segment B ({segment_b[0]}s - {segment_b[1]}s):\n{text_b}\n\n"
                "What are the main differences between the two segments?"
            )
        }],
        max_tokens=400
    )

    return {
        "comparison": response.choices[0].message.content,
        "segment_a_frames": len(frames_a),
        "segment_b_frames": len(frames_b)
    }

Exercise 3: Video Q&A

Implement a system where the user asks questions about a video. Extract frames, analyze, and answer the specific question.

See solution 
def video_qa(video_path: str, question: str, target_frames: int = 15) -> dict:
    frames = extract_frames_hybrid(video_path, max_frames=target_frames)

    analyzed = analyze_frames_batch(frames, context=question)
    successful = [f for f in analyzed if f.get("status") == "success"]

    timeline = "\n".join(
        f"{f['timestamp_formatted']}: {f['description']}"
        for f in successful
    )

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "Answer questions about videos based ONLY on the provided frame analysis. If you can't answer with certainty, say so."
            },
            {
                "role": "user",
                "content": (
                    f"Video frame analysis:\n\n{timeline}\n\n"
                    f"Question: {question}"
                )
            }
        ],
        max_tokens=400,
        temperature=0.2
    )

    return {
        "answer": response.choices[0].message.content,
        "frames_analyzed": len(successful),
        "video_path": video_path
    }

Additional Resources

  1. OpenCV Documentation — Video processing
  2. OpenAI Vision Guide — Multiple images per request
  3. FFmpeg Documentation — Advanced video manipulation
  4. Scene Detection Libraries — Scene detection with PySceneDetect
← Previous3. Automated Image AnalysisNext →5. Multi-Modality Combinations