Home
Studio

Ingestion

Ingestion transforms raw documents into searchable chunks indexed in a vector store. You can swap pipeline components based on your needs.

A FileLoader reads raw bytes from a source (local filesystem, cloud storage, or any custom source) and returns a File object.

A DocumentExtractor converts that File into a structured Document by parsing PDFs via OCR, converting HTML to Markdown, or reading plain text depending on the file type.

A TextSplitter divides the Document into smaller DocumentChunk objects. The strategy and chunk size affect retrieval quality.

Optional DocumentProcessor stages (such as a ChunkEnricher or SummaryEnricher) add metadata to each chunk before indexing, for example an LLM-generated summary or entity tags.

An Embedder converts each chunk into a vector, and the store persists both the vector and the raw text for search.

Pipeline

Pipeline

Pipeline is the main entry point. It handles file loading, extraction, chunking, embedding, and indexing. It supports batch processing with concurrency, progress tracking, and checkpointing.

from mistralai.search.toolkit.ingestion.pipelines import Pipeline
from mistralai.client import Mistral
from mistralai.search.toolkit.embedding import MistralEmbedder, MODEL_1024_EMBEDDING
from mistralai.search.toolkit.ingestion.extractors import PlainTextExtractor
from mistralai.search.toolkit.ingestion.loaders import FilesystemFileLoader
from mistralai.search.toolkit.ingestion.text_splitters import CharacterTextSplitter
from mistralai.search.toolkit.ingestion.enrichment import SummaryEnricher
from mistralai.search.toolkit.llm import MistralChat, LLMConfig
from mistralai.search.toolkit.mistral_search_index import MistralSearchIndexService

mistral_client = Mistral(api_key="...")
llm = MistralChat(client=mistral_client, config=LLMConfig(model="mistral-small-latest"))

index_name = "my_collection"
vector_store = await MistralSearchIndexService(client=mistral_client).get_search_index(index_name)

pipeline = Pipeline(
    loader=FilesystemFileLoader(),
    extractor=PlainTextExtractor(),
    text_splitter=CharacterTextSplitter(chunk_size=512),
    processors=[SummaryEnricher(llm_provider=llm)],
    embedder=MistralEmbedder(client=mistral_client, model_name=MODEL_1024_EMBEDDING),
    stores=vector_store,
)

num_chunks = await pipeline.run(documents=["doc1.txt", "doc2.txt"])
i
Information

index_name, passed to get_search_index, selects the Mistral Search Index that your chunks are indexed into. Create the index first, and use the same name when you retrieve documents later. Other backends work the same way: pass their store to stores.

Checkpointing

For large document collections, enable checkpointing to recover from failures. The progress callback needs tqdm, which isn't a core dependency: install it with uv add tqdm.

from mistralai.search.toolkit.ingestion.progress import create_tqdm_progress_callback
from mistralai.search.toolkit.ingestion.enrichment import SummaryEnricher
from mistralai.search.toolkit.storage import FilesystemStorage

pipeline = Pipeline(
    loader=FilesystemFileLoader(),
    extractor=extractor,
    text_splitter=splitter,
    processors=[SummaryEnricher(llm_provider=llm)],
    embedder=embedder,
    stores=vector_store,
    checkpoint_store=lambda: FilesystemStorage("./checkpoints"),
)

num_chunks = await pipeline.run(
    documents=document_paths,
    use_checkpoint=True,
    progress_callback=create_tqdm_progress_callback(),
)

On restart, documents with an existing checkpoint skip extraction (for example, OCR). Splitting, enrichment, embedding, and indexing run again.

Route files by type

Route files by type

RoutedPipeline routes files to protocol-specific pipelines based on file extension and MIME type. Use it to ingest mixed-format document collections: it selects the extractor and processing pipeline for each file type.

Protocols

RoutedPipeline ships with 9 built-in protocols. Each protocol maps a set of file types to the appropriate extractor pipeline:

ProtocolFile types
ocrPDF, DOCX, PPTX, ODT, EPUB
plain_textTXT, MD, CSV, JS, PY, JSON, YAML, ...
htmlHTML, HTM
xlsxXLSX, XLS, XLSM, XLSB, XLTX, ODS, ODF
numbersNUMBERS
legacy_officeDOC, PPT, HWP, HWPX (only when PYMUPDF_PRO_LICENSE_KEY is set)
emailEML, MSG
imagePNG, JPEG, GIF, WebP
audioMP3, WAV, M4A, FLAC, OGG

Use the default protocols

Map each protocol you need to a Pipeline:

from mistralai.search.toolkit.ingestion.pipelines import RoutedPipeline

router = RoutedPipeline({
    "ocr": ocr_pipeline,
    "html": html_pipeline,
    "plain_text": text_pipeline,
    "xlsx": spreadsheet_pipeline,
})

document = await router.run_file(file=file)

Parameters

ParameterDescription
pipelinesDictionary mapping protocol names to Pipeline instances
mime_registryOptional custom MIME registry for file-type detection
protocol_overridesOptional dictionary that maps a MIME type (for example, "text/html") to the protocol to use instead of the default

Override the default protocol

To send a file type to a different protocol, key the override by MIME type:

router = RoutedPipeline(
    pipelines={
        "ocr": ocr_pipeline,
        "html": html_pipeline,
        "plain_text": text_pipeline,
        "xlsx": spreadsheet_pipeline,
    },
    protocol_overrides={
        "text/html": "plain_text",  # Index raw HTML as text instead of converting it
    },
)

document = await router.run_file(file=file)
Components

Components

Each ingestion component has its own page with configuration options and examples: