Multimodal RAG — vision, tables, audio dans tes pipelines

TL;DR 80 % de la valeur métier dans l'industrie, le BTP, le retail et la pharma vit dans des images, schémas, tables et notices — pas dans du texte propre. Un RAG text-only est aveugle. En 2026, les embeddings multimodaux (CLIP, Voyage Multimodal, Cohere Embed v4, ColPali) permettent de chercher dans des PDF mixtes, des plans, des photos, des audios. Pour un architecte qui cherche "détail d'attache poutre métallique sur dalle béton" dans 5 000 dossiers DCE, le multimodal RAG est la seule réponse possible. Mission spécialisée packagée : 22-45 k€ selon volume + retainer. C'est la mission différenciante pour pivot sénior.

🧠 Mental model

Multimodal RAG = même pipeline RAG, mais avec un espace d'embedding partagé entre texte et image (et audio, et tables).

   ┌───────────┐    ┌──────────────┐
   │  TEXT     │───▶│              │
   ├───────────┤    │   SHARED     │      shared embedding
   │  IMAGE    │───▶│   EMBED      │───▶  space (Voyage/Cohere/CLIP)
   ├───────────┤    │   MODEL      │
   │  TABLE    │───▶│              │
   └───────────┘    └──────────────┘
                           │
                           ▼
                     vector DB (pgvector, Qdrant)
                           │
   query (text/img) ───▶ cosine ───▶ top-k mixed modalities
                           │
                           ▼
                vision LLM (Claude Sonnet vision, GPT-4o)
                           │
                           ▼
                  answer with image refs

Analogie : tu vas chez un kiné avec une douleur. Le kiné regarde comment tu marches, te palpe, lit ton dossier. C'est multi-modal. Si tu lui demandes de diagnostiquer juste avec un texte "j'ai mal à l'épaule", il te dit "viens, je veux te voir bouger". C'est ton RAG : sans la modalité visuelle, tu rates 80 % du signal.

Pour ton cerveau de dev : embeddings multimodaux = modèle entraîné en contrastive learning sur paires (texte, image). L'embedding d'une photo de chaise et l'embedding du texte "chaise en bois" sont proches dans l'espace. Tu fais ton cosine indifféremment entre modalités.

Comment un architecte raisonne sur les 3 stratégies

Il n'y a pas "un" multimodal RAG. Il y a trois familles, et le choix structure tout le coût et la latence du système. Un sénior choisit AVANT de coder, pas après :

Stratégie	Comment ça marche	Force	Faiblesse	Quand
Embed multimodal partagé (Cohere v4, Voyage MM)	1 vecteur par image/texte dans un espace commun	Simple, 1 index, retrieval cross-modal natif	Aplatit le détail fin d'une table dense	Photos produit, plans, recherche par image
ColPali / late-interaction	N vecteurs par page (1 par patch d'image)	Imbattable sur tables/PDF complexes, pas d'OCR	Stockage 10-100× plus lourd, retrieval plus lent	RCP médicaux, rapports financiers, dense layout
Vision LLM as OCR → embed texte	On transcrit l'image en texte, puis embed texte classique	Réutilise ton pipeline RAG texte existant	Coût d'ingestion élevé (1 appel LLM/page), perd la mise en page	Documents où le texte porte 90 % du signal

Le piège du junior : prendre UNE stratégie pour tout le corpus. Le sénior route par type de page (classify_page → plan/coupe = embed image ; cctp = embed texte ; table = ColPali). C'est exactement ce que fait l'exemple end-to-end plus bas. Le routage divise le coût d'ingestion par 3-5× et améliore le recall, parce que chaque modalité va dans l'index qui la sert le mieux.

Pourquoi le coût explose si tu n'y penses pas : un vision LLM compte ~1 568 tokens par image en haute résolution (jusqu'à ~4 784 sur Opus 4.8 en pleine résolution). Indexer 200 000 pages via vision LLM = 200 000 appels. À ~0,015 €/page, c'est 3 000 € rien que pour l'OCR. Le routage (vision uniquement sur 30-50 % des pages) et le cache d'embedding par hash (cf. Pitfall 11) sont les deux leviers qui rendent la mission rentable.

🛠️ Code minimal

Embed et search d'images + texte dans le même espace avec Cohere Embed v4 :

import cohere
import base64

co = cohere.AsyncClient()

async def embed_text(texts: list[str]) -> list[list[float]]:
    r = await co.embed(model="embed-v4.0", input_type="search_document", texts=texts)
    return r.embeddings

