Video RAG: Videos indexieren und durchsuchen

Video kombiniert Audio, visuelle Inhalte und Text. Es ist das reichhaltigste, aber auch das komplexeste Format, das für ein RAG-System indexiert werden muss. Dieser Leitfaden zeigt Ihnen, wie Sie beliebige Videoinhalte zerlegen, analysieren und durchsuchbar machen.

Warum Video RAG?

Die Herausforderung der Videodaten

• Explosives Volumen: 500 Stunden Video werden jede Minute auf YouTube hochgeladen
• Informationsreichtum: Ein Video-Tutorial enthält mehr als nur seine Transkription
• Zeitlichkeit: Die Information ist über die Zeit verteilt
• Multimodalität: Audio + Visuell + Bildschirmtext

Konkrete Anwendungsfälle

Typischer ROI

• 80% Reduktion der Suchzeit in Videoarchiven
• +60% Engagement bei Lerninhalten (dank automatischer Kapitel)
• Compliance: Durchsuchbarer Video-Nachweis für Audits

Video-RAG-Architektur

┌─────────────────────────────────────────────────────────────────┐
│                     VIDEO RAG PIPELINE                           │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  ┌──────────┐                                                   │
│  │  Video   │                                                   │
│  │  Input   │                                                   │
│  └────┬─────┘                                                   │
│       │                                                          │
│       ├─────────────────┬─────────────────┬────────────────┐    │
│       ▼                 ▼                 ▼                ▼    │
│  ┌──────────┐    ┌──────────────┐  ┌──────────────┐ ┌────────┐ │
│  │  Audio   │    │   Frames     │  │   OCR        │ │Metadata│ │
│  │Extraction│    │  Sampling    │  │ (Bildschirm) │ │        │ │
│  └────┬─────┘    └──────┬───────┘  └──────┬───────┘ └───┬────┘ │
│       │                 │                 │             │       │
│       ▼                 ▼                 ▼             │       │
│  ┌──────────┐    ┌──────────────┐  ┌──────────────┐    │       │
│  │ Whisper  │    │ Vision Model │  │  Text Index  │    │       │
│  │Transcribe│    │  (GPT-4V)    │  │              │    │       │
│  └────┬─────┘    └──────┬───────┘  └──────┬───────┘    │       │
│       │                 │                 │             │       │
│       └────────┬────────┴────────┬────────┘             │       │
│                ▼                 │                      │       │
│       ┌────────────────┐        │                      │       │
│       │ Scene Detection│        │                      │       │
│       │  & Chaptering  │        │                      │       │
│       └───────┬────────┘        │                      │       │
│               │                 │                      │       │
│               ▼                 ▼                      ▼       │
│       ┌──────────────────────────────────────────────────┐     │
│       │              Multimodal Fusion                    │     │
│       │         (text + visual + audio embeddings)        │     │
│       └────────────────────────┬─────────────────────────┘     │
│                                ▼                                │
│       ┌──────────────────────────────────────────────────┐     │
│       │              Vector Store (Qdrant)                │     │
│       └──────────────────────────────────────────────────┘     │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

Extraktion und Zerlegung

Audio-Extraktion

DEVELOPERpython
import subprocess
from pathlib import Path

def extract_audio(video_path: str, output_path: str = None) -> str:
    """Extrahiert die Audiospur aus einem Video."""
    if output_path is None:
        output_path = str(Path(video_path).with_suffix('.wav'))

    cmd = [
        'ffmpeg', '-i', video_path,
        '-vn',  # Kein Video
        '-acodec', 'pcm_s16le',
        '-ar', '16000',  # 16kHz für Whisper
        '-ac', '1',  # Mono
        '-y',  # Überschreiben
        output_path
    ]

    subprocess.run(cmd, capture_output=True, check=True)
    return output_path

Frame-Extraktion

DEVELOPERpython
import cv2
from dataclasses import dataclass
from typing import List
import numpy as np

@dataclass
class VideoFrame:
    timestamp: float
    frame_number: int
    image: np.ndarray
    is_keyframe: bool

