Audio RAG : Podcasts, calls et transcriptions

L'audio représente une mine d'or d'informations souvent inexploitée : réunions enregistrées, calls commerciaux, podcasts internes, formations. L'Audio RAG permet de rendre tout ce contenu sonore cherchable et exploitable par vos assistants IA.

Pourquoi l'Audio RAG ?

Le problème des données audio

• Volume massif : Une entreprise moyenne génère 50+ heures d'audio/semaine (meetings, calls)
• Information perdue : 80% du contenu des réunions n'est jamais documenté
• Recherche impossible : Impossible de "ctrl+F" dans un fichier audio
• Temps perdu : Réécouter pour retrouver une information = inefficace

Cas d'usage métier

ROI typique

• Réduction 70% du temps de recherche d'information
• +40% de rétention des connaissances partagées en réunion
• Compliance : Traçabilité des échanges verbaux

Architecture Audio RAG

┌─────────────────────────────────────────────────────────────┐
│                    AUDIO RAG PIPELINE                        │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  ┌──────────┐    ┌──────────────┐    ┌──────────────────┐  │
│  │  Audio   │───▶│  Whisper/    │───▶│  Transcription   │  │
│  │  Input   │    │  STT Model   │    │  + Timestamps    │  │
│  └──────────┘    └──────────────┘    └──────────────────┘  │
│                         │                     │             │
│                         ▼                     ▼             │
│  ┌──────────────────────────────────────────────────────┐  │
│  │              Diarisation (speaker ID)                 │  │
│  └──────────────────────────────────────────────────────┘  │
│                         │                                   │
│                         ▼                                   │
│  ┌──────────────────────────────────────────────────────┐  │
│  │     Segmentation sémantique (topics/chapters)         │  │
│  └──────────────────────────────────────────────────────┘  │
│                         │                                   │
│                         ▼                                   │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────┐  │
│  │  Embedding   │  │   Vector     │  │    Metadata      │  │
│  │  par segment │  │   Store      │  │  (speaker, time) │  │
│  └──────────────┘  └──────────────┘  └──────────────────┘  │
│                         │                                   │
│                         ▼                                   │
│  ┌──────────────────────────────────────────────────────┐  │
│  │          Retrieval + Génération avec source          │  │
│  └──────────────────────────────────────────────────────┘  │
│                                                              │
└─────────────────────────────────────────────────────────────┘

Transcription : Le fondement

Comparatif des modèles STT

Whisper : Le choix recommandé

Whisper d'OpenAI offre le meilleur rapport qualité/prix, surtout en auto-hébergement.

DEVELOPERpython
import whisper
from pathlib import Path

class AudioTranscriber:
    def __init__(self, model_size: str = "large-v3"):
        """
        Modèles disponibles: tiny, base, small, medium, large, large-v3
        VRAM requise: 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:
        """Transcrit un fichier audio avec timestamps."""
        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 minutes
    ) -> list[dict]:
        """
        Transcrit par chunks pour les longs audios.
        Évite les problèmes de mémoire et améliore la précision.
        """
        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]

            # Exporter temporairement
            temp_path = f"/tmp/chunk_{i}.wav"
            chunk.export(temp_path, format="wav")

            # Transcrire
            result = self.transcribe(temp_path)

            # Ajuster les timestamps
            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 (plus simple)

DEVELOPERpython
from openai import OpenAI

def transcribe_with_api(audio_path: str) -> dict:
    """Transcription via l'API OpenAI Whisper."""
    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
    }

Diarisation : Identifier les locuteurs

La diarisation répond à "Qui parle quand ?". Essentielle pour les réunions multi-participants.

Pyannote : Le standard open source

DEVELOPERpython
from pyannote.audio import Pipeline
import torch

class SpeakerDiarizer:
    def __init__(self, hf_token: str):
        """
        Nécessite un token HuggingFace avec accès au modèle 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]:
        """
        Identifie les locuteurs dans un audio.

        Args:
            audio_path: Chemin vers le fichier audio
            num_speakers: Nombre de locuteurs (optionnel, auto-détecté sinon)
        """
        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]:
        """Fusionne transcription et diarisation."""
        merged = []

        for trans_seg in transcription["segments"]:
            # Trouver le speaker qui parle le plus pendant ce segment
            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

            # Assigner le speaker majoritaire
            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

Segmentation sémantique

Découper la transcription en topics/chapitres cohérents pour un meilleur retrieval.

DEVELOPERpython
from openai import OpenAI

def segment_transcript_by_topics(
    transcript_segments: list[dict],
    client: OpenAI
) -> list[dict]:
    """
    Segmente une transcription en topics thématiques.
    """
    # Formater la transcription
    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"""Analyse cette transcription et identifie les différents sujets/topics abordés.

Pour chaque topic, indique:
1. Le titre du topic (court, descriptif)
2. Le timestamp de début (en secondes)
3. Le timestamp de fin (en secondes)
4. Un résumé en 1-2 phrases

Transcription:
{formatted}

Réponds en JSON avec le 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)

    # Enrichir avec le texte correspondant
    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

Indexation pour le RAG

Structure de données recommandée

DEVELOPERpython
from dataclasses import dataclass
from typing import Optional

