RAG conversationnel pour agent vocal : chunking et retrieval

📍 Guide principal de la série : Agents vocaux IA en 2026 : architecture, modèles, latence.

Un agent vocal sans accès à la connaissance métier reste un assistant générique. Le RAG (Retrieval-Augmented Generation) corrige ce manque, mais ses réglages habituels — chunks de 1 000 tokens, top-k à 10, prose dense — sont inadaptés à la voix où l’on dispose de moins de 100 millisecondes pour récupérer le contexte et où la réponse doit s’oraliser. Ce tutoriel construit un pipeline RAG taillé pour le temps réel : chunks courts, embeddings rapides, retrieval sous les 100 ms, reranking ciblé et reformulation orale stricte.

Prérequis

• Python 3.10+, environnement virtuel
• Une clé API OpenAI active
• Un corpus métier (FAQ, documentation produit, base de connaissances) en Markdown ou texte brut
• PostgreSQL 16+ avec l’extension pgvector, ou Chroma 0.5+, ou Qdrant en local
• Notions de Python avancé, embeddings et bases vectorielles
• Temps estimé : 90 minutes

Étape 1 — Comprendre ce qui change pour le RAG vocal

Trois contraintes structurent la conception. La latence : le retrieval s’insère dans un budget end-to-end de 800 ms, donc tout dépassement de 150 ms côté base vectorielle ronge la fluidité. La forme : le LLM ne peut pas réciter une liste à puces ou une URL au téléphone, il doit reformuler en deux phrases naturelles. Le silence : on ne peut pas dire « je cherche dans la base » pendant trois secondes, il faut soit pré-loader le contexte, soit jouer un mot d’attente discret.

La logique d’ensemble est : moins, plus court, plus vite. Chaque dimension a un seuil au-delà duquel la conversation se dégrade. Les étapes suivantes appliquent ces seuils dans le code.

Étape 2 — Découper le corpus en chunks adaptés à la voix

Le chunking doit favoriser des unités sémantiques que le LLM peut citer en bloc dans une phrase orale : un paragraphe, une étape de procédure, une réponse de FAQ. Le découpage par caractères fixes ne marche jamais bien ; le découpage par titres de niveau 2/3 fonctionne presque toujours.

pip install langchain-text-splitters tiktoken

# chunk.py
from pathlib import Path
from langchain_text_splitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter

def chunk_corpus(folder: str):
    md_splitter = MarkdownHeaderTextSplitter(
        headers_to_split_on=[("#", "h1"), ("##", "h2"), ("###", "h3")],
    )
    char_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1200, chunk_overlap=120,   # ≈ 300 tokens
        separators=["\n\n", "\n", ". "],
    )
    chunks = []
    for path in Path(folder).rglob("*.md"):
        md = path.read_text(encoding="utf-8")
        for section in md_splitter.split_text(md):
            for piece in char_splitter.split_text(section.page_content):
                chunks.append({
                    "text": piece,
                    "source": str(path),
                    **section.metadata,
                })
    return chunks

Le double découpage (markdown puis caractères) garantit qu’un chunk reste sous 300 tokens tout en respectant les frontières naturelles du document. chunk_overlap=120 évite qu’une réponse de FAQ soit coupée en deux morceaux qui perdent leur sens. À la sortie, on doit obtenir entre 5 et 20 chunks par document de taille moyenne — un chiffre directement vérifiable avec len(chunk_corpus("docs/")).

Étape 3 — Calculer les embeddings et indexer

Le modèle text-embedding-3-small d’OpenAI offre le meilleur ratio coût/qualité en 2026 : 0,02 USD par million de tokens, dimensions 1 536, qualité multilingue solide. Pour la base, pgvector reste le choix par défaut quand on a déjà PostgreSQL en infra ; Chroma est plus simple à mettre en place pour un prototype.

pip install openai psycopg2-binary

# index.py
import os, openai, psycopg2
client = openai.OpenAI()

def embed(texts):
    out = client.embeddings.create(
        model="text-embedding-3-small", input=texts,
    )
    return [e.embedding for e in out.data]

