Pipeline complet de bout en bout

Assembler les pieces

Les lecons precedentes ont couvert chaque etape isolement. Cette lecon les assemble dans un pipeline complet, de la preparation des donnees a l’exploitation des resultats. Vous repartirez avec un script fonctionnel que vous pourrez adapter a 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"
)

# --- Etape 1 : Preparer les donnees ---
documents = [
    {"id": f"doc-{i}", "text": f"Contenu du document {i}..."}
    for i in range(100)
]

# --- Etape 2 : Generer 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-3",
                "messages": [
                    {"role": "system", "content": "Resumez en 2 phrases."},
                    {"role": "user", "content": doc["text"]}
                ]
            }
        }
        f.write(json.dumps(request, ensure_ascii=False) + "\n")

print(f"{len(documents)} requetes ecrites dans batch_requests.jsonl")

# --- Etape 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 uploade : {file_id}")

# --- Etape 4 : Creer le batch et associer le fichier ---
batch = client.batches.create(input_file_id=file_id)
batch_id = batch.id
print(f"Batch cree : {batch_id}")

# --- Etape 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)  # Verifier toutes les 5 minutes

# --- Etape 6 : Recuperer les resultats ---
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)} resultats sauvegardes")
else:
    print(f"Le batch a echoue avec le statut : {status.status}")

Pipeline cURL pas a pas

Si vous preferez la ligne de commande, voici l’equivalent 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. Creer 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. Verifier le statut
curl -s "https://api.x.ai/v1/batches/$BATCH_ID" \
  -H "Authorization: Bearer $XAI_API_KEY" | jq '.status'

# 4. Recuperer les resultats
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 gerer trois types d’erreurs :

Erreurs de soumission : l’upload ou la creation du batch echoue. Implementez un retry avec backoff exponentiel (1s, 2s, 4s, 8s).

Erreurs de traitement : certaines requetes echouent dans le batch. Parcourez les resultats et identifiez les custom_id en erreur pour les resoumettre.

Timeout : si le batch reste en pending au-dela de 48 heures, considerez-le comme bloque et contactez le support xAI.

# Gestion des requetes echouees
failed_ids = []
for result in results:
    if result.status != "succeeded":
        failed_ids.append(result.custom_id)
        print(f"Echec : {result.custom_id} - {result.error}")

if failed_ids:
    print(f"{len(failed_ids)} requetes a resoumettre")

Optimisation des couts

Pour maximiser les economies :

• Groupez vos requetes : un gros batch est plus efficient que plusieurs petits
• Utilisez des modeles adaptes : pour de la classification simple, un modele plus leger suffit
• Limitez les tokens de sortie : specifiez max_tokens pour eviter les reponses trop longues
• Nettoyez vos entrees : retirez le contenu superflu de vos prompts pour reduire les tokens d’entree

Points cles a retenir

• Un pipeline complet couvre six etapes : preparation, JSONL, upload, creation, surveillance, recuperation
• Le polling toutes les 5 minutes est un bon compromis pour la surveillance
• Gerez les trois types d’erreurs : soumission, traitement, timeout
• Sauvegardez les resultats localement des leur disponibilite
• Groupez les requetes et optimisez les prompts pour maximiser les economies