@dataclass
class AudioChunk:
    """Représente un segment audio indexable."""
    chunk_id: str
    audio_source_id: str
    audio_source_title: str

    # Contenu
    text: str
    speaker: Optional[str]

    # Temporel
    start_time: float
    end_time: float

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

    # Métadonnées
    language: str
    confidence: float
    created_at: str

    def to_indexable_text(self) -> str:
        """Texte enrichi pour l'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)

Pipeline d'indexation complet

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):
        """Crée la collection Qdrant."""
        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]:
        """Pipeline complet de traitement audio."""

        # 1. Transcription
        print("Transcription en cours...")
        transcription = self.transcriber.transcribe(audio_path)

        # 2. Diarisation
        print("Diarisation en cours...")
        diarization = self.diarizer.diarize(audio_path, num_speakers)

        # 3. Fusion
        merged = self.diarizer.merge_transcription_diarization(
            transcription, diarization
        )

        # 4. Segmentation par topics
        print("Segmentation par topics...")
        topics = segment_transcript_by_topics(merged, self.openai)

        # 5. Créer les chunks
        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,  # Multi-speaker dans un topic
                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]):
        """Indexe les chunks dans Qdrant."""
        points = []

        for chunk in chunks:
            # Générer l'embedding
            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"Indexé {len(points)} chunks")

Retrieval et génération

Recherche avec contexte temporel

DEVELOPERpython
def search_audio_rag(
    query: str,
    indexer: AudioRAGIndexer,
    limit: int = 5
) -> list[dict]:
    """Recherche dans les transcriptions avec contexte."""
    # Embedding de la requête
    response = indexer.openai.embeddings.create(
        model="text-embedding-3-small",
        input=query
    )
    query_embedding = response.data[0].embedding

    # Recherche
    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
    ]

Génération avec citation de source audio

DEVELOPERpython
def generate_answer_with_audio_sources(
    query: str,
    retrieved_chunks: list[dict],
    client: OpenAI
) -> str:
    """Génère une réponse en citant les sources audio."""

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

    prompt = f"""Tu es un assistant qui répond aux questions en te basant sur des transcriptions audio.

Contexte (extraits de transcriptions):
{context}

Question: {query}

Instructions:
1. Réponds en te basant uniquement sur les transcriptions fournies
2. Cite tes sources avec le format [Source: titre, timestamp]
3. Si l'information n'est pas dans les transcriptions, dis-le clairement
4. Sois concis mais précis"""

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

    return response.choices[0].message.content

Optimisations avancées

Résumé automatique de réunions

DEVELOPERpython
def summarize_meeting(
    transcript_with_speakers: list[dict],
    client: OpenAI
) -> dict:
    """Génère un résumé structuré de réunion."""
    formatted = "\n".join([
        f"{seg['speaker']}: {seg['text']}"
        for seg in transcript_with_speakers
    ])

    prompt = f"""Analyse cette transcription de réunion et génère un résumé structuré.

Transcription:
{formatted}

Génère un JSON avec:
{{
  "title": "Titre suggéré pour cette réunion",
  "participants": ["Liste des participants identifiés"],
  "duration_minutes": X,
  "key_points": ["Point clé 1", "Point clé 2", ...],
  "decisions": ["Décision 1", "Décision 2", ...],
  "action_items": [
    {{"assignee": "Nom", "task": "Description", "deadline": "si mentionné"}}
  ],
  "next_steps": ["Prochaine étape 1", ...],
  "summary": "Résumé en 2-3 paragraphes"
}}"""

    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)

Détection de moments clés

DEVELOPERpython
def detect_key_moments(
    transcript_segments: list[dict],
    client: OpenAI
) -> list[dict]:
    """Identifie les moments importants dans un audio."""
    formatted = "\n".join([
        f"[{seg['start']:.0f}s] {seg.get('speaker', 'Speaker')}: {seg['text']}"
        for seg in transcript_segments
    ])

    prompt = f"""Identifie les moments clés de cette transcription:
- Questions importantes posées
- Décisions prises
- Désaccords ou débats
- Informations critiques partagées
- Moments d'humour ou tension

Transcription:
{formatted}

Pour chaque moment clé, donne:
- timestamp (en secondes)
- type (question/decision/debate/info/other)
- description courte
- importance (1-5)

Réponds en JSON: [{{"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)

Coûts et performance

Coûts de transcription

Coûts de diarisation

• Pyannote local : $0 (mais GPU requis)
• AssemblyAI : Inclus
• AWS Transcribe : +$0.024/min

Stockage estimé

• 1 heure audio = ~15,000 mots = ~100 chunks
• Embeddings : ~600KB/heure
• Metadata : ~50KB/heure

Intégration avec Ailog

Ailog simplifie l'Audio RAG avec une intégration native :

1. Upload audio : Formats supportés : MP3, WAV, M4A, WEBM
2. Transcription automatique : Whisper intégré
3. Indexation intelligente : Segmentation par topics
4. Recherche unifiée : Audio + texte + images dans une seule requête

Essayez l'Audio RAG sur Ailog

Guides connexes

• Guide complet du RAG multimodal - Article pilier
• RAG sur images : Vision models
• Vidéo RAG : Indexer vos vidéos
• Stratégies de chunking