Pipeline complet de bout en bout

Assembler les pièces

Les leçons précédentes ont couvert chaque étape isolément. Cette leçon les assemble dans un pipeline complet, de la préparation des données à l’exploitation des résultats. Vous repartirez avec un script fonctionnel que vous pourrez adapter à votre cas d’usage.

Pipeline Python complet

import openai
import json
import time

# Configuration
client = openai.OpenAI(
    api_key="votre-cle-xai",
    base_url="https://api.x.ai/v1"
)

# --- Étape 1 : Préparer les données ---
documents = [
    {"id": f"doc-{i}", "text": f"Contenu du document {i}..."}
    for i in range(100)
]

# --- Étape 2 : Générer le fichier JSONL ---
with open("batch_requests.jsonl", "w") as f:
    for doc in documents:
        request = {
            "custom_id": doc["id"],
            "method": "POST",
            "url": "/v1/chat/completions",
            "body": {
                "model": "grok-4.5",
                "messages": [
                    {"role": "system", "content": "Résumez en 2 phrases."},
                    {"role": "user", "content": doc["text"]}
                ]
            }
        }
        f.write(json.dumps(request, ensure_ascii=False) + "\n")

print(f"{len(documents)} requêtes écrites dans batch_requests.jsonl")

# --- Étape 3 : Uploader le fichier ---
with open("batch_requests.jsonl", "rb") as f:
    file_response = client.files.create(file=f, purpose="batch")

file_id = file_response.id
print(f"Fichier uploadé : {file_id}")

# --- Étape 4 : Créer le batch et associer le fichier ---
batch = client.batches.create(input_file_id=file_id)
batch_id = batch.id
print(f"Batch créé : {batch_id}")

# --- Étape 5 : Surveiller la progression ---
while True:
    status = client.batches.retrieve(batch_id)
    print(f"Statut : {status.status}")

    if status.status in ("succeeded", "failed", "cancelled"):
        break

    time.sleep(300)  # Vérifier toutes les 5 minutes

# --- Étape 6 : Récupérer les résultats ---
if status.status == "succeeded":
    results = client.batches.results(batch_id, page_size=100)

    output = {}
    for result in results:
        output[result.custom_id] = result.response

    with open("batch_results.json", "w") as f:
        json.dump(output, f, ensure_ascii=False, indent=2)

    print(f"{len(output)} résultats sauvegardés")
else:
    print(f"Le batch a échoué avec le statut : {status.status}")

Pipeline cURL pas à pas

Si vous préférez la ligne de commande, voici l’équivalent en cURL :

# 1. Uploader le fichier JSONL
FILE_ID=$(curl -s -X POST https://api.x.ai/v1/files \
  -H "Authorization: Bearer $XAI_API_KEY" \
  -F "file=@batch_requests.jsonl" \
  -F "purpose=batch" | jq -r '.id')

echo "Fichier : $FILE_ID"

# 2. Créer le batch
BATCH_ID=$(curl -s -X POST https://api.x.ai/v1/batches \
  -H "Authorization: Bearer $XAI_API_KEY" \
  -H "Content-Type: application/json" \
  -d "{\"input_file_id\": \"$FILE_ID\"}" | jq -r '.id')

echo "Batch : $BATCH_ID"

# 3. Vérifier le statut
curl -s "https://api.x.ai/v1/batches/$BATCH_ID" \
  -H "Authorization: Bearer $XAI_API_KEY" | jq '.status'

# 4. Récupérer les résultats
curl -s "https://api.x.ai/v1/batches/$BATCH_ID/results?page_size=100" \
  -H "Authorization: Bearer $XAI_API_KEY" > results.json

Gestion des erreurs en production

Un pipeline robuste doit gérer trois types d’erreurs :

Erreurs de soumission : l’upload ou la création du batch échoue. Implémentez un retry avec backoff exponentiel (1s, 2s, 4s, 8s).

Erreurs de traitement : certaines requêtes échouent dans le batch. Parcourez les résultats et identifiez les custom_id en erreur pour les resoumettre.

Timeout : si le batch reste en pending au-delà de 48 heures, considérez-le comme bloqué et contactez le support xAI.

# Gestion des requêtes échouées
failed_ids = []
for result in results:
    if result.status != "succeeded":
        failed_ids.append(result.custom_id)
        print(f"Échec : {result.custom_id} - {result.error}")

if failed_ids:
    print(f"{len(failed_ids)} requêtes à resoumettre")

Optimisation des coûts

Les économies du batch se cumulent sur quatre leviers, du plus structurel au plus fin. Le regroupement d’abord : un gros batch est plus efficient que plusieurs petits — moins d’allers-retours de gestion, un seul cycle de suivi, une seule récupération de résultats. Le choix du modèle ensuite, qui est le levier le plus puissant : pour de la classification simple ou de l’extraction courte, un modèle léger produit les mêmes résultats pour une fraction du prix — réserver le modèle phare aux tâches qui le méritent divise souvent la facture par plusieurs. Côté sortie, spécifier max_tokens évite de payer des réponses inutilement longues : pour une classification, quelques tokens suffisent, laisser le modèle libre en génère parfois des centaines. Et côté entrée, le nettoyage des prompts — retirer les instructions redondantes, les exemples superflus, le contexte non utilisé — se multiplie par le nombre de requêtes du lot : dix tokens économisés sur un prompt répété cinquante mille fois, c’est un demi-million de tokens en moins.

Points clés à retenir

• Un pipeline complet couvre six étapes : préparation, JSONL, upload, création, surveillance, récupération
• Le polling toutes les 5 minutes est un bon compromis pour la surveillance
• Gérez les trois types d’erreurs : soumission, traitement, timeout
• Sauvegardez les résultats localement dès leur disponibilité
• Groupez les requêtes et optimisez les prompts pour maximiser les économies