Chargeurs de fichiers | Mistral Docs

Les chargeurs de fichiers chargent des fichiers depuis différentes sources dans des objets File pouvant être traités par des extracteurs de documents.

Chargeurs de fichiers disponibles

Chargeur	Source
Chargeur de fichiers du système de fichiers	Système de fichiers local
Chargeur de fichiers AWS S3	AWS S3 (et compatibles S3 : MinIO, Ceph)
Chargeur de fichiers Google Cloud Storage	Google Cloud Storage
Chargeur de fichiers Azure Blob Storage	Azure Blob Storage
Chargeurs personnalisés	Toute source

Chargeur de fichiers du système de fichiers

FilesystemFileLoader charge des fichiers depuis le système de fichiers local. Il accepte un paramètre root qui restreint l’accès : tout chemin se résolvant en dehors de root est rejeté, empêchant ainsi les attaques par traversée de répertoire.

Installation : Bibliothèque principale (aucune dépendance supplémentaire requise)

Exemple :

from pathlib import Path
from mistralai.search.toolkit.ingestion.loaders import FilesystemFileLoader

loader = FilesystemFileLoader(root="/data/documents")
file = await loader.load_file(Path("report.pdf"))

from pathlib import Path
from mistralai.search.toolkit.ingestion.loaders import FilesystemFileLoader

loader = FilesystemFileLoader(root="/data/documents")
file = await loader.load_file(Path("report.pdf"))

Limitation de la taille des fichiers :

Utilisez max_file_size pour rejeter les fichiers dépassant une taille donnée en octets. La vérification est effectuée via une recherche de métadonnées avant le chargement ; les fichiers trop volumineux ne sont donc jamais chargés en mémoire :

from mistralai.search.toolkit.ingestion.loaders import FilesystemFileLoader

loader = FilesystemFileLoader(
    root="/data/documents",
    max_file_size=50 * 1024 * 1024,  # 50 MiB
)
file = await loader.load_file("report.pdf")

from mistralai.search.toolkit.ingestion.loaders import FilesystemFileLoader

loader = FilesystemFileLoader(
    root="/data/documents",
    max_file_size=50 * 1024 * 1024,  # 50 MiB
)
file = await loader.load_file("report.pdf")

Une exception FileSizeLimitExceededException est levée lorsqu’un fichier dépasse la limite.

Paramètres :

Paramètre	Type	Valeur par défaut	Description
`root`	`Path \| str`	`"/"`	Répertoire racine — les chemins se résolvant en dehors de ce répertoire sont rejetés
`max_file_size`	`int \| None`	`None`	Taille maximale du fichier en octets. Les fichiers dépassant cette limite sont rejetés avant le chargement. `None` signifie aucune limite.

Sécurité :

Le paramètre root empêche les attaques par traversée de répertoire. Tout chemin se résolvant en dehors du répertoire racine est rejeté avant toute tentative d’ouverture du fichier.

Chargeur de fichiers AWS S3

Charge des fichiers depuis des buckets AWS S3 (ou des services compatibles S3 comme MinIO ou Ceph).

Installation :

uv add "mistralai-search-toolkit-storage-s3"

uv add "mistralai-search-toolkit-storage-s3"

Exemple :

from mistralai.search.toolkit.ingestion.loaders import FileLoader
from mistralai.search.toolkit.plugins.storage.s3 import S3BlobStorage

loader = FileLoader(
    lambda: S3BlobStorage(
        bucket_name="my-bucket",
        region_name="us-east-1",
        endpoint_url="http://localhost:9000",  # optional for MinIO / other S3-compatible backends
    )
)

file = await loader.load_file("reports/example.pdf")

from mistralai.search.toolkit.ingestion.loaders import FileLoader
from mistralai.search.toolkit.plugins.storage.s3 import S3BlobStorage

