Audio RAG : Podcasts, Anrufe und Transkriptionen

Audio stellt eine Goldgrube an Informationen dar, die oft ungenutzt bleibt : aufgezeichnete Meetings, Sales-Calls, interne Podcasts, Schulungen. Audio RAG macht all diese Audioinhalte durchsuchbar und für Ihre KI-Assistenten nutzbar.

Warum Audio RAG ?

Das Problem mit Audiodaten

• Massives Volumen : Ein durchschnittliches Unternehmen erzeugt 50+ Stunden Audio pro Woche (meetings, calls)
• Verlorene Informationen : 80% des Inhalts von Meetings wird nie dokumentiert
• Unmöglich zu durchsuchen : In einer Audiodatei ist kein „Strg+F" möglich
• Zeitverschwendung : Zum Finden einer Information alles erneut anhören = ineffizient

Anwendungsfälle nach Branche

Typischer ROI

• 70% Reduktion der Zeit zur Informationssuche
• +40% höhere Beibehaltung des in Meetings geteilten Wissens
• Compliance : Nachverfolgbarkeit verbaler Austausche

Architektur Audio RAG

┌─────────────────────────────────────────────────────────────┐
│                    AUDIO RAG PIPELINE                        │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  ┌──────────┐    ┌──────────────┐    ┌──────────────────┐  │
│  │  Audio   │───▶│  Whisper/    │───▶│  Transcription   │  │
│  │  Input   │    │  STT Model   │    │  + Timestamps    │  │
│  └──────────┘    └──────────────┘    └──────────────────┘  │
│                         │                     │             │
│                         ▼                     ▼             │
│  ┌──────────────────────────────────────────────────────┐  │
│  │              Diarisation (speaker ID)                 │  │
│  └──────────────────────────────────────────────────────┘  │
│                         │                                   │
│                         ▼                                   │
│  ┌──────────────────────────────────────────────────────┐  │
│  │     Semantische Segmentierung (topics/chapters)       │  │
│  └──────────────────────────────────────────────────────┘  │
│                         │                                   │
│                         ▼                                   │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────┐  │
│  │  Embedding   │  │   Vector     │  │    Metadata      │  │
│  │  par segment │  │   Store      │  │  (speaker, time) │  │
│  └──────────────┘  └──────────────┘  └──────────────────┘  │
│                         │                                   │
│                         ▼                                   │
│  ┌──────────────────────────────────────────────────────┐  │
│  │          Retrieval + Generierung mit Quelle           │  │
│  └──────────────────────────────────────────────────────┘  │
│                                                              │
└─────────────────────────────────────────────────────────────┘

Transkription : Die Grundlage

Vergleich der STT-Modelle

Whisper : Die empfohlene Wahl

Whisper von OpenAI bietet das beste Preis-Leistungs-Verhältnis, besonders beim Selbsthosting.

DEVELOPERpython
import whisper
from pathlib import Path

class AudioTranscriber:
    def __init__(self, model_size: str = "large-v3"):
        """
        Verfügbare Modelle: tiny, base, small, medium, large, large-v3
        Erforderlicher VRAM: tiny=1GB, base=1GB, small=2GB, medium=5GB, large=10GB
        """
        self.model = whisper.load_model(model_size)

    def transcribe(
        self,
        audio_path: str,
        language: str = None,
        word_timestamps: bool = True
    ) -> dict:
        """Transkribiert eine Audiodatei mit Zeitstempeln."""
        result = self.model.transcribe(
            audio_path,
            language=language,
            word_timestamps=word_timestamps,
            verbose=False
        )

        return {
            "text": result["text"],
            "segments": result["segments"],
            "language": result["language"],
            "duration": result["segments"][-1]["end"] if result["segments"] else 0
        }

    def transcribe_with_chunks(
        self,
        audio_path: str,
        chunk_duration: int = 300  # 5 Minuten
    ) -> list[dict]:
        """
        Transkribiert in Chunks für lange Audiodateien.
        Vermeidet Speicherprobleme und verbessert die Genauigkeit.
        """
        from pydub import AudioSegment

        audio = AudioSegment.from_file(audio_path)
        duration_ms = len(audio)
        chunk_ms = chunk_duration * 1000

        chunks = []
        for i, start in enumerate(range(0, duration_ms, chunk_ms)):
            end = min(start + chunk_ms, duration_ms)
            chunk = audio[start:end]

            # Temporär exportieren
            temp_path = f"/tmp/chunk_{i}.wav"
            chunk.export(temp_path, format="wav")

            # Transkribieren
            result = self.transcribe(temp_path)

            # Zeitstempel anpassen
            for seg in result["segments"]:
                seg["start"] += start / 1000
                seg["end"] += start / 1000

            chunks.append({
                "chunk_index": i,
                "start_time": start / 1000,
                "end_time": end / 1000,
                **result
            })

        return chunks