class FrameExtractor:
    def __init__(self, video_path: str):
        self.video_path = video_path
        self.cap = cv2.VideoCapture(video_path)
        self.fps = self.cap.get(cv2.CAP_PROP_FPS)
        self.total_frames = int(self.cap.get(cv2.CAP_PROP_FRAME_COUNT))
        self.duration = self.total_frames / self.fps

    def extract_at_interval(self, interval_seconds: float = 1.0) -> List[VideoFrame]:
        """Extrahiert alle N Sekunden ein Frame."""
        frames = []
        frame_interval = int(self.fps * interval_seconds)

        for frame_num in range(0, self.total_frames, frame_interval):
            self.cap.set(cv2.CAP_PROP_POS_FRAMES, frame_num)
            ret, frame = self.cap.read()

            if ret:
                frames.append(VideoFrame(
                    timestamp=frame_num / self.fps,
                    frame_number=frame_num,
                    image=frame,
                    is_keyframe=False
                ))

        return frames

    def extract_keyframes(self, threshold: float = 30.0) -> List[VideoFrame]:
        """
        Extrahiert die Keyframes (signifikante Szenenwechsel).
        Nutzt die Histogrammdifferenz zwischen aufeinanderfolgenden Frames.
        """
        keyframes = []
        prev_hist = None

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

            # Histogramm berechnen
            gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
            hist = cv2.calcHist([gray], [0], None, [256], [0, 256])
            hist = cv2.normalize(hist, hist).flatten()

            if prev_hist is not None:
                # Mit vorherigem Frame vergleichen
                diff = cv2.compareHist(prev_hist, hist, cv2.HISTCMP_BHATTACHARYYA)
                is_keyframe = diff > threshold / 100

                if is_keyframe:
                    keyframes.append(VideoFrame(
                        timestamp=frame_num / self.fps,
                        frame_number=frame_num,
                        image=frame,
                        is_keyframe=True
                    ))
            else:
                # Erstes Frame = immer Keyframe
                keyframes.append(VideoFrame(
                    timestamp=0,
                    frame_number=0,
                    image=frame,
                    is_keyframe=True
                ))

            prev_hist = hist
            frame_num += 1

        return keyframes

    def __del__(self):
        self.cap.release()

Szenenerkennung mit PySceneDetect

DEVELOPERpython
from scenedetect import detect, ContentDetector, split_video_ffmpeg

def detect_scenes(video_path: str, threshold: float = 27.0) -> List[dict]:
    """
    Erkennt Szenenwechsel in einem Video.
    Gibt die Start-/End-Timestamps jeder Szene zurück.
    """
    scene_list = detect(video_path, ContentDetector(threshold=threshold))

    scenes = []
    for i, scene in enumerate(scene_list):
        scenes.append({
            "scene_number": i + 1,
            "start_time": scene[0].get_seconds(),
            "end_time": scene[1].get_seconds(),
            "start_frame": scene[0].get_frames(),
            "end_frame": scene[1].get_frames(),
            "duration": scene[1].get_seconds() - scene[0].get_seconds()
        })

    return scenes

Multimodale Frame-Analyse

Frame-Beschreibung mit GPT-4V

DEVELOPERpython
import base64
from openai import OpenAI

def analyze_frame(
    frame: np.ndarray,
    context: str = "",
    client: OpenAI = None
) -> dict:
    """Analysiert ein Video-Frame mit GPT-4V."""
    if client is None:
        client = OpenAI()

    # In base64 kodieren
    _, buffer = cv2.imencode('.jpg', frame)
    img_base64 = base64.b64encode(buffer).decode('utf-8')

    prompt = """Analysiere dieses Video-Frame für ein RAG-System. Beschreibe:

1. **Hauptinhalt**: Was zeigt dieses Frame?
2. **Sichtbarer Text**: Jeglicher Bildschirmtext (Titel, Untertitel, UI)
3. **Visuelle Elemente**: Grafiken, Diagramme, Demonstrationen
4. **Kontext**: Ist es eine Intro, eine Demo, ein Fazit?

Sei präzise und sachlich. Ziel ist es, die Suche zu ermöglichen."""

    if context:
        prompt += f"\n\nKontext: {context}"

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

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

Batch-Verarbeitung

DEVELOPERpython
async def analyze_frames_batch(
    frames: List[VideoFrame],
    client: OpenAI,
    max_concurrent: int = 5
) -> List[dict]:
    """Analysiert mehrere Frames parallel."""
    import asyncio
    from openai import AsyncOpenAI

    async_client = AsyncOpenAI()
    semaphore = asyncio.Semaphore(max_concurrent)

    async def analyze_one(frame: VideoFrame) -> dict:
        async with semaphore:
            # Kodieren
            _, buffer = cv2.imencode('.jpg', frame.image)
            img_base64 = base64.b64encode(buffer).decode('utf-8')

            response = await async_client.chat.completions.create(
                model="gpt-4o-mini",  # Wirtschaftlicher für Batch-Verarbeitung
                messages=[{
                    "role": "user",
                    "content": [
                        {"type": "text", "text": "Beschreibe dieses Video-Frame kurz."},
                        {
                            "type": "image_url",
                            "image_url": {
                                "url": f"data:image/jpeg;base64,{img_base64}",
                                "detail": "low"
                            }
                        }
                    ]
                }],
                max_tokens=200
            )

            return {
                "timestamp": frame.timestamp,
                "description": response.choices[0].message.content
            }

    tasks = [analyze_one(f) for f in frames]
    results = await asyncio.gather(*tasks)

    return results

