Aller au contenu principal

Intégration OCR + RAG

Construire un système documentaire intelligent

Le RAG (Retrieval-Augmented Generation) est la clé pour interroger de grands corpus documentaires. En combinant l’OCR de Mistral, les embeddings et un vector store, vous créez un système capable de répondre à n’importe quelle question sur vos archives.

Pourquoi OCR + RAG ?

Le problème fondamental du RAG documentaire est que les chunks indexés doivent être du texte. Les embeddings multimodaux ne sont pas encore assez performants pour comparer directement des questions textuelles avec des images de documents.

La solution : utilisez l’OCR pour convertir vos documents en texte structuré de haute qualité, puis indexez ce texte dans votre vector store.

Architecture du pipeline

Le pipeline complet se décompose en 5 étapes :

  1. Ingestion : réception des documents (PDF, images, DOCX)
  2. OCR + Annotations : extraction du texte et des métadonnées structurées
  3. Chunking : découpage intelligent du texte en segments
  4. Embedding + Indexation : vectorisation et stockage dans le vector store
  5. Requête : recherche sémantique puis génération de réponse

Implémentation complète

Étape 1 : Extraction OCR enrichie

import os
import base64
from mistralai import Mistral
from mistralai.extra import response_format_from_pydantic_model
from pydantic import BaseModel, Field
from typing import Optional

client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

class DescriptionImage(BaseModel):
    type_image: str = Field(..., description="Type: graph, table, photo, logo")
    description: str = Field(..., description="Description du contenu")

def extraire_document_enrichi(chemin: str) -> list[dict]:
    """Extrait le contenu enrichi d'un document pour le RAG."""
    with open(chemin, "rb") as f:
        doc_b64 = base64.b64encode(f.read()).decode("utf-8")

    response = client.ocr.process(
        model="mistral-ocr-latest",
        document={
            "type": "document_base64",
            "document_base64": doc_b64
        },
        table_format="markdown",
        bbox_annotation_format=response_format_from_pydantic_model(
            DescriptionImage
        ),
        include_image_base64=False  # Pas besoin des images pour le RAG
    )

    pages = []
    for page in response.pages:
        contenu = page.markdown

        # Remplacer les références images par leurs descriptions
        for image in page.images:
            if image.annotation:
                desc = DescriptionImage.model_validate_json(
                    image.annotation
                )
                placeholder = f"![{image.id}]({image.id})"
                remplacement = (
                    f"[{desc.type_image}: {desc.description}]"
                )
                contenu = contenu.replace(placeholder, remplacement)

        pages.append({
            "page": page.index + 1,
            "contenu": contenu,
            "header": page.header or "",
            "footer": page.footer or "",
        })

    return pages

Étape 2 : Chunking intelligent

def chunker_par_sections(
    texte: str,
    taille_max: int = 1000,
    chevauchement: int = 200
) -> list[str]:
    """Découpe le texte en respectant les sections Markdown."""
    sections = texte.split("\n## ")
    chunks = []

    for section in sections:
        if not section.strip():
            continue

        # Reconstituer le titre de section
        if not section.startswith("## "):
            section = "## " + section

        if len(section) <= taille_max:
            chunks.append(section)
        else:
            # Sous-découpage par paragraphes
            paragraphes = section.split("\n\n")
            chunk_courant = ""

            for para in paragraphes:
                if len(chunk_courant) + len(para) > taille_max:
                    if chunk_courant:
                        chunks.append(chunk_courant.strip())
                    chunk_courant = para
                else:
                    chunk_courant += "\n\n" + para

            if chunk_courant:
                chunks.append(chunk_courant.strip())

    return chunks

Étape 3 : Embedding et indexation

import numpy as np

def creer_index(documents: dict[str, list[dict]]) -> list[dict]:
    """Crée un index vectoriel à partir des documents extraits."""
    index = []

    for nom_doc, pages in documents.items():
        # Concaténer toutes les pages
        texte_complet = "\n\n".join(p["contenu"] for p in pages)

        # Chunking
        chunks = chunker_par_sections(texte_complet)

        for i, chunk in enumerate(chunks):
            # Embedding
            response = client.embeddings.create(
                model="mistral-embed",
                inputs=[chunk]
            )

            index.append({
                "document": nom_doc,
                "chunk_id": i,
                "texte": chunk,
                "embedding": np.array(response.data[0].embedding),
                "page": pages[0]["page"] if pages else 0,
            })

    return index


def rechercher_chunks(
    question: str,
    index: list[dict],
    top_k: int = 5
) -> list[dict]:
    """Recherche les chunks les plus pertinents."""
    response = client.embeddings.create(
        model="mistral-embed",
        inputs=[question]
    )
    q_emb = np.array(response.data[0].embedding)

    scores = []
    for item in index:
        similarite = np.dot(q_emb, item["embedding"]) / (
            np.linalg.norm(q_emb) * np.linalg.norm(item["embedding"])
        )
        scores.append((similarite, item))

    scores.sort(key=lambda x: x[0], reverse=True)
    return [item for _, item in scores[:top_k]]

Étape 4 : Génération de réponse

def repondre_question(question: str, index: list[dict]) -> str:
    """Répond à une question en utilisant le RAG."""
    # Recherche des chunks pertinents
    resultats = rechercher_chunks(question, index, top_k=5)

    # Construction du contexte
    contexte = "\n\n---\n\n".join(
        f"[Source: {r['document']}, Chunk {r['chunk_id']}]\n{r['texte']}"
        for r in resultats
    )

    # Génération de la réponse
    response = client.chat.complete(
        model="mistral-small-latest",
        messages=[
            {
                "role": "system",
                "content": (
                    "Vous êtes un assistant documentaire. Répondez "
                    "aux questions en vous basant uniquement sur le "
                    "contexte fourni. Citez les sources."
                )
            },
            {
                "role": "user",
                "content": (
                    f"Contexte :\n{contexte}\n\n"
                    f"Question : {question}"
                )
            }
        ]
    )

    return response.choices[0].message.content

Utilisation en production

Pour un déploiement en production, remplacez le stockage numpy par un vrai vector store (Pinecone, Weaviate, Qdrant, ChromaDB) :

# Exemple avec ChromaDB
import chromadb

chroma_client = chromadb.Client()
collection = chroma_client.create_collection(name="documents")

# Indexation
for item in index:
    collection.add(
        ids=[f"{item['document']}_{item['chunk_id']}"],
        embeddings=[item["embedding"].tolist()],
        documents=[item["texte"]],
        metadatas=[{"document": item["document"]}]
    )

# Recherche
resultats = collection.query(
    query_embeddings=[q_emb.tolist()],
    n_results=5
)

Points clés à retenir

  • Le pipeline OCR + RAG convertit vos documents en une base de connaissances interrogeable
  • L’enrichissement par annotations (descriptions d’images) évite la perte d’information
  • Le chunking intelligent respecte les sections et paragraphes du document
  • Utilisez un vrai vector store en production (ChromaDB, Pinecone, Qdrant)
  • La qualité de l’extraction OCR détermine la qualité des réponses RAG
← Lecon precedenteQnA multi-documentsLecon suivante →Pipeline de production