Whisper API (einfacher)

DEVELOPERpython
from openai import OpenAI

def transcribe_with_api(audio_path: str) -> dict:
    """Transkription über die OpenAI Whisper API."""
    client = OpenAI()

    with open(audio_path, "rb") as audio_file:
        transcript = client.audio.transcriptions.create(
            model="whisper-1",
            file=audio_file,
            response_format="verbose_json",
            timestamp_granularities=["word", "segment"]
        )

    return {
        "text": transcript.text,
        "segments": transcript.segments,
        "words": transcript.words,
        "language": transcript.language,
        "duration": transcript.duration
    }

Diarisierung : Sprecher identifizieren

Diarisierung beantwortet die Frage "Wer spricht wann?". Unverzichtbar für Meetings mit mehreren Teilnehmern.

Pyannote : Der Open-Source-Standard

DEVELOPERpython
from pyannote.audio import Pipeline
import torch

class SpeakerDiarizer:
    def __init__(self, hf_token: str):
        """
        Erfordert ein HuggingFace-Token mit Zugriff auf das Modell pyannote/speaker-diarization-3.1
        """
        self.pipeline = Pipeline.from_pretrained(
            "pyannote/speaker-diarization-3.1",
            use_auth_token=hf_token
        )

        if torch.cuda.is_available():
            self.pipeline.to(torch.device("cuda"))

    def diarize(self, audio_path: str, num_speakers: int = None) -> list[dict]:
        """
        Identifiziert die Sprecher in einer Audiodatei.

        Args:
            audio_path: Pfad zur Audiodatei
            num_speakers: Anzahl der Sprecher (optional, sonst automatisch erkannt)
        """
        diarization = self.pipeline(
            audio_path,
            num_speakers=num_speakers
        )

        segments = []
        for turn, _, speaker in diarization.itertracks(yield_label=True):
            segments.append({
                "speaker": speaker,
                "start": turn.start,
                "end": turn.end,
                "duration": turn.end - turn.start
            })

        return segments

    def merge_transcription_diarization(
        self,
        transcription: dict,
        diarization: list[dict]
    ) -> list[dict]:
        """Führt Transkription und Diarisierung zusammen."""
        merged = []

        for trans_seg in transcription["segments"]:
            # Den Speaker finden, der während dieses Segments am meisten spricht
            seg_start = trans_seg["start"]
            seg_end = trans_seg["end"]

            speaker_times = {}
            for diar_seg in diarization:
                overlap_start = max(seg_start, diar_seg["start"])
                overlap_end = min(seg_end, diar_seg["end"])

                if overlap_start < overlap_end:
                    overlap = overlap_end - overlap_start
                    speaker = diar_seg["speaker"]
                    speaker_times[speaker] = speaker_times.get(speaker, 0) + overlap

            # Den mehrheitlichen Speaker zuweisen
            speaker = max(speaker_times, key=speaker_times.get) if speaker_times else "UNKNOWN"

            merged.append({
                "speaker": speaker,
                "start": seg_start,
                "end": seg_end,
                "text": trans_seg["text"]
            })

        return merged

Semantische Segmentierung

Transkription in kohärente Topics/Kapitel aufteilen für besseres retrieval.

DEVELOPERpython
from openai import OpenAI

def segment_transcript_by_topics(
    transcript_segments: list[dict],
    client: OpenAI
) -> list[dict]:
    """
    Segmentiert eine Transkription in thematische Topics.
    """
    # Transkription formatieren
    formatted = "\n".join([
        f"[{seg['start']:.1f}s - {seg['end']:.1f}s] {seg.get('speaker', 'Speaker')}: {seg['text']}"
        for seg in transcript_segments
    ])

    prompt = f"""Analysiere diese Transkription und identifiziere die verschiedenen behandelten Themen/Topics.

Gib für jedes Topic an:
1. Den Titel des Topics (kurz, beschreibend)
2. Den Start-Zeitstempel (in Sekunden)
3. Den End-Zeitstempel (in Sekunden)
4. Eine Zusammenfassung in 1-2 Sätzen

Transkription:
{formatted}

Antworte im JSON-Format:
[
  {{"title": "...", "start": 0.0, "end": 120.0, "summary": "..."}},
  ...
]"""

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

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

    # Mit dem entsprechenden Text anreichern
    for topic in topics:
        topic_text = []
        for seg in transcript_segments:
            if seg["start"] >= topic["start"] and seg["end"] <= topic["end"]:
                topic_text.append(seg["text"])
        topic["full_text"] = " ".join(topic_text)

    return topics