async def embed_image(path: str) -> list[float]:
    with open(path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode()
    r = await co.embed(
        model="embed-v4.0", input_type="image",
        images=[f"data:image/jpeg;base64,{b64}"],
    )
    return r.embeddings[0]

# Search texte → matche aussi des images dans le même espace
async def search(query: str) -> list[dict]:
    q_emb = (await embed_text([query]))[0]
    return await pgvector_search(q_emb, k=10)

C'est tout : la magie est dans le modèle, pas le code.

🎬 Cas d'usage concrets

Scénario 1 — Cabinet d'architectes parisien, dossiers DCE

Qui : cabinet 60 architectes, 18 ans d'archives, ~5 000 dossiers DCE (Dossier de Consultation des Entreprises) — chaque DCE = 200-800 pages mixant plans, coupes, notices techniques, CCTP texte.

Problème : un architecte cherche "détail d'attache poutre métallique IPN sur dalle béton plancher mixte" pour un projet courant. Cette info est sur un plan détail (image vectorielle exportée PDF) dans un DCE de 2019. Pas de texte indexable. Vector RAG text-only : ne trouve rien. L'archi passe 2h à fouiller le SharePoint.

Solution : pipeline OCR/vision pour extraire le texte des plans + embedding multimodal (Voyage Multimodal) sur les images des plans + metadata {projet, type_doc, échelle, étage}.

async def index_dce(pdf_path: str):
    images = pdf_to_images(pdf_path, dpi=300)  # par page
    for i, img in enumerate(images):
        # double indexation
        text = await vision_ocr(img)  # vision LLM → texte structuré
        text_emb = (await embed_text([text]))[0]  # parenthèse obligatoire : await lie embed_text(...), pas [0]
        img_emb = await embed_image_bytes(img)
        await store({
            "dce_id": ..., "page": i, "text": text,
            "text_emb": text_emb, "img_emb": img_emb,
            "type": classify_page(img),  # plan | coupe | cctp | notice
        })

Gains : recherche d'un détail technique passe de 2h à 90 secondes. 60 architectes × 5h économisées/mois × 75 €/h chargé = 22,5 k€/mois récupérés. Le cabinet a aussi capitalisé sur 18 ans d'archives qu'il ne valorisait pas. Mission : 22 jours × 1 450 €/j = 32 k€ + retainer 4 k€/mois.

Scénario 2 — Marketplace mode, recherche par image

Qui : marketplace mode (vintage + créateurs), 480 K produits, 30 % de la conversion vient de la recherche.

Problème : un utilisateur voit une chemise sur Instagram, veut "la même". Il upload la photo. Le moteur classique ne trouve rien. Concurrent : ASOS et Vestiaire Collective ont la feature, eux pas.

Solution : embeddings multimodaux Cohere Embed v4 sur les photos de fiches produits + endpoint upload image → embed → vector search.

@app.post("/visual-search")
async def visual_search(file: UploadFile):
    data = await file.read()
    emb = await embed_image_bytes(data)
    hits = await vector_search(emb, k=24)
    return {"products": hits}

Gains : nouvelle source de trafic conversion (8 % des sessions utilisent le visual search). Conversion sur visual search : 3,8 % (vs 1,9 % sur search texte). Sur CA 22 M€/an : +540 k€/an de revenus incrémentaux. Mission : 12 j × 1 350 €/j = 16,2 k€.

Scénario 2 bis — Immobilier (annonces avec photos)

Qui : startup PropTech, 280 K annonces avec 4-12 photos chacune, search texte uniquement.

Problème : un acheteur cherche "appartement style haussmannien moulures parquet" — la description du bien ne mentionne pas tous ces termes, mais les photos les montrent.

Solution : double indexation : embedding texte sur la description, embedding multimodal sur chaque photo. Au search, on cherche dans les deux index et on fusionne par bien (RRF agrégé par bien_id).

async def index_listing(listing):
    text_emb = (await embed_text([listing["description"]]))[0]
    photo_embs = await asyncio.gather(*[embed_image_url(p) for p in listing["photos"]])
    await store_text(listing["id"], text_emb)
    for i, e in enumerate(photo_embs):
        await store_photo(listing["id"], i, e)

async def hybrid_search(query: str):
    q_emb = (await embed_text([query]))[0]
    text_hits = await search_text(q_emb)
    photo_hits = await search_photos(q_emb)
    return rrf_by_listing([text_hits, photo_hits])

Gains : recall@10 sur queries esthétiques : 0.31 → 0.74. Engagement bondit (+28 % de mises en favori). Mission : 14 j × 1 400 €/j = 19,6 k€.

Scénario 3 — Industrie pharmaceutique, notices médicaments + extraction tables

Qui : laboratoire pharma mid-cap (300 personnes), 1 800 produits, notices RCP (Résumé des Caractéristiques du Produit) sur plusieurs langues, interactions médicamenteuses sous forme de tables.

Problème : un médecin pose une question naturelle ("Doliprane + paracétamol + grossesse"). La réponse vit dans une table d'interactions au milieu de 30 pages de RCP. Vector RAG text qui aplatit la table : illisible. L'agent fournit une réponse incorrecte.

Solution : pipeline ColPali pour les pages de tables (préserve la structure visuelle) + LLM vision (Claude Sonnet vision) au moment de la génération pour lire la table en image.

# Détection des pages-table
async def is_table_page(img: bytes) -> bool:
    # quick CV heuristic ou classifier
    return await classifier.predict(img) > 0.8

# Indexation différenciée
async def index_rcp_page(pdf_page):
    if await is_table_page(pdf_page.image):
        emb = await colpali_embed(pdf_page.image)
        await store_as_image_page(pdf_page, emb)
    else:
        text = pdf_page.extract_text()
        emb = (await embed_text([text]))[0]  # idem : (await ...)[0]
        await store_as_text_page(pdf_page, text, emb)

# Au moment de générer, on passe l'image au LLM vision
async def generate(query, hits):
    contents = []
    for h in hits:
        if h["type"] == "image":
            contents.append({"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": h["b64"]}})
        else:
            contents.append({"type": "text", "text": h["text"]})
    contents.append({"type": "text", "text": f"Question : {query}"})
    # Lecture de table d'interactions médicamenteuses = enjeu de responsabilité.
    # On laisse Opus 4.8 raisonner (adaptive thinking) et on monte l'effort.
    return await CLAUDE.messages.create(
        model=CHAT, max_tokens=1500,
        thinking={"type": "adaptive"},
        output_config={"effort": "high"},
        messages=[{"role": "user", "content": contents}])

Gains : précision sur questions interactions médicaments : 67 % → 94 %. Crucial pour la responsabilité métier. Mission packagée : 20 j × 1 500 €/j = 30 k€ + retainer 5 k€/mois.

