Home
Studio

Guide de démarrage : construire un pipeline RAG

Construisez un pipeline RAG : ingérez des documents texte dans une base vectorielle Vespa locale, puis interrogez-les.

Prérequis

Prérequis

  • Python 3.12 à 3.14
  • Docker (pour exécuter Vespa en local)
  • Une clé API Mistral provenant de console.mistral.ai
Installer

Installer

Installez Search Toolkit avec le plugin Vespa à l'aide de uv :

uv add "mistralai-search-toolkit[vespa]"
Configurer Vespa

Configurer Vespa

Démarrez une instance locale de Vespa avec Docker :

docker run --detach \
  --name vespa \
  --hostname vespa-container \
  --publish 8080:8080 \
  --publish 19071:19071 \
  vespaengine/vespa

Attendez que Vespa soit opérationnel :

curl --retry 10 --retry-delay 3 --retry-all-errors \
  http://localhost:19071/state/v1/health

Définissez votre clé API Mistral :

export MISTRAL_API_KEY=your-api-key
Définir votre schéma

Définir votre schéma

Créez une migration pour décrire la structure de vos documents :

mistral-vespa generate-migration \
  --app-dir ./vespa_app \
  initial_schema

La commande crée le package vespa_app et un fichier horodaté dans vespa_app/migrations/. Complétez le fichier généré :

from mistralai.search.toolkit.embedding import MistralEmbeddingPreset
from mistralai.search.toolkit.plugins.vespa.app.schemas.app import FieldDefinition, IndexingMode, SearchMode
from mistralai.search.toolkit.plugins.vespa.migration import VespaMigration, create_schema, set_app_name


class InitialSchema(VespaMigration):
    def migrate(self) -> None:
        set_app_name("myquickstart")
        create_schema(
            name="quickstart_collection",
            mode=SearchMode.INDEX,
            embedding_model=MistralEmbeddingPreset.MISTRAL_EMBED_DIM_1024,
            indexing_mode=IndexingMode.DOCUMENT_PER_CHUNK,
            # The standard chunk fields (content, embedding, identity, metadata) are
            # added automatically. `fields` only declares your extra fields.
            fields=[
                FieldDefinition.TextField(name="title"),
            ],
        )
i
Information

Restrictions sur le nom de l'application : le nom passé à set_app_name() ne doit contenir que des lettres minuscules (a-z). Les chiffres, tirets bas, tirets et autres caractères spéciaux ne sont pas autorisés.

Pour plus de détails sur la gestion et le déploiement d'applications Vespa, consultez Vespa.

Déployer à partir des migrations

Déployer à partir des migrations

Déployez le schéma sur votre instance locale de Vespa :

mistral-vespa migrate \
  --app-dir ./vespa_app \
  --config-server http://localhost:19071 \
  --query-port 8080

Cette commande construit le package applicatif en mémoire à partir de vos migrations et le déploie sur le serveur de configuration. Avant d'ingérer, attendez que curl http://localhost:8080/ApplicationStatus renvoie un code HTTP 200.

Ingérer des documents

Ingérer des documents

Créez un pipeline qui charge des fichiers, en extrait le texte, le découpe en fragments, en génère les représentations vectorielles et les indexe dans Vespa. Enregistrez le code sous le nom ingest.py, à côté du répertoire vespa_app :

import asyncio
import os
from pathlib import Path

from mistralai.client import Mistral
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.plugins.vespa import VespaClientConfig
from vespa_app import app


async def main():
    mistral_client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

    # Configure Vespa
    config = VespaClientConfig(
        endpoint="http://localhost:8080",
    )
    vector_store = app.get_search_index(config, collection_name="quickstart_collection")

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

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


asyncio.run(main())

Ajoutez deux fichiers texte, doc1.txt et doc2.txt, dans le même répertoire, puis exécutez uv run python ingest.py. Le script affiche Documents ingested!.

Le pipeline enchaîne cinq étapes :

  1. FilesystemFileLoader lit les octets bruts des fichiers depuis le disque.
  2. PlainTextExtractor extrait le contenu textuel du fichier. Pour les PDF, utilisez plutôt MistralOCRExtractor.
  3. CharacterTextSplitter découpe le texte en fragments de 500 caractères.
  4. MistralEmbedder génère une représentation vectorielle pour chaque fragment.
  5. vector_store (Vespa) indexe chaque fragment en stockant la représentation vectorielle pour la recherche vectorielle.
Rechercher

Rechercher

Interrogez les documents indexés avec une recherche vectorielle :

import asyncio
import os

from mistralai.client import Mistral
from mistralai.search.toolkit.embedding import MistralEmbedder
from mistralai.search.toolkit.plugins.vespa import VespaClientConfig
from mistralai.search.toolkit.retrieval import QueryEngine
from mistralai.search.toolkit.retrieval.retrievers import VectorRetriever
from vespa_app import app


async def main():
    mistral_client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
    embedder = MistralEmbedder(client=mistral_client, model_name="mistral-embed")

    # Configure Vespa for search
    config = VespaClientConfig(
        endpoint="http://localhost:8080",
    )
    vector_store = app.get_search_index(config, collection_name="quickstart_collection")

    # Build query engine
    query_engine = QueryEngine(
        retriever=[VectorRetriever(client=vector_store, embedder=embedder)],
    )

    # Search
    result = await query_engine.search(
        query="What is RAG?",
        top_k=5,
        include_metadata=True,
        include_content=True,
    )

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


asyncio.run(main())
Ingérer des PDF avec OCR

Ingérer des PDF avec OCR

Pour les documents PDF, remplacez PlainTextExtractor par MistralOCRExtractor et utilisez MarkdownTextSplitter pour un découpage en fragments qui tient compte de la structure. MarkdownTextSplitter nécessite l'extra text-splitter-langchain. Conservez model_name="mistral-embed" pour que les vecteurs correspondent au schéma à 1024 dimensions :

from mistralai.search.toolkit.ingestion.extractors import MistralOCRExtractor
from mistralai.search.toolkit.ingestion.text_splitters import (
    MarkdownTextSplitter,
    MarkdownTextSplitterConfig,
)

pipeline = Pipeline(
    loader=FilesystemFileLoader(),
    extractor=MistralOCRExtractor(client=mistral_client),
    text_splitter=MarkdownTextSplitter(
        MarkdownTextSplitterConfig(chunk_size=5048, chunk_overlap=50)
    ),
    embedder=MistralEmbedder(client=mistral_client, model_name="mistral-embed"),
    stores=vector_store,
)