Indexierung für RAG

Empfohlene Datenstruktur

DEVELOPERpython
from dataclasses import dataclass
from typing import Optional

@dataclass
class AudioChunk:
    """Repräsentiert ein indexierbares Audiosegment."""
    chunk_id: str
    audio_source_id: str
    audio_source_title: str

    # Inhalt
    text: str
    speaker: Optional[str]

    # Zeitlich
    start_time: float
    end_time: float

    # Kontext
    topic: Optional[str]
    topic_summary: Optional[str]

    # Metadaten
    language: str
    confidence: float
    created_at: str

    def to_indexable_text(self) -> str:
        """Angereicherter Text für das Embedding."""
        parts = []
        if self.topic:
            parts.append(f"Topic: {self.topic}")
        if self.speaker:
            parts.append(f"Speaker: {self.speaker}")
        parts.append(self.text)
        return "\n".join(parts)

Vollständige Indexierungspipeline

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

class AudioRAGIndexer:
    def __init__(self):
        self.qdrant = QdrantClient(url="http://localhost:6333")
        self.openai = OpenAI()
        self.transcriber = AudioTranscriber()
        self.diarizer = SpeakerDiarizer(hf_token="...")
        self.collection_name = "audio_rag"

    def create_collection(self):
        """Erstellt die Qdrant-Collection."""
        self.qdrant.recreate_collection(
            collection_name=self.collection_name,
            vectors_config=VectorParams(
                size=1536,  # text-embedding-3-small
                distance=Distance.COSINE
            )
        )

    def process_audio(
        self,
        audio_path: str,
        title: str,
        num_speakers: int = None
    ) -> list[AudioChunk]:
        """Vollständige Audio-Verarbeitungspipeline."""

        # 1. Transkription
        print("Transkription läuft...")
        transcription = self.transcriber.transcribe(audio_path)

        # 2. Diarisierung
        print("Diarisierung läuft...")
        diarization = self.diarizer.diarize(audio_path, num_speakers)

        # 3. Zusammenführung
        merged = self.diarizer.merge_transcription_diarization(
            transcription, diarization
        )

        # 4. Segmentierung nach Themen
        print("Segmentierung nach Topics...")
        topics = segment_transcript_by_topics(merged, self.openai)

        # 5. Chunks erstellen
        chunks = []
        source_id = hashlib.md5(audio_path.encode()).hexdigest()

        for topic in topics:
            chunk = AudioChunk(
                chunk_id=f"{source_id}_{topic['start']}",
                audio_source_id=source_id,
                audio_source_title=title,
                text=topic["full_text"],
                speaker=None,  # Mehrere Speaker innerhalb eines Topics
                start_time=topic["start"],
                end_time=topic["end"],
                topic=topic["title"],
                topic_summary=topic["summary"],
                language=transcription["language"],
                confidence=0.95,
                created_at=datetime.now().isoformat()
            )
            chunks.append(chunk)

        return chunks

    def index_chunks(self, chunks: list[AudioChunk]):
        """Indexiert die Chunks in Qdrant."""
        points = []

        for chunk in chunks:
            # Embedding generieren
            text = chunk.to_indexable_text()
            response = self.openai.embeddings.create(
                model="text-embedding-3-small",
                input=text
            )
            embedding = response.data[0].embedding

            point = PointStruct(
                id=hash(chunk.chunk_id) % (2**63),
                vector=embedding,
                payload={
                    "chunk_id": chunk.chunk_id,
                    "audio_source_id": chunk.audio_source_id,
                    "audio_source_title": chunk.audio_source_title,
                    "text": chunk.text,
                    "speaker": chunk.speaker,
                    "start_time": chunk.start_time,
                    "end_time": chunk.end_time,
                    "topic": chunk.topic,
                    "topic_summary": chunk.topic_summary,
                    "language": chunk.language
                }
            )
            points.append(point)

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

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

Retrieval und Generierung

Suche mit zeitlichem Kontext

DEVELOPERpython
def search_audio_rag(
    query: str,
    indexer: AudioRAGIndexer,
    limit: int = 5
) -> list[dict]:
    """Sucht in den Transkriptionen mit Kontext."""
    # Embedding der Anfrage
    response = indexer.openai.embeddings.create(
        model="text-embedding-3-small",
        input=query
    )
    query_embedding = response.data[0].embedding

    # Suche
    results = indexer.qdrant.search(
        collection_name=indexer.collection_name,
        query_vector=query_embedding,
        limit=limit
    )

    return [
        {
            "text": r.payload["text"],
            "source": r.payload["audio_source_title"],
            "topic": r.payload["topic"],
            "timestamp": f"{r.payload['start_time']:.0f}s - {r.payload['end_time']:.0f}s",
            "score": r.score
        }
        for r in results
    ]

