video-rag

What this skill does

# Video RAG

## Pipeline Overview

```
video.mp4
   |-- ffmpeg demux --------> audio.wav  --> Whisper --> transcript (timestamped segments)
   |                                                       |
   |                                                       v
   |                                         speaker-aware chunks
   |
   |-- PySceneDetect -------> scene list --> ffmpeg extract keyframes --> images
                                                                            |
                                                                            v
                                              CLIP / SigLIP / VoyageAI multimodal embeddings

Index:  text chunks (+ time) + image chunks (+ time) in vector store with shared schema.
Query:  embed text -> search text index AND multimodal index -> merge -> answer with timestamp.
```

## Keyframe Extraction

### ffmpeg — Fixed Interval

```bash
# One frame every 5 seconds, scaled to 512px wide
ffmpeg -i input.mp4 -vf "fps=1/5,scale=512:-1" -q:v 2 frames/f_%05d.jpg
```

### PySceneDetect — Shot Boundaries

```python
from scenedetect import open_video, SceneManager
from scenedetect.detectors import ContentDetector, AdaptiveDetector
from scenedetect.scene_manager import save_images

video = open_video("input.mp4")
mgr = SceneManager()
mgr.add_detector(ContentDetector(threshold=27.0))   # or AdaptiveDetector()
mgr.detect_scenes(video)

scenes = mgr.get_scene_list()
for i, (start, end) in enumerate(scenes):
    print(i, start.get_seconds(), end.get_seconds())

save_images(
    scenes,
    video,
    num_images=1,          # 1 representative frame per scene
    image_name_template="scene-$SCENE_NUMBER",
    output_dir="keyframes",
)
```

### ffmpeg — Scene-Change Detection (no Python)

```bash
# Dump frames at scene cuts with timestamps
ffmpeg -i input.mp4 -vf "select='gt(scene,0.3)',showinfo" \
       -vsync vfr keyframes/kf_%04d.jpg 2> scene-log.txt
```

## Audio Track Extraction

```python
import subprocess

def extract_audio(video_path: str, wav_path: str) -> None:
    # 16 kHz mono WAV - ideal for Whisper
    subprocess.run([
        "ffmpeg", "-y", "-i", video_path,
        "-vn", "-ac", "1", "-ar", "16000", "-c:a", "pcm_s16le",
        wav_path,
    ], check=True)
```

## Whisper Transcription

See `audio-transcription` skill for the full engine comparison. Minimal example:

```python
from faster_whisper import WhisperModel

model = WhisperModel("large-v3", device="cuda", compute_type="float16")
segments, info = model.transcribe("audio.wav", vad_filter=True,
                                  word_timestamps=True)

transcript_segments = [
    {"start": s.start, "end": s.end, "text": s.text.strip()}
    for s in segments
]
```

## Visual Embeddings

### CLIP (OpenAI / open_clip)

```python
import torch
from PIL import Image
import open_clip

device = "cuda" if torch.cuda.is_available() else "cpu"
model, _, preprocess = open_clip.create_model_and_transforms(
    "ViT-L-14", pretrained="openai"
)
model = model.to(device).eval()
tokenizer = open_clip.get_tokenizer("ViT-L-14")

def embed_image(path: str) -> list[float]:
    img = preprocess(Image.open(path).convert("RGB")).unsqueeze(0).to(device)
    with torch.no_grad():
        vec = model.encode_image(img)
        vec = vec / vec.norm(dim=-1, keepdim=True)
    return vec[0].cpu().tolist()

def embed_text(text: str) -> list[float]:
    tokens = tokenizer([text]).to(device)
    with torch.no_grad():
        vec = model.encode_text(tokens)
        vec = vec / vec.norm(dim=-1, keepdim=True)
    return vec[0].cpu().tolist()
```

### SigLIP (better zero-shot than CLIP)

```python
from transformers import AutoProcessor, AutoModel
import torch

processor = AutoProcessor.from_pretrained("google/siglip-so400m-patch14-384")
model = AutoModel.from_pretrained("google/siglip-so400m-patch14-384").to("cuda").eval()

def embed_image_siglip(path: str) -> list[float]:
    inputs = processor(images=Image.open(path).convert("RGB"), return_tensors="pt").to("cuda")
    with torch.no_grad():
        feats = model.get_image_features(**inputs)
        feats = feats / feats.norm(dim=-1, keepdim=True)
    return feats[0].cpu().tolist()
```