Vollständige Indexierungs-Pipeline

Datenstruktur

DEVELOPERpython
from dataclasses import dataclass, field
from typing import Optional, List

@dataclass
class VideoSegment:
    """Repräsentiert ein indexierbares Video-Segment."""
    segment_id: str
    video_id: str
    video_title: str

    # Zeitlich
    start_time: float
    end_time: float
    duration: float

    # Inhalt
    transcript: str
    frame_descriptions: List[str]
    on_screen_text: Optional[str]

    # Semantisch
    topic: Optional[str]
    summary: Optional[str]
    keywords: List[str] = field(default_factory=list)

    # Metadaten
    speaker: Optional[str] = None
    scene_type: Optional[str] = None  # intro, demo, explanation, outro

    def to_embedding_text(self) -> str:
        """Kombinierter Text für das Embedding."""
        parts = []

        if self.topic:
            parts.append(f"Topic: {self.topic}")

        if self.summary:
            parts.append(f"Summary: {self.summary}")

        parts.append(f"Transcript: {self.transcript}")

        if self.frame_descriptions:
            parts.append(f"Visual: {' '.join(self.frame_descriptions[:3])}")

        if self.on_screen_text:
            parts.append(f"On-screen text: {self.on_screen_text}")

        return "\n".join(parts)

Vollständige Indexierungs-Pipeline

DEVELOPERpython
from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance, PointStruct
import hashlib

class VideoRAGPipeline:
    def __init__(self):
        self.qdrant = QdrantClient(url="http://localhost:6333")
        self.openai = OpenAI()
        self.collection_name = "video_rag"

    def create_collection(self):
        """Erstellt die Collection mit multimodalen Embeddings."""
        self.qdrant.recreate_collection(
            collection_name=self.collection_name,
            vectors_config={
                "text": VectorParams(size=1536, distance=Distance.COSINE),
                "visual": VectorParams(size=768, distance=Distance.COSINE)  # CLIP
            }
        )

    def process_video(self, video_path: str, title: str) -> List[VideoSegment]:
        """Vollständige Video-Verarbeitungs-Pipeline."""
        video_id = hashlib.md5(video_path.encode()).hexdigest()

        print("1. Audio-Extraktion...")
        audio_path = extract_audio(video_path)

        print("2. Transkription...")
        transcriber = AudioTranscriber()
        transcription = transcriber.transcribe(audio_path)

        print("3. Szenenerkennung...")
        scenes = detect_scenes(video_path)

        print("4. Keyframe-Extraktion...")
        extractor = FrameExtractor(video_path)
        keyframes = extractor.extract_keyframes()

        print("5. Frame-Analyse...")
        frame_analyses = []
        for kf in keyframes[:20]:  # Begrenzung wegen der Kosten
            analysis = analyze_frame(kf.image, client=self.openai)
            frame_analyses.append({
                "timestamp": kf.timestamp,
                **analysis
            })

        print("6. Erstellung der Segmente...")
        segments = self._create_segments(
            video_id=video_id,
            video_title=title,
            transcription=transcription,
            scenes=scenes,
            frame_analyses=frame_analyses
        )

        print("7. Generierung der Topics...")
        segments = self._add_topics(segments)

        return segments

    def _create_segments(
        self,
        video_id: str,
        video_title: str,
        transcription: dict,
        scenes: List[dict],
        frame_analyses: List[dict]
    ) -> List[VideoSegment]:
        """Erstellt die Segmente aus den extrahierten Daten."""
        segments = []

        for scene in scenes:
            # Passende Transkription finden
            scene_transcript = []
            for seg in transcription["segments"]:
                if seg["start"] >= scene["start_time"] and seg["end"] <= scene["end_time"]:
                    scene_transcript.append(seg["text"])

            # Frame-Beschreibungen finden
            frame_descs = []
            for fa in frame_analyses:
                if scene["start_time"] <= fa["timestamp"] <= scene["end_time"]:
                    frame_descs.append(fa["description"])

            segment = VideoSegment(
                segment_id=f"{video_id}_{scene['scene_number']}",
                video_id=video_id,
                video_title=video_title,
                start_time=scene["start_time"],
                end_time=scene["end_time"],
                duration=scene["duration"],
                transcript=" ".join(scene_transcript),
                frame_descriptions=frame_descs,
                on_screen_text=None,  # OCR bei Bedarf ergänzen
                topic=None,
                summary=None
            )
            segments.append(segment)

        return segments

    def _add_topics(self, segments: List[VideoSegment]) -> List[VideoSegment]:
        """Ergänzt Topics und Summaries via LLM."""
        for segment in segments:
            if not segment.transcript:
                continue

            prompt = f"""Analysiere dieses Video-Segment:

Transkription: {segment.transcript[:1000]}
Visuelle Inhalte: {' '.join(segment.frame_descriptions[:2]) if segment.frame_descriptions else 'N/A'}

Generiere:
1. Einen Topic-Titel (5-10 Wörter)
2. Eine Zusammenfassung (1-2 Sätze)
3. 3-5 Schlagwörter

JSON-Format: {{"topic": "", "summary": "", "keywords": []}}"""

            response = self.openai.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": prompt}],
                response_format={"type": "json_object"}
            )

            import json
            data = json.loads(response.choices[0].message.content)
            segment.topic = data.get("topic")
            segment.summary = data.get("summary")
            segment.keywords = data.get("keywords", [])

        return segments

    def index_segments(self, segments: List[VideoSegment]):
        """Indexiert die Segmente in Qdrant."""
        from sentence_transformers import SentenceTransformer

        # CLIP-Modell für visuelle Embeddings
        clip_model = SentenceTransformer('clip-ViT-B-32')

        points = []
        for segment in segments:
            # Text-Embedding
            text = segment.to_embedding_text()
            text_response = self.openai.embeddings.create(
                model="text-embedding-3-small",
                input=text
            )
            text_embedding = text_response.data[0].embedding

            # Visuelles Embedding (Durchschnitt der Beschreibungen)
            if segment.frame_descriptions:
                visual_text = " ".join(segment.frame_descriptions)
                visual_embedding = clip_model.encode(visual_text).tolist()
            else:
                visual_embedding = [0.0] * 768

            point = PointStruct(
                id=hash(segment.segment_id) % (2**63),
                vector={
                    "text": text_embedding,
                    "visual": visual_embedding
                },
                payload={
                    "segment_id": segment.segment_id,
                    "video_id": segment.video_id,
                    "video_title": segment.video_title,
                    "start_time": segment.start_time,
                    "end_time": segment.end_time,
                    "duration": segment.duration,
                    "transcript": segment.transcript,
                    "topic": segment.topic,
                    "summary": segment.summary,
                    "keywords": segment.keywords
                }
            )
            points.append(point)

        self.qdrant.upsert(
            collection_name=self.collection_name,
            points=points
        )

        print(f"{len(points)} Segmente indexiert")