Generierung mit Zitierung von Audioquellen

DEVELOPERpython
def generate_answer_with_audio_sources(
    query: str,
    retrieved_chunks: list[dict],
    client: OpenAI
) -> str:
    """Generiert eine Antwort unter Angabe der Audioquellen."""

    context = "\n\n".join([
        f"**Quelle: {c['source']}** (Topic: {c['topic']}, {c['timestamp']})\n{c['text']}"
        for c in retrieved_chunks
    ])

    prompt = f"""Du bist ein Assistent, der Fragen auf Basis von Audiotranskriptionen beantwortet.

Kontext (Auszüge aus Transkriptionen):
{context}

Frage: {query}

Anweisungen:
1. Antworte ausschließlich auf Basis der bereitgestellten Transkriptionen
2. Zitiere deine Quellen im Format [Quelle: Titel, Zeitstempel]
3. Wenn die Information nicht in den Transkriptionen enthalten ist, sag das klar
4. Sei präzise, aber prägnant"""

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

    return response.choices[0].message.content

Erweiterte Optimierungen

Automatische Meeting-Zusammenfassung

DEVELOPERpython
def summarize_meeting(
    transcript_with_speakers: list[dict],
    client: OpenAI
) -> dict:
    """Generiert eine strukturierte Meeting-Zusammenfassung."""
    formatted = "\n".join([
        f"{seg['speaker']}: {seg['text']}"
        for seg in transcript_with_speakers
    ])

    prompt = f"""Analysiere diese Meeting-Transkription und erstelle eine strukturierte Zusammenfassung.

Transkription:
{formatted}

Erstelle ein JSON mit:
{{
  "title": "Vorgeschlagener Titel für dieses Meeting",
  "participants": ["Liste der identifizierten Teilnehmer"],
  "duration_minutes": X,
  "key_points": ["Kernpunkt 1", "Kernpunkt 2", ...],
  "decisions": ["Entscheidung 1", "Entscheidung 2", ...],
  "action_items": [
    {{"assignee": "Name", "task": "Beschreibung", "deadline": "falls erwähnt"}}
  ],
  "next_steps": ["Nächster Schritt 1", ...],
  "summary": "Zusammenfassung in 2-3 Absätzen"
}}"""

    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)

Erkennung wichtiger Momente

DEVELOPERpython
def detect_key_moments(
    transcript_segments: list[dict],
    client: OpenAI
) -> list[dict]:
    """Identifiziert wichtige Momente in einer Audiodatei."""
    formatted = "\n".join([
        f"[{seg['start']:.0f}s] {seg.get('speaker', 'Speaker')}: {seg['text']}"
        for seg in transcript_segments
    ])

    prompt = f"""Identifiziere die Schlüsselmomente dieser Transkription:
- Wichtige gestellte Fragen
- Getroffene Entscheidungen
- Meinungsverschiedenheiten oder Debatten
- Geteilte kritische Informationen
- Momente von Humor oder Anspannung

Transkription:
{formatted}

Gib für jeden Schlüsselmoment an:
- timestamp (in Sekunden)
- type (question/decision/debate/info/other)
- kurze Beschreibung
- importance (1-5)

Antworte im JSON-Format: [{{"timestamp": X, "type": "...", "description": "...", "importance": X}}, ...]"""

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

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

Kosten und Leistung

Transkriptionskosten

Diarisierungskosten

• Pyannote lokal : $0 (aber GPU erforderlich)
• AssemblyAI : Inklusive
• AWS Transcribe : +$0.024/min

Geschätzter Speicherbedarf

• 1 Stunde Audio = ~15.000 Wörter = ~100 Chunks
• Embeddings : ~600KB/Stunde
• Metadaten : ~50KB/Stunde

Integration mit Ailog

Ailog vereinfacht das Audio RAG mit einer nativen Integration :

1. Audio hochladen : Unterstützte Formate : MP3, WAV, M4A, WEBM
2. Automatische Transkription : Whisper integriert
3. Intelligente Indexierung : Segmentierung nach Topics
4. Einheitliche Suche : Audio + Text + Bilder in einer einzigen Anfrage

Probieren Sie Audio RAG auf Ailog aus

Verwandte Guides

• Vollständiger Leitfaden zu multimodalem RAG - Grundlagenartikel
• RAG für Bilder: Vision-Modelle
• Video-RAG: Ihre Videos indexieren
• Chunking-Strategien