Skip to main content
Lesson 15 of 16 · Document AI: OCR and Document Intelligence

OCR + RAG Integration

Lesson 15 of 16 Updated November 3, 2026 Developer
On this page
  1. Building an intelligent document system
  2. The five-step architecture
  3. Step 1: Extraction that loses nothing
  4. Step 2: Chunking where the document allows
  5. Step 3: Vectorize and search
  6. Step 4: Generate a sourced answer
  7. From prototype to vector store
  8. Key takeaways

Update November 2, 2026 — OCR. OCR 4.0 (mistral-ocr-4-0) was retired on September 30, 2026. Use OCR 4.1 (mistral-ocr-4-1), generally available since August 31 at the same price, or the alias mistral-ocr-latest, which has pointed to it since July 16. OCR 4.1 adds confidence scores at the block level. Addendum November 3, 2026: A local PDF is transmitted in base64 as {"type": "document_url", "document_url": "data:application/pdf;base64,…"}; the document_base64 type from older examples exists neither in the documentation nor in the mistralai SDK, and those examples would have failed. Sources: OCR Processor · Document QnA. Addendum November 3: Image annotations are read from image.image_annotation (SDK mistralai 3.7.0), not image.annotation.

Building an intelligent document system

Retrieval-Augmented Generation (RAG) is the technique that makes large document corpora queryable. By combining Mistral’s OCR, embeddings, and a vector store, you create a system capable of answering questions about your archives—ten years of reports, a contract repository, or a digitized technical library.

Why use OCR instead of indexing files as-is? Because the fundamental challenge of document RAG is that indexed chunks must be text. Multimodal embeddings are not yet performant enough to directly compare text questions with document images. OCR serves as the bridge: it converts your documents into high-quality structured text, which you then index in your vector store.

The five-step architecture

It all starts with ingestion, meaning the receipt of documents, whether they arrive as PDFs, images, or DOCX files. Next comes OCR with annotations, producing both text and structured metadata. Chunking cuts this text into coherent segments, embedding vectorizes them and stores them in the vector store, and the query closes the loop: semantic search among the vectors, followed by answer generation. Each step conditions the next, and extraction determines everything else—if OCR misses it, no search will find it.

The following sections detail the code for the four steps you will write yourself; ingestion depends on your specific intake system.

Step 1: Extraction that loses nothing

The OCR call here is configured for RAG, evident in three choices. table_format="markdown" preserves tables in a format the LLM can read. Bounding box annotation describes images, and the code subsequently replaces each ![image](image) reference with its text description: a revenue distribution chart becomes an indexable sentence instead of a gap in the text. Finally, include_image_base64=False avoids fetching megabytes of pixels that the index has no use for.

import os
import base64
from mistralai.client 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="Content description")

def extraire_document_enrichi(chemin: str) -> list[dict]:
    """Extracts enriched content from a document for 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_url",
            "document_url": f"data:application/pdf;base64,{doc_b64}"
        },
        table_format="markdown",
        bbox_annotation_format=response_format_from_pydantic_model(
            DescriptionImage
        ),
        include_image_base64=False  # Images not needed for RAG
    )

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

        # Replace image references with their descriptions
        for image in page.images:
            if image.image_annotation:
                desc = DescriptionImage.model_validate_json(
                    image.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

Step 2: Chunking where the document allows

Splitting every thousand characters cuts mid-sentence and separates a heading from its content. Since OCR gives you Markdown, leverage its structure: the function first segments on level-two headings, keeps the entire section only if it fits within the maximum size, and falls back to paragraphs only when it overflows. Each chunk thus retains its section heading, which significantly improves semantic search relevance.

def chunker_par_sections(
    texte: str,
    taille_max: int = 1000,
    chevauchement: int = 200
) -> list[str]:
    """Splits text respecting Markdown sections."""
    sections = texte.split("\n## ")
    chunks = []

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

        # Reconstruct section heading
        if not section.startswith("## "):
            section = "## " + section

        if len(section) <= taille_max:
            chunks.append(section)
        else:
            # Sub-split by paragraphs
            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

Indexing vectorizes each chunk with mistral-embed and stores, alongside the vector, the document name and chunk ID—without this metadata, you can answer but not cite sources. Search vectorizes the question the same way, then calculates cosine similarity normalized by the norms of both vectors, preventing a long chunk from winning simply because it is long.

import numpy as np

def creer_index(documents: dict[str, list[dict]]) -> list[dict]:
    """Creates a vector index from extracted documents."""
    index = []

    for nom_doc, pages in documents.items():
        # Concatenate all 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]:
    """Searches for the most relevant chunks."""
    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]]

Step 4: Generate a sourced answer

The final piece assembles the top five chunks into a context, each prefixed with its document and number, and imposes a strict instruction on the model: answer only from the provided context and cite sources. This system prompt distinguishes a document assistant from a model that improvises, and it is why you preserved metadata during indexing.

def repondre_question(question: str, index: list[dict]) -> str:
    """Answers a question using RAG."""
    # Search for relevant chunks
    resultats = rechercher_chunks(question, index, top_k=5)

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

    # Generate answer
    response = client.chat.complete(
        model="mistral-small-latest",
        messages=[
            {
                "role": "system",
                "content": (
                    "You are a document assistant. Answer "
                    "questions based solely on the provided "
                    "context. Cite sources."
                )
            },
            {
                "role": "user",
                "content": (
                    f"Context:\n{contexte}\n\n"
                    f"Question: {question}"
                )
            }
        ]
    )

    return response.choices[0].message.content

From prototype to vector store

An in-memory index works as long as you stay under a few thousand chunks. Beyond that, linear search collapses and nothing survives a restart: switch to a real vector store (Pinecone, Weaviate, Qdrant, ChromaDB). The switch is inexpensive, as shown in this ChromaDB variant where only storage and querying change—your extraction and chunking functions remain intact.

# Example with ChromaDB
import chromadb

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

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

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

Key takeaways

  • The OCR + RAG pipeline converts your documents into a queryable knowledge base
  • Enrichment via annotations (image descriptions) prevents information loss
  • Intelligent chunking respects document sections and paragraphs
  • Use a real vector store in production (ChromaDB, Pinecone, Qdrant)
  • OCR extraction quality determines RAG answer quality