Index de recherche

Les backends de stockage conservent les fragments traités et permettent une recherche efficace dans votre collection de documents. Les bases vectorielles assurent la recherche sémantique en stockant les embeddings de fragments et en retrouvant les vecteurs similaires.

Utilisez Vespa comme base vectorielle dans les pipelines d’ingestion et de recherche du Search Toolkit. Vespa propose une recherche vectorielle évolutive, la gestion des schémas, le ranking et la mise en cluster prête pour la production.

Fonctionnalités :

• Recherche vectorielle évolutive optimisée nativement avec indexation HNSW
• Recherche textuelle BM25 pour le classement hybride
• Classement multi-phases avec fonctions de scoring personnalisées
• Prêt pour la production grâce au clustering et à la réplication

Installation :

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

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

Prérequis

Avant d’utiliser Vespa comme index de recherche :

1. Définir votre application — Créez les schémas avec des champs et des profils de ranking via les migrations Python
2. Déployer Vespa — Exécutez mistral-vespa migrate pour déployer votre application
3. Obtenir l’endpoint — Notez l’endpoint de requête Vespa (ex. : http://localhost:8080)

Consultez Gérer et déployer Vespa pour les instructions complètes de configuration.

Guide de démarrage

Configurez Vespa comme base vectorielle :

from mistralai.search.toolkit.plugins.vespa import VespaClientConfig
from vespa_app import app

collection_name = "my_collection"
config = VespaClientConfig(
    endpoint="http://localhost:8080",
)

vector_store = app.get_search_index(config, collection_name=collection_name)

from mistralai.search.toolkit.plugins.vespa import VespaClientConfig
from vespa_app import app

collection_name = "my_collection"
config = VespaClientConfig(
    endpoint="http://localhost:8080",
)

vector_store = app.get_search_index(config, collection_name=collection_name)

Utilisation dans un pipeline d’ingestion :

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

pipeline = Pipeline(
    loader=loader,
    extractor=extractor,
    text_splitter=splitter,
    embedder=embedder,
    stores=vector_store,
)

num_chunks = await pipeline.run(documents=["doc1.pdf", "doc2.pdf"])

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

pipeline = Pipeline(
    loader=loader,
    extractor=extractor,
    text_splitter=splitter,
    embedder=embedder,
    stores=vector_store,
)

num_chunks = await pipeline.run(documents=["doc1.pdf", "doc2.pdf"])

Utilisation dans un pipeline de recherche :

from mistralai.search.toolkit.retrieval import QueryEngine
from mistralai.search.toolkit.retrieval.retrievers import VectorRetriever

embedder = MistralEmbedder(client=mistral_client)
query_engine = QueryEngine(
    retriever=VectorRetriever(client=vector_store, embedder=embedder),
)

result = await query_engine.search(query="What is RAG?", top_k=5)

from mistralai.search.toolkit.retrieval import QueryEngine
from mistralai.search.toolkit.retrieval.retrievers import VectorRetriever

embedder = MistralEmbedder(client=mistral_client)
query_engine = QueryEngine(
    retriever=VectorRetriever(client=vector_store, embedder=embedder),
)

result = await query_engine.search(query="What is RAG?", top_k=5)

Pour une configuration complète, la conception des schémas et les opérations, consultez Gérer et déployer Vespa.

Implémentez vos propres backends de stockage en créant l’interface VectorStoreClient :

from mistralai.search.toolkit.indices import VectorStoreClient
from mistralai.search.toolkit.document import Document
from mistralai.search.toolkit.indices import VectorSearchQuery, SearchResult

class MyVectorStore(VectorStoreClient):
    async def create_index(self) -> bool:
        pass

    async def upsert(self, document: Document) -> bool:
        pass

    async def search(self, query: VectorSearchQuery) -> list[SearchResult]:
        pass

    async def delete_document(self, doc_id: str) -> None:
        pass

from mistralai.search.toolkit.indices import VectorStoreClient
from mistralai.search.toolkit.document import Document
from mistralai.search.toolkit.indices import VectorSearchQuery, SearchResult

class MyVectorStore(VectorStoreClient):
    async def create_index(self) -> bool:
        pass

    async def upsert(self, document: Document) -> bool:
        pass

    async def search(self, query: VectorSearchQuery) -> list[SearchResult]:
        pass

    async def delete_document(self, doc_id: str) -> None:
        pass