🛠️ Exemple end-to-end

Use case : moteur de recherche pour cabinet d'architecture qui retrieve plans + spec textes depuis 5 000 dossiers DCE archivés. Tu factures 36 k€ (24 j × 1 500 €/j).

# multimodal_dce.py
import os
import asyncio
import base64
import hashlib
import json
import logging
from pathlib import Path
from dataclasses import dataclass
from typing import Any

import anthropic
import asyncpg
import cohere
import fitz  # PyMuPDF
from PIL import Image
import io

PG_DSN = os.environ["PG_DSN"]
CLAUDE = anthropic.AsyncAnthropic()
CO = cohere.AsyncClient()

CHAT = "claude-opus-4-8"          # flagship vision LLM pour la génération (5 $/25 $ /M tok, 1M ctx)
CHEAP = "claude-haiku-4-5"        # classification de page (1 $/5 $ /M tok)
EMBED_MODEL = "embed-v4.0"        # Cohere multimodal

# ------------ Schema SQL ------------
"""
CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE dce_pages (
    id          BIGSERIAL PRIMARY KEY,
    dce_id      TEXT NOT NULL,
    project     TEXT NOT NULL,
    page_num    INT  NOT NULL,
    page_type   TEXT NOT NULL,   -- plan|coupe|cctp|notice|cover
    text        TEXT,
    image_path  TEXT,
    embedding   VECTOR(1536) NOT NULL,
    created_at  TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX ON dce_pages USING hnsw (embedding vector_cosine_ops);
CREATE INDEX ON dce_pages (dce_id);
CREATE INDEX ON dce_pages (project);
"""

# ------------ Classification page ------------

CLASSIFY_PROMPT = """Tu classes une page de DCE architecture. Réponds avec UN seul label :
plan | coupe | cctp | notice | cover | other"""