### VoyageAI Multimodal (API)

```python
import os, base64, voyageai

vo = voyageai.Client(api_key=os.environ["VOYAGE_API_KEY"])

def encode_image_b64(path: str) -> str:
    with open(path, "rb") as f:
        return "data:image/jpeg;base64," + base64.b64encode(f.read()).decode()

result = vo.multimodal_embed(
    inputs=[[encode_image_b64("keyframes/scene-0001.jpg"), "a laptop on a desk"]],
    model="voyage-multimodal-3",
    input_type="document",
)
vectors = result.embeddings
```

## Unified Chunk Schema

```python
from dataclasses import dataclass
from typing import Literal

@dataclass
class VideoChunk:
    source: str
    modality: Literal["transcript", "visual"]
    start: float
    end: float
    text: str | None           # transcript text or caption
    image_path: str | None     # keyframe on disk or blob URL
    vector: list[float]
    metadata: dict             # speaker, scene_id, etc.
```

Both modalities share the schema so they can live in the same index (with a `modality` filter) or in two parallel indexes merged at query time.

## Chunking + Embedding

```python
def build_video_chunks(video_path: str) -> list[VideoChunk]:
    # 1. Audio
    extract_audio(video_path, "audio.wav")
    model = WhisperModel("large-v3", device="cuda", compute_type="float16")
    segments, info = model.transcribe("audio.wav", vad_filter=True,
                                      word_timestamps=True)
    transcript_segments = [
        {"start": s.start, "end": s.end, "text": s.text.strip()}
        for s in segments
    ]

    # 2. Scenes + keyframes
    video = open_video(video_path)
    mgr = SceneManager()
    mgr.add_detector(ContentDetector(threshold=27.0))
    mgr.detect_scenes(video)
    scenes = mgr.get_scene_list()
    save_images(scenes, video, num_images=1,
                image_name_template="$SCENE_NUMBER",
                output_dir="keyframes")

    # 3. Text chunks
    from openai import OpenAI
    oa = OpenAI()
    text_chunks = []
    for seg in merge_into_windows(transcript_segments, max_chars=900):
        emb = oa.embeddings.create(model="text-embedding-3-large",
                                   input=seg["text"]).data[0].embedding
        text_chunks.append(VideoChunk(
            source=video_path, modality="transcript",
            start=seg["start"], end=seg["end"],
            text=seg["text"], image_path=None,
            vector=emb, metadata={"language": info.language},
        ))

    # 4. Visual chunks
    visual_chunks = []
    for i, (start, end) in enumerate(scenes, start=1):
        img_path = f"keyframes/{i:03d}.jpg"
        vec = embed_image(img_path)
        visual_chunks.append(VideoChunk(
            source=video_path, modality="visual",
            start=start.get_seconds(), end=end.get_seconds(),
            text=None, image_path=img_path,
            vector=vec, metadata={"scene_id": i},
        ))
    return text_chunks + visual_chunks
```

## Multi-Modal Retrieval

Two strategies:

**1. Shared CLIP-space index** (requires CLIP text embeddings for text chunks too, or caption the transcript first). Query text -> embed with CLIP text encoder -> single search.

**2. Dual-index fusion** (recommended): text chunks in a text-embedding space, visual chunks in CLIP/SigLIP space. Query both, combine scores.

```python
def hybrid_search(query: str, k: int = 10) -> list[VideoChunk]:
    # Text side
    text_vec = oa.embeddings.create(
        model="text-embedding-3-large", input=query
    ).data[0].embedding
    text_hits = text_index.query(text_vec, k=k)

    # Visual side
    vis_vec = embed_text(query)   # CLIP text encoder
    vis_hits = visual_index.query(vis_vec, k=k)

    # Reciprocal Rank Fusion
    def rrf(ranks: list[list], k_rrf: int = 60) -> dict:
        scores = {}
        for results in