loader = FileLoader(
    lambda: S3BlobStorage(
        bucket_name="my-bucket",
        region_name="us-east-1",
        endpoint_url="http://localhost:9000",  # optional for MinIO / other S3-compatible backends
    )
)

file = await loader.load_file("reports/example.pdf")

Paramètres (S3BlobStorage) :

Paramètre	Type	Valeur par défaut	Description
`bucket_name`	`str`	(requis)	Nom du bucket S3
`region_name`	`str`	`"us-east-1"`	Région AWS
`endpoint_url`	`str \| None`	`None`	URL de l’endpoint personnalisé (pour MinIO, Ceph ou d’autres services compatibles S3)

Authentification :

Utilise la chaîne de acredentials AWS par défaut :

Rôle IAM (recommandé pour la production)
Variables d’environnement : AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY
Fichier d’accréditation AWS : ~/.aws/credentials

Chargeur de fichiers Azure Blob Storage

Charge des fichiers depuis Azure Blob Storage.

Installation :

uv add "mistralai-search-toolkit-storage-azure"

uv add "mistralai-search-toolkit-storage-azure"

Exemple :

from mistralai.search.toolkit.ingestion.loaders import FileLoader
from mistralai.search.toolkit.plugins.storage.azure import AzureBlobStorage

loader = FileLoader(
    lambda: AzureBlobStorage(
        container_name="my-container",
        account_url="https://myaccount.blob.core.windows.net",
        use_workload_identity=True,  # or pass azure_connection_string=... for connection-string auth
    )
)

file = await loader.load_file("reports/example.pdf")

from mistralai.search.toolkit.ingestion.loaders import FileLoader
from mistralai.search.toolkit.plugins.storage.azure import AzureBlobStorage

loader = FileLoader(
    lambda: AzureBlobStorage(
        container_name="my-container",
        account_url="https://myaccount.blob.core.windows.net",
        use_workload_identity=True,  # or pass azure_connection_string=... for connection-string auth
    )
)

file = await loader.load_file("reports/example.pdf")

Paramètres (AzureBlobStorage) :

Paramètre	Type	Valeur par défaut	Description
`container_name`	`str`	(requis)	Nom du conteneur Azure Blob Storage
`account_url`	`str`	(requis)	URL du compte de stockage Azure (par ex. `https://myaccount.blob.core.windows.net`)
`use_workload_identity`	`bool`	`False`	Utiliser workload identity pour l’authentification (recommandé pour les VM/Functions Azure)
`azure_connection_string`	`str \| None`	`None`	Chaîne de connexion pour l’authentification (alternative à workload identity)

Authentification :

Workload Identity (recommandé pour les VM/Functions Azure)
Chaîne de connexion : Transmettre via le paramètre azure_connection_string ou définir la variable d’environnement AZURE_STORAGE_CONNECTION_STRING
Jeton SAS : Accès temporaire (configurable via les identifiants du compte)

Chargeur de fichiers Google Cloud Storage

Charge des fichiers depuis Google Cloud Storage (GCS).

Installation :

uv add "mistralai-search-toolkit-storage-gcs"

uv add "mistralai-search-toolkit-storage-gcs"

Exemple :

from mistralai.search.toolkit.ingestion.loaders import FileLoader
from mistralai.search.toolkit.plugins.storage.gcs import GCSBlobStorage

loader = FileLoader(
    lambda: GCSBlobStorage(
        bucket_name="my-bucket",
        project="my-gcp-project",
        service_account_file="/path/to/service-account.json",  # optional; falls back to ADC
    )
)

file = await loader.load_file("reports/example.pdf")

from mistralai.search.toolkit.ingestion.loaders import FileLoader
from mistralai.search.toolkit.plugins.storage.gcs import GCSBlobStorage

loader = FileLoader(
    lambda: GCSBlobStorage(
        bucket_name="my-bucket",
        project="my-gcp-project",
        service_account_file="/path/to/service-account.json",  # optional; falls back to ADC
    )
)

file = await loader.load_file("reports/example.pdf")

