Home
Studio

Quickstart: build a RAG pipeline

In this quickstart, you create a Mistral Search Index, index two text files with a Search Toolkit pipeline, and search them by meaning. It takes about 10 minutes, and you don't need to run a search engine yourself. For how the toolkit's pieces fit together, see Search Toolkit.

Before you begin

Before you begin

  • A Mistral API key, exported as MISTRAL_API_KEY. To create one, see Create your first API key.
  • A workspace with search indexes enabled, and the create_search_indexes permission in that workspace.
  • Python 3.12, 3.13, or 3.14, and uv.
Set up your environment

Set up your environment

The Mistral Search Index client ships with the core package. Create a project and install the Search Toolkit:

uv init rag-quickstart --python 3.12 && cd rag-quickstart
uv add mistralai-search-toolkit
export MISTRAL_API_KEY="your-api-key"
1. Create the index

1. Create the index

A collection schema describes your documents and the embedding model that the index uses. Create create_index.py:

import asyncio

from mistralai.search.toolkit.document import Document
from mistralai.search.toolkit.embedding import MistralEmbeddingPreset
from mistralai.search.toolkit.mistral_search_index import MistralSearchIndexService
from mistralai.search.toolkit.mistral_search_index.models import MistralSearchIndexStatus
from mistralai.search.toolkit.mistral_search_index.schema import MistralSearchIndexCollectionSchema

INDEX = "rag_quickstart"


async def main():
    schema = MistralSearchIndexCollectionSchema.from_document_type(
        collection_name=INDEX,
        document_type=Document,
        embedding_model=MistralEmbeddingPreset.MISTRAL_EMBED_DIM_1024,
    )
    service = MistralSearchIndexService()
    info = await service.create_index(schema)

    # Provisioning is asynchronous: wait until the index is ready.
    while info.status == MistralSearchIndexStatus.PROVISIONING:
        await asyncio.sleep(1)
        info = await service.get_index_info(INDEX)
    if info.status != MistralSearchIndexStatus.READY:
        raise RuntimeError(f"provisioning failed: {info.status_message}")
    print("Index ready!")


asyncio.run(main())

The schema stores each chunk with a 1024-dimension mistral-embed vector. The base Document model has no custom fields. To add fields that you can filter on, see Declare a collection.

Run it:

uv run python create_index.py
Index ready!
2. Index two documents

2. Index two documents

Create two short text files to index:

cat > doc1.txt <<'EOF'
Retrieval-augmented generation (RAG) answers a question in two steps. First, it retrieves the passages that are most relevant to the question from your documents. Then, it gives those passages to a language model, which writes an answer grounded in them.
EOF
cat > doc2.txt <<'EOF'
A search index stores document chunks, their text, and their vector embeddings, and runs keyword, vector, and hybrid queries over them.
EOF

Create ingest.py. The pipeline reads each file, extracts its text, splits it into chunks of 500 characters, embeds each chunk with mistral-embed, and indexes the chunks:

import asyncio
from pathlib import Path

from mistralai.search.toolkit.embedding import MistralEmbedder
from mistralai.search.toolkit.ingestion.extractors import PlainTextExtractor
from mistralai.search.toolkit.ingestion.loaders import FilesystemFileLoader
from mistralai.search.toolkit.ingestion.pipelines import Pipeline
from mistralai.search.toolkit.ingestion.text_splitters import CharacterTextSplitter
from mistralai.search.toolkit.mistral_search_index import MistralSearchIndexService


async def main():
    vector_store = await MistralSearchIndexService().get_search_index("rag_quickstart")

    pipeline = Pipeline(
        loader=FilesystemFileLoader(),
        extractor=PlainTextExtractor(),
        text_splitter=CharacterTextSplitter(chunk_size=500),
        embedder=MistralEmbedder(model_name="mistral-embed"),
        stores=vector_store,
    )

    await pipeline.run(documents=[Path("doc1.txt"), Path("doc2.txt")])
    print("Documents ingested!")


asyncio.run(main())

Run it:

uv run python ingest.py
Documents ingested!
3. Search

3. Search

Create search.py. The script embeds the question with the same model, then runs a hybrid search that matches both the meaning and the keywords of the question:

import asyncio

from mistralai.search.toolkit.embedding import MistralEmbedder
from mistralai.search.toolkit.mistral_search_index import MistralSearchIndexService
from mistralai.search.toolkit.search import VectorSearchQuery


async def main():
    index = await MistralSearchIndexService().get_search_index("rag_quickstart")
    embedder = MistralEmbedder(model_name="mistral-embed")

    question = "What is RAG?"
    results = await index.search(
        VectorSearchQuery(
            query=question,
            embedding=await embedder.embed_query(question),
            top_k=5,
        )
    )

    for i, r in enumerate(results, 1):
        print(f"{i}. [Score: {r.score:.3f}] {r.chunk.content[:200]}...")


asyncio.run(main())

Run it:

uv run python search.py
You should see

You should see

A numbered list of chunks, each with its score, the best match first. The first result is the chunk from doc1.txt, which explains retrieval-augmented generation; the chunk from doc2.txt follows with a lower score.

If a script fails with MISTRAL_API_KEY is not set, export the key in the shell that runs it. If create_index.py fails with 403, the workspace of your API key either doesn't have search indexes enabled or lacks the create_search_indexes permission.

Clean up

Clean up

Delete the index and the documents that it holds:

uv run python -c 'import asyncio; from mistralai.search.toolkit.mistral_search_index import MistralSearchIndexService; asyncio.run(MistralSearchIndexService().delete_index("rag_quickstart"))'
Going further

Going further

Index PDFs with Mistral OCR
Improve retrieval
Mistral Search Index
Run your own search engine with Vespa
Agentic Search