Suche und Generierung

Hybride Video-Suche

DEVELOPERpython
def search_video_rag(
    query: str,
    pipeline: VideoRAGPipeline,
    search_type: str = "hybrid",  # text, visual, hybrid
    limit: int = 5
) -> List[dict]:
    """Durchsucht die indexierten Videos."""

    # Embedding der Anfrage
    text_response = pipeline.openai.embeddings.create(
        model="text-embedding-3-small",
        input=query
    )
    text_embedding = text_response.data[0].embedding

    if search_type == "text":
        results = pipeline.qdrant.search(
            collection_name=pipeline.collection_name,
            query_vector=("text", text_embedding),
            limit=limit
        )
    elif search_type == "visual":
        from sentence_transformers import SentenceTransformer
        clip = SentenceTransformer('clip-ViT-B-32')
        visual_embedding = clip.encode(query).tolist()

        results = pipeline.qdrant.search(
            collection_name=pipeline.collection_name,
            query_vector=("visual", visual_embedding),
            limit=limit
        )
    else:  # hybrid
        # RRF über beide Suchen
        text_results = pipeline.qdrant.search(
            collection_name=pipeline.collection_name,
            query_vector=("text", text_embedding),
            limit=limit * 2
        )

        from sentence_transformers import SentenceTransformer
        clip = SentenceTransformer('clip-ViT-B-32')
        visual_embedding = clip.encode(query).tolist()

        visual_results = pipeline.qdrant.search(
            collection_name=pipeline.collection_name,
            query_vector=("visual", visual_embedding),
            limit=limit * 2
        )

        # RRF-Fusion
        scores = {}
        for rank, r in enumerate(text_results):
            scores[r.payload["segment_id"]] = {
                "score": 0.6 / (rank + 60),
                "payload": r.payload
            }
        for rank, r in enumerate(visual_results):
            sid = r.payload["segment_id"]
            if sid in scores:
                scores[sid]["score"] += 0.4 / (rank + 60)
            else:
                scores[sid] = {
                    "score": 0.4 / (rank + 60),
                    "payload": r.payload
                }

        sorted_results = sorted(scores.items(), key=lambda x: x[1]["score"], reverse=True)
        results = [{"payload": v["payload"], "score": v["score"]} for _, v in sorted_results[:limit]]

    return [
        {
            "video_title": r.payload["video_title"] if hasattr(r, 'payload') else r["payload"]["video_title"],
            "topic": r.payload["topic"] if hasattr(r, 'payload') else r["payload"]["topic"],
            "transcript": r.payload["transcript"][:200] + "..." if hasattr(r, 'payload') else r["payload"]["transcript"][:200] + "...",
            "timestamp": f"{r.payload['start_time']:.0f}s - {r.payload['end_time']:.0f}s" if hasattr(r, 'payload') else f"{r['payload']['start_time']:.0f}s - {r['payload']['end_time']:.0f}s",
            "score": r.score if hasattr(r, 'score') else r["score"]
        }
        for r in results
    ]