con = psycopg2.connect(os.environ["DATABASE_URL"])
con.cursor().execute("""
    CREATE EXTENSION IF NOT EXISTS vector;
    CREATE TABLE IF NOT EXISTS kb (
        id BIGSERIAL PRIMARY KEY,
        text TEXT, source TEXT, h2 TEXT,
        embedding vector(1536)
    );
    CREATE INDEX IF NOT EXISTS kb_emb_idx
        ON kb USING ivfflat (embedding vector_cosine_ops) WITH (lists=100);
""")
con.commit()

chunks = chunk_corpus("docs/")
batch  = 96
for i in range(0, len(chunks), batch):
    rows  = chunks[i:i+batch]
    embs  = embed([r["text"] for r in rows])
    with con.cursor() as cur:
        for r, e in zip(rows, embs):
            cur.execute(
                "INSERT INTO kb(text, source, h2, embedding) "
                "VALUES (%s, %s, %s, %s)",
                (r["text"], r["source"], r.get("h2",""), e),
            )
    con.commit()

L’index ivfflat avec 100 listes est le réglage par défaut pour un corpus de quelques dizaines de milliers de chunks ; au-delà du million, basculer sur hnsw donne une latence inférieure mais consomme plus de mémoire. À ce stade, SELECT count(*) FROM kb doit refléter le nombre attendu de chunks et la première requête vectorielle doit revenir en moins de 50 ms sur une base correctement chauffée.

Étape 4 — Retrieval rapide (sous 100 ms)

La requête de retrieval doit être minimaliste : on embed la question reformulée, on cherche les K plus proches, on retourne texte + métadonnées. Le seul piège est l’embedding de la question lui-même, qui peut prendre 80-150 ms : on l’optimise en passant par un modèle hébergé proche (Cloudflare Workers AI, Together) si la latence devient un blocage.

# retrieve.py
import psycopg2, openai
client = openai.OpenAI()
con    = psycopg2.connect(os.environ["DATABASE_URL"])

def retrieve(query: str, k: int = 4):
    qemb = client.embeddings.create(
        model="text-embedding-3-small", input=[query],
    ).data[0].embedding
    with con.cursor() as cur:
        cur.execute(
            "SELECT text, source FROM kb "
            "ORDER BY embedding <=> %s::vector LIMIT %s",
            (qemb, k),
        )
        return [{"text": r[0], "source": r[1]} for r in cur.fetchall()]

L’opérateur de distance cosinus de pgvector est noté <=>. Avec k=4 on a quatre chunks candidats, suffisant pour la majorité des questions opérationnelles d’un agent vocal. La requête prend typiquement 30-60 ms sur une base de 100 000 chunks ; ajoutée à l’embedding (~80 ms), le retrieval total tient sous 150 ms — dans le budget vocal acceptable.

Étape 5 — Reranking pour ne garder que le pertinent

Le top-k vectoriel ramène souvent du contexte tangentiellement pertinent. Un reranker (modèle léger spécialisé) trie les K candidats par pertinence sémantique et permet de ne passer que 1 à 2 chunks au LLM, ce qui réduit le bruit et la latence de génération.

pip install sentence-transformers

from sentence_transformers import CrossEncoder
reranker = CrossEncoder("BAAI/bge-reranker-v2-m3", max_length=512)

def rerank(query, candidates, top=2):
    pairs  = [(query, c["text"]) for c in candidates]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(candidates, scores), key=lambda x: -x[1])
    return [c for c, s in ranked[:top]]

BAAI/bge-reranker-v2-m3 est le modèle de reranking multilingue le plus utilisé en 2026 : 600 millions de paramètres (0,6 milliard), latence ~30 ms par paire sur GPU, ~120 ms sur CPU optimisé. En passant K=4 puis top=2, on conserve les deux chunks les plus utiles et on coupe la moitié du bruit, ce qui se ressent immédiatement sur la qualité de la réponse orale et sur la latence du LLM.

Étape 6 — Brancher le retrieval comme outil dans LiveKit Agents

