Módulo 7: Casos de Uso

4. Video: Frames + LLM

Siguiente →

Descripción

Los LLMs multimodales no procesan video directamente (todavía). La técnica estándar es extraer frames representativos del video, enviarlos como imágenes a una Vision API, y combinar los análisis individuales en un resumen coherente. En esta cápsula construyes un pipeline completo de análisis de video: extracción de frames con OpenCV, selección inteligente de frames clave, análisis con Vision API, y síntesis con LLM.

Por qué importa: Video es la modalidad más rica en información y la más difícil de procesar. Plataformas educativas analizan clases grabadas, sistemas de seguridad revisan cámaras, equipos de marketing analizan contenido, y plataformas de e-commerce procesan video-reviews. El patrón frames → Vision → LLM es la base de todos estos casos.

Conexión con el módulo: Este pipeline es otro destino del Use Case Selector (cápsula 08). Cuando el router detecta un archivo de video, aplica la extracción de frames y el análisis que construyes aquí. Los patrones de batch y rate limit de la cápsula 06 son críticos para procesar muchos frames.

Pipeline Visual

┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│    Video     │────▶│   Extraer    │────▶│  Seleccionar │
│   (.mp4)     │     │   frames     │     │  key frames  │
└──────────────┘     └──────────────┘     └──────┬───────┘
                                                  │
                                                  ▼
┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│   Resumen    │◀────│   Sintetizar │◀────│  Vision API  │
│   final      │     │   con LLM    │     │  por frame   │
└──────────────┘     └──────────────┘     └──────────────┘

Etapas:

  1. Cargar video — Abrir con OpenCV
  2. Extraer frames — Por intervalo fijo o detección de cambio de escena
  3. Seleccionar key frames — Filtrar duplicados y frames irrelevantes
  4. Analizar con Vision — Enviar cada frame a GPT-4o para descripción
  5. Sintetizar — LLM combina descripciones en resumen temporal

Paso 1: Información del Video

Antes de extraer frames, obtener metadata del video:

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"No se puede abrir: {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}"

Paso 2: Extracción de Frames por Intervalo

Extracción básica

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

Cálculo de intervalo óptimo

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))

Paso 3: Detección de Cambio de Escena

En lugar de extraer a intervalo fijo (que puede capturar frames redundantes), detectar cambios visuales significativos:

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

Estrategia híbrida

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]

Paso 4: Análisis de Frames con 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 esta escena de video en 2-3 oraciones. Incluye: personas visibles, acciones, objetos, texto en pantalla, ambiente."
    if context:
        prompt += f"\nContexto adicional: {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

Análisis multi-frame en un solo request

Para reducir llamadas, enviar múltiples frames en un request:

def analyze_frame_group(frames: list[dict], max_per_request: int = 5) -> str:
    content = [{
        "type": "text",
        "text": (
            f"Estos son {len(frames)} frames de un video, en orden cronológico. "
            "Para cada frame, describe la escena en 1-2 oraciones. "
            "Formato: 'Frame N (MM:SS): descripción'"
        )
    }]

    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

Paso 5: Síntesis con 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"Información del video: duración {video_info['duration_formatted']}, "
                f"{video_info['total_frames']} frames totales.\n\n"
                f"Análisis de {len(successful)} frames clave:\n\n"
                f"{timeline_text}\n\n"
                "Genera:\n"
                "1. Resumen general (3-5 oraciones)\n"
                "2. Secciones principales con timestamps\n"
                "3. Temas o acciones clave detectadas\n"
                "4. Observaciones relevantes"
            )
        }],
        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"]
    }

Pipeline Completo

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

Uso:

result = analyze_video_pipeline(
    "clase_grabada.mp4",
    method="hybrid",
    target_frames=15,
    context="Video de una clase universitaria sobre machine learning"
)
print(result["summary"])

Detección de Acciones Específicas

Para buscar momentos específicos en el 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"¿Esta imagen muestra: '{action_description}'?\n"
                            "Responde en 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

Uso:

moments = detect_action_in_video(
    "presentacion.mp4",
    "persona mostrando un gráfico o diagrama"
)
for m in moments:
    print(f"{m['timestamp_formatted']}: {m['details']} (conf: {m['confidence']})")

Troubleshooting

Problema 1: Demasiados frames, costo alto

Síntoma: Un video de 1 hora genera 720 frames a 1 frame/5s.

Solución: Calcular intervalo óptimo antes de extraer:

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

Para videos largos (>30 min), usar scene detection para capturar solo cambios significativos.

Problema 2: Frames borrosos o redundantes

Síntoma: Muchos frames son casi idénticos o están borrosos.

Solución:

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

Problema 3: OpenCV no abre el video

Síntoma: cap.isOpened() retorna False.

Causa: Codec no soportado o ffmpeg no instalado.

Solución:

# macOS
brew install ffmpeg

# Linux
sudo apt-get install ffmpeg libavcodec-extra

Verificar:

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

Problema 4: Límite de imágenes por request

Síntoma: Error al enviar más de 10 imágenes en un request a OpenAI.

Solución: Dividir en grupos (ya implementado en analyze_frame_group con max_per_request=5). Para combinar resultados de múltiples grupos:

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"Combina estas descripciones de frames en un resumen cronológico coherente:\n\n{combined}"
        }],
        max_tokens=500
    )
    return response.choices[0].message.content

Problema 5: Videos con audio relevante

Síntoma: El análisis de frames pierde el contexto del audio (narración, diálogos).

Solución: Combinar con transcripción (ver cápsula 05 — Combinaciones Multi-Modalidad):

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"Análisis visual del video:\n{video_summary['summary']}\n\n"
                f"Transcripción del audio:\n{transcript[:4000]}\n\n"
                "Genera un resumen completo que integre lo visual con lo hablado."
            )
        }],
        max_tokens=600
    )

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

Ejercicios

Ejercicio 1: Extractor de timestamps clave

Crea una función que analice un video y devuelva una lista de timestamps donde ocurren cambios importantes, con descripción de cada cambio.

Ver solución 
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]

Ejercicio 2: Comparar dos segmentos de video

Extrae frames de dos rangos temporales del mismo video y compara qué cambió.

Ver solución 
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"Segmento A ({segment_a[0]}s - {segment_a[1]}s):\n{text_a}\n\n"
                f"Segmento B ({segment_b[0]}s - {segment_b[1]}s):\n{text_b}\n\n"
                "¿Qué diferencias principales hay entre los dos segmentos?"
            )
        }],
        max_tokens=400
    )

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

Ejercicio 3: Video Q&A

Implementa un sistema donde el usuario hace preguntas sobre un video. Extrae frames, analiza, y responde la pregunta específica.

Ver solución 
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": "Responde preguntas sobre videos basándote SOLO en el análisis de frames proporcionado. Si no puedes responder con certeza, dilo."
            },
            {
                "role": "user",
                "content": (
                    f"Análisis de frames del video:\n\n{timeline}\n\n"
                    f"Pregunta: {question}"
                )
            }
        ],
        max_tokens=400,
        temperature=0.2
    )

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

Recursos Adicionales

  1. OpenCV Documentation — Procesamiento de video
  2. OpenAI Vision Guide — Múltiples imágenes por request
  3. FFmpeg Documentation — Manipulación avanzada de video
  4. Scene Detection Libraries — Detección de escenas con PySceneDetect
← Anterior3. Image Analysis AutomatizadoSiguiente →5. Combinaciones Multi-Modalidad