Antwortgenerierung mit Timestamp

DEVELOPERpython
def generate_video_answer(
    query: str,
    retrieved_segments: List[dict],
    client: OpenAI
) -> str:
    """Generiert eine Antwort mit Video-Referenzen."""

    context = "\n\n".join([
        f"**{s['video_title']}** [{s['timestamp']}]\n"
        f"Topic: {s['topic']}\n"
        f"Inhalt: {s['transcript']}"
        for s in retrieved_segments
    ])

    prompt = f"""Du bist ein Assistent, der Fragen anhand von Videos als Quelle beantwortet.

Verfügbare Video-Segmente:
{context}

Frage: {query}

Anweisungen:
1. Stütze deine Antwort ausschließlich auf die bereitgestellten Segmente
2. Zitiere deine Quellen mit [Video: Titel, Timestamp]
3. Wenn sich die Frage auf ein visuelles Element bezieht, erwähne es
4. Schlage vor, zum exakten Timestamp zu springen, falls relevant"""

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=1000
    )

    return response.choices[0].message.content

Fortgeschrittene Optimierungen

Automatisches YouTube-Kapitel

DEVELOPERpython
def generate_youtube_chapters(
    segments: List[VideoSegment]
) -> str:
    """Generiert Kapitel im YouTube-Format."""
    chapters = []

    for segment in segments:
        if segment.topic:
            # Format: MM:SS Titel
            minutes = int(segment.start_time // 60)
            seconds = int(segment.start_time % 60)
            chapters.append(f"{minutes:02d}:{seconds:02d} {segment.topic}")

    return "\n".join(chapters)

# Beispielausgabe:
# 00:00 Einführung
# 02:15 Installation und Konfiguration
# 05:30 Erstes praktisches Beispiel
# 10:45 Fortgeschrittene Anwendungsfälle
# 15:20 Fazit und Ressourcen

Extraktion von Schlüsselmomenten

DEVELOPERpython
def extract_highlight_moments(
    segments: List[VideoSegment],
    client: OpenAI
) -> List[dict]:
    """Identifiziert die Höhepunkte des Videos."""

    all_content = "\n\n".join([
        f"[{s.start_time:.0f}s-{s.end_time:.0f}s] {s.topic}: {s.transcript[:300]}"
        for s in segments
    ])

    prompt = f"""Analysiere dieses Video und identifiziere die 5 wichtigsten Momente:

{all_content}

Gib für jeden Moment an:
- timestamp (Sekunden)
- type: tutorial_step, key_insight, demo, announcement, qa
- kurze Beschreibung
- warum es wichtig ist

JSON: [{{"timestamp": X, "type": "", "description": "", "importance": ""}}]"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"}
    )

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

Kosten und Performance

Kosten pro Stunde Video

Verarbeitungszeit

Speicherung

• 1 Stunde Video = ~50-100 Segmente
• Embeddings: ~1MB
• Metadaten: ~100KB
• Thumbnails (optional): ~5MB

Integration mit Ailog

Ailog unterstützt native Video-Indexierung:

1. Video-Upload: MP4, MOV, WEBM, AVI
2. Automatisches Processing: Transkription + Szenen + Keyframes
3. Intelligente Kapitelbildung: Topics werden automatisch generiert
4. Einheitliche Suche: "In welchem Video wird über X gesprochen?"

Testen Sie Video RAG bei Ailog

Verwandte Leitfäden

• Vollständiger Leitfaden zu multimodalem RAG - Pillar-Artikel
• Audio RAG: Podcasts und Transkriptionen
• RAG auf Bildern: Vision Models
• Diagramme und Schemata für RAG