Plutôt que de faire le retrieval avant chaque tour, on l’expose comme un tool que le modèle décide d’appeler quand il en a besoin. C’est le pattern recommandé par LiveKit Agents 1.5 : le modèle reste en charge de juger quand la base interne est nécessaire.

from livekit.agents import function_tool, Agent

class SupportAgent(Agent):
    def __init__(self):
        super().__init__(instructions=(
            "Tu es Aïda, assistante du support. "
            "Quand la question concerne nos produits, appelle 'lookup_kb'. "
            "Réponds en deux phrases, sans listes, sans URL."
        ))

    @function_tool()
    async def lookup_kb(self, query: str) -> str:
        """Cherche dans la base de connaissances produit."""
        cands = retrieve(query, k=4)
        best  = rerank(query, cands, top=2)
        return "\n---\n".join(c["text"] for c in best)

Le décorateur @function_tool() expose la méthode comme outil disponible au LLM, sans avoir à écrire le schéma JSON manuellement. Le docstring devient la description envoyée au modèle, donc il faut le rédiger avec soin — c’est lui qui décide si l’outil sera appelé. À l’usage, le modèle déclenche typiquement le retrieval sur 30-50 % des tours, plutôt que sur chaque tour, ce qui fait économiser du contexte et accélère les questions sans rapport avec la base.

Étape 7 — Forcer la reformulation orale

Les chunks renvoyés par lookup_kb contiennent souvent du markdown, des liens ou des phrases construites pour la lecture. Le modèle doit les reformuler avant de les oraliser. Deux mécanismes se complètent : un prompt système qui interdit explicitement les éléments non-vocaux, et un post-processing optionnel qui nettoie la sortie avant le TTS.

instructions = (
    "Quand tu cites la base de connaissances, "
    "reformule en français parlé : pas de listes, pas de markdown, "
    "pas d'URL, pas de noms de fichiers. "
    "Maximum deux phrases, ton chaleureux, niveau de langue courant."
)

import re
def voice_clean(text: str) -> str:
    text = re.sub(r"https?://\S+", "le lien partagé en chat", text)
    text = re.sub(r"\*+|`+", "", text)
    text = re.sub(r"\[(.+?)\]\(.+?\)", r"\1", text)
    return text.strip()

Le filtre voice_clean est appliqué avant l’envoi au TTS et sert de filet de sécurité quand le LLM laisse passer un caractère markdown. Les URL sont remplacées par « le lien partagé en chat » : un agent vocal ne lit jamais d’URL au téléphone — il faut soit envoyer le lien par SMS (Twilio), soit le sortir dans le canal data WebRTC (LiveKit), soit l’afficher dans le widget de chat compagnon.

Étape 8 — Tester avec un jeu de questions

Sans batterie de test, le RAG dérive en silence. Une suite minimale de 30 à 50 questions étiquetées avec la bonne réponse attendue permet de mesurer la régression à chaque modification du corpus ou du modèle.

# eval_rag.py
import json, statistics
from retrieve import retrieve
from rerank   import rerank

dataset = json.load(open("eval_questions.json"))
hits = []
for q in dataset:
    cands = retrieve(q["question"], k=4)
    best  = rerank(q["question"], cands, top=2)
    found = any(q["expected_id"] in c.get("source","") for c in best)
    hits.append(int(found))
print(f"Recall@2 = {statistics.mean(hits)*100:.1f}%")

Un recall@2 au-dessus de 85 % est un objectif raisonnable pour un corpus métier propre. En dessous, les pistes habituelles sont : chunking trop grossier (résorber les chunks à 200 tokens), question reformulée trop différemment du corpus (envisager un HyDE), ou base mal indexée (rebuilder l’index ivfflat après import massif).

Erreurs fréquentes

Dans la continuité

• Documentation pgvector
• Modèle OpenAI Embeddings
• Reranker bge-reranker-v2-m3
• 🔝 Retour au guide principal : Agents vocaux IA en 2026 : architecture, modèles, latence
• Tutoriel précédent : Agent vocal sur numéro Twilio
• Tutoriel suivant : Silero VAD et turn-detector LiveKit