async def classify_page(img_bytes: bytes) -> str:
    b64 = base64.standard_b64encode(img_bytes).decode()
    msg = await CLAUDE.messages.create(
        model=CHEAP, max_tokens=20,
        # Classification = tâche simple → Haiku, pas de thinking. On force le label
        # via output_config.format (structured output) plutôt qu'un prompt fragile.
        # Variante plus propre : CLAUDE.messages.parse(..., output_config={"format": ...})
        # valide la réponse contre un schéma Pydantic et te rend l'objet typé directement
        # (pas de json.loads à la main). On reste sur create() ici pour montrer le raw.
        output_config={"format": {"type": "json_schema", "schema": {
            "type": "object",
            "properties": {"label": {"type": "string",
                "enum": ["plan", "coupe", "cctp", "notice", "cover", "other"]}},
            "required": ["label"], "additionalProperties": False,
        }}},
        messages=[{
            "role": "user",
            "content": [
                {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": b64}},
                {"type": "text", "text": CLASSIFY_PROMPT},
            ],
        }],
    )
    return json.loads(msg.content[0].text)["label"]

# ------------ OCR / vision text extraction ------------

OCR_PROMPT = """Tu extrais le texte exploitable de cette page de DCE.
Pour un plan/coupe : retranscris les annotations, cotes, légendes, labels.
Pour du CCTP : extrait le texte brut.
Garde les références techniques (IPN 200, BA-30, etc.) et les niveaux (R+1, RDC, etc.).
Format : texte brut, pas de markdown."""

async def vision_extract(img_bytes: bytes) -> str:
    b64 = base64.standard_b64encode(img_bytes).decode()
    msg = await CLAUDE.messages.create(
        model=CHAT, max_tokens=1500,
        messages=[{
            "role": "user",
            "content": [
                {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": b64}},
                {"type": "text", "text": OCR_PROMPT},
            ],
        }],
    )
    return msg.content[0].text

# ------------ Embedding multimodal Cohere ------------

async def embed_text(texts: list[str]) -> list[list[float]]:
    r = await CO.embed(model=EMBED_MODEL, input_type="search_document", texts=texts)
    return r.embeddings

async def embed_image(img_bytes: bytes) -> list[float]:
    b64 = base64.b64encode(img_bytes).decode()
    r = await CO.embed(
        model=EMBED_MODEL, input_type="image",
        images=[f"data:image/png;base64,{b64}"],
    )
    return r.embeddings[0]

# ------------ Pipeline ingest ------------

@dataclass
class PageRecord:
    dce_id: str
    project: str
    page_num: int
    page_type: str
    text: str
    image_path: str
    embedding: list[float]

def pdf_to_page_images(pdf_path: Path) -> list[bytes]:
    doc = fitz.open(pdf_path)
    out = []
    for page in doc:
        pix = page.get_pixmap(dpi=200)
        out.append(pix.tobytes("png"))
    return out

async def process_dce(pdf_path: Path, project: str, image_dir: Path) -> list[PageRecord]:
    dce_id = hashlib.sha1(str(pdf_path).encode()).hexdigest()[:12]
    pages = pdf_to_page_images(pdf_path)
    records: list[PageRecord] = []
    for i, img in enumerate(pages):
        ptype = await classify_page(img)
        if ptype in ("plan", "coupe"):
            # double embedding : on prend l'embedding image qui domine
            emb = await embed_image(img)
            text = await vision_extract(img)
        else:
            text = await vision_extract(img) if ptype != "cover" else "(cover)"
            emb = (await embed_text([text]))[0]
        img_path = image_dir / f"{dce_id}_p{i:04d}.png"
        img_path.write_bytes(img)
        records.append(PageRecord(
            dce_id=dce_id, project=project, page_num=i, page_type=ptype,
            text=text, image_path=str(img_path), embedding=emb,
        ))
    return records

async def store_pages(records: list[PageRecord]):
    pool = await asyncpg.create_pool(PG_DSN)
    async with pool.acquire() as con:
        for r in records:
            await con.execute(
                """INSERT INTO dce_pages (dce_id, project, page_num, page_type, text, image_path, embedding)
                   VALUES ($1, $2, $3, $4, $5, $6, $7)""",
                r.dce_id, r.project, r.page_num, r.page_type, r.text, r.image_path, r.embedding,
            )
    await pool.close()

# ------------ Retrieval + génération ------------

async def search(query: str, k: int = 8, type_filter: str | None = None) -> list[dict]:
    q_emb = (await embed_text([query]))[0]
    pool = await asyncpg.create_pool(PG_DSN)
    where = "WHERE page_type = $2" if type_filter else ""
    params = [q_emb, type_filter] if type_filter else [q_emb]
    sql = f"""SELECT id, dce_id, project, page_num, page_type, text, image_path,
                     1 - (embedding <=> $1::vector) AS score
              FROM dce_pages {where}
              ORDER BY embedding <=> $1::vector
              LIMIT {k}"""
    rows = await pool.fetch(sql, *params)
    await pool.close()
    return [dict(r) for r in rows]

ANSWER_PROMPT = """Tu réponds à un architecte qui cherche un détail technique dans
des dossiers DCE archivés. Tu DOIS citer chaque source [dce_id p.page_num] et
indiquer le type (plan, coupe, cctp). Soit concis et technique."""

async def answer_with_vision(query: str) -> str:
    hits = await search(query, k=6)
    content: list[dict] = []
    for h in hits:
        if h["page_type"] in ("plan", "coupe"):
            b64 = base64.b64encode(Path(h["image_path"]).read_bytes()).decode()
            content.append({"type": "text", "text": f"[{h['dce_id']} p.{h['page_num']}] ({h['project']}, {h['page_type']}) :"})
            content.append({"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": b64}})
        else:
            content.append({"type": "text", "text": f"[{h['dce_id']} p.{h['page_num']}] ({h['project']}, {h['page_type']}) {h['text'][:1500]}"})
    content.append({"type": "text", "text": f"\nQuestion : {query}"})
    msg = await CLAUDE.messages.create(
        model=CHAT, max_tokens=2000,
        # Le system (consignes de citation) ne change jamais → on le cache.
        # cache_control sur le DERNIER bloc system : tools+system se cachent ensemble.
        # Les images de support changent à chaque query → elles vont APRÈS, dans le
        # user turn, donc hors préfixe caché (c'est normal, elles ne se cachent pas).
        system=[{"type": "text", "text": ANSWER_PROMPT, "cache_control": {"type": "ephemeral"}}],
        thinking={"type": "adaptive"},        # Opus 4.8 : adaptive only (budget_tokens => HTTP 400)
        output_config={"effort": "high"},     # synthèse technique → on privilégie la qualité
        messages=[{"role": "user", "content": content}],
    )
    # Toujours logger l'usage pour suivre le coût/query en prod (cf. § Pricing).
    logging.info("answer usage: in=%s out=%s cache_read=%s",
                 msg.usage.input_tokens, msg.usage.output_tokens,
                 getattr(msg.usage, "cache_read_input_tokens", 0))
    return msg.content[0].text

# ------------ Demo ------------

async def main():
    image_dir = Path("./images"); image_dir.mkdir(exist_ok=True)
    for pdf in Path("./dce_archives").glob("**/*.pdf"):
        project = pdf.parent.name
        records = await process_dce(pdf, project, image_dir)
        await store_pages(records)
        print(f"Indexed {pdf.name} : {len(records)} pages")

    queries = [
        "Détail d'attache poutre métallique IPN sur dalle béton plancher mixte",
        "Coupe ascenseur PMR pour ERP catégorie 5",
        "CCTP isolation thermique murs extérieurs ITE R6 minimum",
    ]
    for q in queries:
        print("\n== Q :", q)
        print(await answer_with_vision(q))

if __name__ == "__main__":
    asyncio.run(main())

Tu livres avec : Next.js UI qui affiche les plans en mosaïque cliquable, un panel "chunks utilisés" pour la traçabilité, et un dashboard Grafana pour suivre les requêtes. Le retainer mensuel : tu ajoutes 50-200 DCE par mois au pipeline (le cabinet en produit en permanence).

🎯 Patterns courants

Pattern 1 — Vision LLM as OCR

Tu n'utilises plus Tesseract. Tu envoies la page à Claude Sonnet vision avec un prompt structuré (OCR_PROMPT). Marche sur tables, plans, handwriting, FR/EN/AR/etc.

text = await vision_extract(page_image_bytes)

Gain vs Tesseract : ~3× plus précis sur documents complexes, mais 10× plus cher. ROI gagnant dès que la qualité prime.

Pattern 2 — ColPali / ColBERT pour tables

ColPali = embedding par patch d'image (comme ColBERT pour text), permet un match fin sur tables sans OCR. Stockage plus lourd (multi-vector par page) mais retrieval imbattable sur documents visuels.

from colpali_engine.models import ColPali
model = ColPali.from_pretrained("vidore/colpali-v1.2")
# Indexation : un page → N vecteurs (patches)
# Recherche : late-interaction scoring sur les patches

Quand : RCP médicaux, rapports financiers, dossiers d'instructions complexes.

Pattern 3 — Double indexation texte + image

Tu indexes la même page deux fois : avec son texte extrait (embed_text) ET avec son image (embed_image). Tu fusionnes les scores au retrieval.

text_hits = await search(query, embedding_type="text")
image_hits = await search(query, embedding_type="image")
return rrf([text_hits, image_hits])

Quand : tu n'es pas sûr de la modalité dominante. Filet de sécurité.

Pattern 4 — Vision LLM au moment de génération

Tu retrieves des chunks textuels classiques + des images de support (plans, schémas). Tu les envoies tous au LLM vision qui synthétise.

content = [
    {"type": "image", "source": ...},
    {"type": "text", "text": chunk_text},
    {"type": "text", "text": f"Q: {query}"},
]

Marche sur Claude Opus 4.8 / Sonnet 4.6 (vision native FR), GPT-4o, Gemini 2.5 Pro. Pattern dominant 2026.

Optimisation prod — prompt caching sur le préfixe stable. Ton system (consignes de citation) et tes consignes de format ne changent jamais entre deux requêtes. Mets un cache_control: {"type": "ephemeral"} sur le dernier bloc system : tu paies ~0,1× sur cette portion à chaque query au lieu du plein tarif. Attention : le préfixe doit être byte-identique — les images de support changent à chaque requête, donc elles vont APRÈS le breakpoint (elles ne se cachent pas, c'est normal). Vérifie usage.cache_read_input_tokens > 0 sinon un invalidant silencieux (timestamp, ordre JSON) casse le cache.

Pattern 5 — Parallélisme d'ingestion (le vrai goulot)

L'ingestion de 200 000 pages est I/O-bound : chaque page = 1-2 appels réseau (classify + vision/embed). En séquentiel, c'est des heures. Utilise AsyncAnthropic + asyncio.gather borné par un sémaphore (sinon tu te fais rate-limiter en 429), et compte sur les retries typés du SDK.

sem = asyncio.Semaphore(8)  # borne le parallélisme → évite les 429

async def process_page_guarded(img: bytes):
    async with sem:
        try:
            return await process_page(img)
        except anthropic.RateLimitError:
            await asyncio.sleep(2)      # le SDK retry déjà 2× ; ceci est le filet
            return await process_page(img)

results = await asyncio.gather(*[process_page_guarded(p) for p in pages])

Configure aussi AsyncAnthropic(max_retries=4, timeout=60) : sur un batch de 200 k pages, une poignée de OverloadedError/APITimeoutError est statistiquement certaine — sans retry, ton ingestion meurt à 70 %.

Pattern 6 — Audio retrieval via Whisper + chunking

Tu transcribe l'audio (Whisper), tu chunks par speaker turn (voir fichier 05), tu embed le texte. Pour la lecture, tu garde le start_ts pour rejouer le passage.

async def index_call(audio_path: str):
    transcript = await whisper.transcribe(audio_path)
    for turn in transcript["segments"]:
        emb = await embed_text([turn["text"]])[0]
        await store({"text": turn["text"], "start_ts": turn["start"], "embedding": emb, "audio_id": ...})

Quand : support call centers, podcasts, réunions enregistrées, notaires (signatures audio).

Pattern 7 — Domain-specific fine-tuning

Pour les domaines très spécialisés (radiologie, plans techniques, mode), on fine-tune un CLIP/DinoV2/SigLIP sur des paires (image, légende) métier. ~10k paires labellisées, 1-3 jours sur 1 GPU. Boost massif vs modèles génériques.

Quand : ton domaine est très éloigné des données d'entraînement web. Évite si modèle générique convient (test toujours d'abord).

🔄 Versions & écosystème 2026

Modèle / outil	Statut 2026
Cohere Embed v4 Multimodal	Standard, multilingue FR fort, 1536 dims
Voyage Multimodal v3	Bon FR, support PDF natif, 32K context
OpenAI CLIP	Toujours utile en local, gratuit
SigLIP (Google)	Meilleur CLIP open-source
ColPali v1.2 / v1.3	Référence tables/PDF complexes
Claude Opus 4.8 vision	Flagship vision 2026, FR natif, 1M ctx (5 $/25 $ /M tok)
Claude Sonnet 4.6 vision	Bon rapport qualité/prix vision (3 $/15 $ /M tok)
Claude Haiku 4.5 vision	Classification de page bon marché (1 $/5 $ /M tok)
GPT-4o vision	Alternative solide
Gemini 2.5 Pro vision	1 M tokens context, excellent sur très longs PDF
Whisper v4 (OpenAI)	Multilingue FR très bon
WhisperX	Whisper + diarization (speaker detection)
PyMuPDF (fitz)	PDF → images, gratuit, rapide
Unstructured.io	Pipeline OCR + structure (CV-based)
LayoutLMv3	Open-source pour layout analysis

Tendance 2026 : on ne fait plus d'OCR Tesseract sur des PDF mixtes. Tout passe par vision LLM ou ColPali. C'est la différence générationnelle entre un junior (Tesseract) et un sénior (vision-native).

⚠️ Pitfalls

1. Image embeddings sans normalisation → résolution différente entre indexation et query → drift. Toujours resize à dim cohérente (Cohere recommande 448×448 ou similaire).
2. Stocker les images en blob SQL → ta DB grossit anormalement. Stocke sur S3/MinIO + path en colonne.
3. OCR Tesseract sur des plans techniques → désastre. Vision LLM ou ColPali obligatoire.
4. Vision LLM sur toutes les pages → coût explose. Routes : page-cover/page-cctp → texte ; page-plan/coupe → vision.
5. Embedding multimodal pas multilingue → CLIP original est faible en FR. Cohere Embed v4 ou Voyage Multimodal.
6. Pas de page_type metadata → tu ne peux pas filtrer "uniquement plans" au retrieval. Toujours classifier + stocker.
7. PDF scanné mal traité → si DPI source < 150, l'OCR foire. Tester sur échantillon, monter à 300 dpi en preprocessing.
8. Confondre similarity et useful → l'embedding trouve une image visuellement proche mais sémantiquement non pertinente. Toujours rerank avec un vision LLM si l'enjeu est haut.
9. Token explosion sur images → Claude vision compte ~1 568 tokens par image en haute res. 8 images = 12 k tokens. Budget en conséquence.
10. Pas de citation visuelle → tu cites "[plan 23]" mais l'utilisateur ne sait pas quel plan. UI : toujours rendre la miniature cliquable.
11. Pas de cache d'embedding images → tu re-ingest, tu paies de nouveau. Hash MD5 du contenu image en clé.
12. Latence vision LLM sous-estimée → 3-5 s par appel avec 4 images. Streaming + UI optimiste obligatoires.
13. Mauvais format media_type → tu envoies image/jpg au lieu de image/jpeg, Claude rejette. Validate avant envoi.
14. Syntaxe thinking obsolète → sur Opus 4.8/4.7, thinking: {type: "enabled", budget_tokens: N} renvoie un HTTP 400. Idem temperature/top_p. Utilise thinking: {type: "adaptive"} + output_config: {effort: "low"|"medium"|"high"}. Haiku/Sonnet 4.6 ne prennent pas de budget de thinking.
15. Pas d'observabilité du coût → tu factures un retainer mais tu ne logs pas resp.usage. Sans input_tokens/output_tokens/cache_read_input_tokens par query, tu ne peux pas défendre ta marge ni détecter une dérive (un client qui upload des images 4K = explosion silencieuse). Logge l'usage à chaque appel.
16. Mélanger sécurité et coût sur le routage → router "table → ColPali" est un choix de qualité ; router "cover → texte (1 ligne)" est un choix de coût. Ne les confonds pas : une page mal classée cover qui est en réalité un plan = signal perdu pour toujours. Mets un seuil de confiance sur le classifier et un fallback "double indexation" en cas de doute.
17. Recréer un pool DB par requête → l'exemple end-to-end fait asyncpg.create_pool() + .close() à chaque appel pour rester lisible. En prod c'est interdit : tu paies le handshake TCP+TLS+auth Postgres à chaque query (10-50 ms de latence morte) et tu épuises les connexions sous charge. Crée un pool au démarrage (lifespan FastAPI / @app.on_event) et injecte-le. Sur 10 000 queries/mois c'est invisible ; sur 50 req/s c'est ce qui fait tomber ton service.
18. Double-compter les retries du SDK → le try/except RateLimitError: sleep + retry du Pattern 5 s'empile sur les max_retries du SDK (qui retry déjà 429/529/timeout avec backoff). Résultat : un retry manuel par-dessus 4 retries SDK = jusqu'à 8 tentatives, et ta métrique "appels émis" ment. Choisis : soit tu fais confiance au SDK (max_retries=4) et tu ne rattrapes que les erreurs non-retryables par le SDK, soit tu mets max_retries=0 et tu gères tout. Ne fais pas les deux en aveugle.
19. media_type déduit de l'extension → un .png renommé qui est en réalité un JPEG fait rejeter l'image. Sniff les magic bytes (b"\x89PNG", b"\xff\xd8\xff" pour JPEG) ou laisse PyMuPDF normaliser en PNG à l'ingestion, plutôt que de faire confiance au nom de fichier.

💰 Pricing / ROI client

Coût d'un pipeline multimodal d'ingestion typique (1 000 PDF × 200 pages = 200k pages) :

Étape	Modèle / coût
PDF → images (PyMuPDF, local)	~0,5 €/1k pages (compute)
Classification page (Haiku 4.5 vision, 1 $/5 $)	~0,002 €/page → 400 €
Vision extraction text (Sonnet 4.6 sur 50 % des pages)	~0,015 €/page → 1 500 €
Embedding multimodal Cohere v4	~0,0001 €/embed → 40 €
Storage images (S3, 200 GB)	~25 €/mois
pgvector storage	~30 €/mois
Total one-shot ingest	~2 000 €
Run mensuel	~80 €/mois

Run par query :

• search Cohere : 0,0001 €
• generate avec 4 images : ~0,025 € sur Sonnet 4.6, ~0,06-0,12 € sur Opus 4.8 (selon thinking/effort)
• → 0,025 €/query en routine, 10 000 queries/mois = 250 €/mois

Arbitrage modèle de génération : tu réserves Opus 4.8 (flagship, le plus cher, 5 $/25 $ /M tok à 1M ctx) aux requêtes à enjeu — lecture de table d'interactions médicamenteuses, détail technique engageant ta responsabilité. Pour le visual search e-commerce (pas de génération, juste du retrieval), tu n'appelles aucun LLM de génération : embed + cosine, point. Le coût LLM de génération ne doit jamais dominer ta facture si ton routage est correct.

Économie du prompt caching (souvent mal calculée). Le cache_control sur le system te fait payer ~0,1× la portion cachée en lecture, contre ~1,25× à la première écriture (TTL 5 min). Donc le cache est rentable dès la 2ᵉ query qui partage le préfixe (1,25× + 0,1× = 1,35× vs 2× sans cache). Mais attention au piège multimodal : tes images de support changent à chaque query et passent après le breakpoint — elles ne se cachent JAMAIS, c'est attendu. Ce qui se cache, c'est le system + les tools (≈ quelques centaines de tokens). Sur un system court (< 1024 tok sur Opus 4.8, le minimum cacheable), le cache ne s'active même pas silencieusement : vérifie usage.cache_read_input_tokens > 0 avant de te féliciter. Si ton system fait 300 tokens, le gain de cache est dans le bruit — le vrai levier coût reste le routage (vision sur 30-50 % des pages) et le cache d'embedding par hash, pas le prompt caching.

ROI client :

• Cabinet d'archi : 22 k€/mois récupérés (cf. cas 1) → ROI < 1 mois.
• Marketplace visual search : +540 k€/an de revenus incrémentaux.
• Pharma : un risque clinique évité = priceless.

Mission packaging :

• POC technique (5-7 j, 7-10 k€) : démontrer la valeur sur 100 documents mixtes.
• Implémentation ingest + search (15-25 j, 22-37 k€) : pipeline complet.
• UI + intégration (5-10 j, 7-15 k€) : visual search frontend, click-to-PDF.
• Retainer ingest (2-3 j/mois, 3-5 k€/mois) : nouveaux docs, monitoring, re-eval.

🧪 Testing / Eval

Tests d'ingestion :

@pytest.mark.asyncio
async def test_classify_page_plan():
    img = Path("fixtures/plan_dalle.png").read_bytes()
    assert await classify_page(img) == "plan"

@pytest.mark.asyncio
async def test_vision_extract_keeps_refs():
    img = Path("fixtures/coupe_ipn.png").read_bytes()
    text = await vision_extract(img)
    assert "IPN" in text
    assert any(re.search(r"\d+", text))  # contient des cotes

Eval multimodale (Ragas adapté) :

EVAL_SET = [
    {
        "q": "Détail attache poutre IPN sur dalle béton",
        "expected_pages": [("dce_2019_villa_x", 47), ("dce_2021_ehpad_y", 132)],
        "must_contain_in_answer": ["IPN", "scellement chimique"],
    },
]

async def eval_multimodal():
    ok_recall = 0
    ok_quality = 0
    for ex in EVAL_SET:
        hits = await search(ex["q"], k=10)
        retrieved = {(h["dce_id"], h["page_num"]) for h in hits}
        if any(p in retrieved for p in ex["expected_pages"]):
            ok_recall += 1
        ans = await answer_with_vision(ex["q"])
        if all(c in ans for c in ex["must_contain_in_answer"]):
            ok_quality += 1
    return {
        "recall@10": ok_recall / len(EVAL_SET),
        "answer_quality": ok_quality / len(EVAL_SET),
    }

Métriques prod :

• Visual recall@10 : ≥ 0.75 attendu
• Latence p95 : 2-4s (vision LLM est cher en time)
• Coût/query : ≤ 0,03 € sinon réinjecter du caching

🔁 Quand utiliser / éviter

Multimodal RAG utile	À éviter
BTP, archi (plans, coupes)	Corpus 100 % textuel propre
Pharma (notices, tables interaction)	Volume très haut (> 1 M pages, coût explose)
Mode / retail (recherche par image)	Latence sub-second exigée (vision LLM = lent)
Industrie (notices techniques, schémas)	Budget < 10 k€ (coût ingestion non amorti)
Audio (calls, podcasts) avec retrieval temporal	Pure text Q&A simple
Immo (photos biens + plans)	Domaine où text-only RAG donne déjà > 0.85 recall

🏋️ Exercices

Progressifs. Les premiers construisent ; les derniers te demandent de casser, mesurer et défendre un chiffre — c'est le niveau attendu en mission.

1. Routeur de page + double indexation conditionnelle

Objectif : implémenter classify_page (Haiku 4.5 + structured output) qui route chaque page vers embed_image, embed_text ou ColPali, avec un seuil de confiance et un fallback double-indexation sous le seuil. Indice/Solution : renvoie {"label": ..., "confidence": ...} via output_config.format. Si confidence < 0.7, indexe la page DEUX fois (texte + image) et fusionne au retrieval par RRF. Mesure le recall avant/après le fallback sur 50 pages annotées — le gain doit être visible sur les pages ambiguës (page de garde avec un petit plan en médaillon).

2. Pipeline d'ingestion parallèle qui survit aux 429

Objectif : ingérer 1 000 pages en parallèle sans te faire couper par le rate-limit, avec reprise sur erreur. Indice/Solution : asyncio.Semaphore(8) + asyncio.gather, AsyncAnthropic(max_retries=4, timeout=60), et un cache d'embedding par hashlib.sha256(img_bytes). Relance le script deux fois : la 2ᵉ passe doit être quasi gratuite (tout en cache). Compte les appels LLM réellement émis — si la 2ᵉ passe en émet > 0, ton cache fuit.

3. Eval multimodale honnête (et casse-la)

Objectif : étendre eval_multimodal pour séparer recall@k (le bon doc est-il retrouvé) et faithfulness (la réponse cite-t-elle la bonne page sans halluciner). Indice/Solution : ajoute un juge LLM qui vérifie que chaque affirmation de la réponse est supportée par une image/texte présent dans le contexte. Puis casse ton système : injecte une page visuellement proche mais sémantiquement fausse (deux plans de poutre quasi identiques, ferraillage différent) et montre que le cosine seul la remonte. Corrige avec un rerank vision LLM et re-mesure.

4. Rends-le production-grade : observabilité du coût

Objectif : instrumenter le pipeline pour produire un dashboard coût/query et coût/ingestion à partir de resp.usage. Indice/Solution : logge input_tokens, output_tokens, cache_read_input_tokens par appel, agrège par dce_id et par jour. Simule un client qui upload des images 4K (≈3× plus de tokens image sur Opus 4.8) et montre la dérive sur le dashboard. C'est exactement ce que tu présentes en revue de retainer pour justifier ta facture.

5. Défends le chiffre : Opus 4.8 vs Sonnet 4.6 sur la table pharma

Objectif : sur 30 questions d'interactions médicamenteuses (pages-table), comparer Sonnet 4.6 (effort: medium) et Opus 4.8 (effort: high, adaptive thinking) sur précision ET coût/query, puis recommander un modèle avec un chiffre à l'appui. Indice/Solution : tu vas trouver Opus 4.8 plus précis mais 3-5× plus cher. La bonne réponse de sénior n'est pas "le plus cher" : c'est un routage par enjeu (Opus seulement quand la table contient une contre-indication grossesse/insuffisance rénale, Sonnet sinon), classé par un Haiku en amont. Chiffre le coût mensuel des trois stratégies (tout-Sonnet / tout-Opus / routé) et défends le ratio précision/€.

6. Casse le ColPali, puis rends-le viable

Objectif : indexer 500 pages-table en ColPali (multi-vecteur), mesurer le coût de stockage et la latence p95, puis ramener la latence sous 400 ms. Indice/Solution : ColPali stocke N vecteurs/page → ta table pgvector explose et le late-interaction scoring devient lent. Leviers : pré-filtrer par metadata (page_type = 'table') avant le scoring multi-vecteur, quantizer les vecteurs (PQ/int8), et limiter le late-interaction au top-50 d'un premier passage mono-vecteur grossier. Montre le trade-off recall ↓ vs latence ↓ sur une courbe.

7. Défends ton prompt caching (et prouve qu'il marche)

Objectif : ajouter cache_control sur le system de answer_with_vision, puis prouver chiffres à l'appui que le cache est lu — ou expliquer pourquoi il ne l'est pas. Indice/Solution : lance 5 queries identiques de préfixe et logge usage.cache_creation_input_tokens (1ʳᵉ query) et usage.cache_read_input_tokens (queries 2-5). Si cache_read reste à 0, deux causes probables : (1) ton system fait < 1024 tokens (minimum cacheable sur Opus 4.8 — il ne se cache pas, sans erreur), ou (2) un invalidant silencieux dans le préfixe (un datetime.now(), un ordre JSON non déterministe, le set de tools qui bouge). Corrige, re-mesure, et calcule le coût réel/query avec et sans cache. Conclusion attendue de sénior : sur un system court le prompt caching est dans le bruit — le vrai levier coût reste le routage et le cache d'embedding. Sais défendre pourquoi tu actives ou n'actives pas le cache, pas juste "j'ai mis cache_control".

🎤 En entretien

• "Pourquoi pas juste de l'OCR Tesseract + RAG texte sur des plans ?" → Tesseract échoue sur le layout 2D non linéaire (cotes, légendes, annotations dispersées) ; un vision LLM ou ColPali préserve la structure spatiale, qui EST le signal sur un plan technique.
• "Comment tu empêches le coût d'exploser sur 200 k pages ?" → Routage par type de page (vision LLM uniquement sur 30-50 %), cache d'embedding par hash de contenu, et modèle de génération réservé aux requêtes à enjeu (Opus 4.8) — le reste en retrieval pur sans LLM.
• "Ton embedding remonte une image visuellement proche mais fausse. Que fais-tu ?" → Je distingue similarity et usefulness : cosine seul ne suffit pas sur les cas à enjeu, je rajoute un rerank vision LLM et une eval faithfulness qui vérifie le support de chaque affirmation.
• "Opus 4.8 ou Sonnet 4.6 pour la génération ?" → Ça dépend de l'enjeu, pas du budget : Opus 4.8 (flagship, adaptive thinking + effort high) sur les requêtes engageant la responsabilité métier ; Sonnet 4.6 en routine ; routage classé par un Haiku 4.5 en amont. Je défends le choix avec un coût/query mesuré, pas une intuition.
• "Une query texte peut-elle vraiment matcher une image, ou c'est de la magie marketing ?" → Oui, à condition que l'embedding texte et l'embedding image vivent dans le même espace (modèle entraîné en contrastive learning sur paires texte/image : Cohere v4, Voyage MM). Le cosine est alors défini entre modalités. Si tu embeddes le texte avec un modèle et les images avec un autre, les espaces ne sont pas alignés et le cross-modal retrieval renvoie du bruit — erreur classique de junior.
• "Comment tu instrumentes le coût d'un pipeline multimodal en prod ?" → Je logge usage (input/output/cache_read_input_tokens) à chaque appel LLM, agrégé par dce_id et par jour. Sans ça je ne peux ni défendre ma marge en revue de retainer, ni détecter une dérive silencieuse — typiquement un client qui se met à uploader des images 4K (jusqu'à ~4 784 tokens/image sur Opus 4.8 en pleine résolution) et triple ma facture sans que personne ne s'en aperçoive.

🔗 Liens

• Cohere Embed v4 : https://docs.cohere.com/docs/cohere-embed
• Voyage Multimodal embeddings : https://docs.voyageai.com/docs/multimodal-embeddings
• ColPali paper : https://arxiv.org/abs/2407.01449
• ColPali code : https://github.com/illuin-tech/colpali
• CLIP (OpenAI) : https://github.com/openai/CLIP
• SigLIP (Google) : https://huggingface.co/google/siglip-base-patch16-224
• Claude vision docs : https://docs.anthropic.com/en/docs/build-with-claude/vision
• Whisper v4 : https://github.com/openai/whisper
• WhisperX (diarization) : https://github.com/m-bain/whisperX
• Unstructured.io : https://unstructured.io/
• PyMuPDF : https://pymupdf.readthedocs.io/