Paramètres (GCSBlobStorage) :

Paramètre	Type	Valeur par défaut	Description
`bucket_name`	`str`	(requis)	Nom du bucket GCS
`project`	`str`	(requis)	ID de projet GCP
`service_account_file`	`str \| None`	`None`	Chemin vers le fichier JSON du compte de service. Si non fourni, utilise les Application Default Credentials (ADC)

Authentification :

Application Default Credentials (ADC) (recommandé) — utilise les identifiants issu de l’environnement, du service de métadonnées ou de l’outil CLI gcloud
Fichier de compte de service : Transmettre via le paramètre service_account_file ou définir la variable d’environnement GOOGLE_APPLICATION_CREDENTIALS

Chargement par lots

Chargez plusieurs fichiers avec un contrôle de simultanéité :

import asyncio
from mistralai.search.toolkit.ingestion.loaders import FileLoader

async def load_files_batch(
    loader: FileLoader,
    paths: list[str],
    max_concurrent: int = 10,
) -> list:
    """Load multiple files concurrently with semaphore."""
    semaphore = asyncio.Semaphore(max_concurrent)

    async def load_with_semaphore(path: str):
        async with semaphore:
            try:
                return await loader.load_file(path)
            except Exception as e:
                print(f"Failed to load {path}: {e}")
                return None

    results = await asyncio.gather(
        *[load_with_semaphore(p) for p in paths],
        return_exceptions=False,
    )
    return [f for f in results if f is not None]

# Usage
loader = FileLoader(lambda: S3BlobStorage(bucket_name="my-bucket"))
files = await load_files_batch(
    loader,
    ["doc1.pdf", "doc2.pdf", "doc3.pdf"],
    max_concurrent=5,
)

import asyncio
from mistralai.search.toolkit.ingestion.loaders import FileLoader

async def load_files_batch(
    loader: FileLoader,
    paths: list[str],
    max_concurrent: int = 10,
) -> list:
    """Load multiple files concurrently with semaphore."""
    semaphore = asyncio.Semaphore(max_concurrent)

    async def load_with_semaphore(path: str):
        async with semaphore:
            try:
                return await loader.load_file(path)
            except Exception as e:
                print(f"Failed to load {path}: {e}")
                return None

    results = await asyncio.gather(
        *[load_with_semaphore(p) for p in paths],
        return_exceptions=False,
    )
    return [f for f in results if f is not None]

# Usage
loader = FileLoader(lambda: S3BlobStorage(bucket_name="my-bucket"))
files = await load_files_batch(
    loader,
    ["doc1.pdf", "doc2.pdf", "doc3.pdf"],
    max_concurrent=5,
)

Chargeurs personnalisés

Implémentez le protocole FileLoader pour des sources personnalisées non couvertes ci-dessus :

from pathlib import Path
from mistralai.search.toolkit.ingestion.loaders import FileLoader
from mistralai.search.toolkit.ingestion import File

class CustomFileLoader(FileLoader):
    """Load files from a custom source."""

    async def load_file(self, path: Path | str) -> File:
        # Implement your custom loading logic
        content = await self._fetch_from_custom_source(str(path))
        filename = Path(str(path)).name

        return File(
            path=str(path),
            name=filename,
            raw=content,
        )

    async def _fetch_from_custom_source(self, path: str) -> bytes:
        # Your implementation here
        ...

from pathlib import Path
from mistralai.search.toolkit.ingestion.loaders import FileLoader
from mistralai.search.toolkit.ingestion import File

class CustomFileLoader(FileLoader):
    """Load files from a custom source."""

    async def load_file(self, path: Path | str) -> File:
        # Implement your custom loading logic
        content = await self._fetch_from_custom_source(str(path))
        filename = Path(str(path)).name

        return File(
            path=str(path),
            name=filename,
            raw=content,
        )

    async def _fetch_from_custom_source(self, path: str) -> bytes:
        # Your implementation here
        ...