Technical Tutorials

Pour les développeurs travaillant avec les grands modèles de langage (LLM), le coût et la latence sont souvent les principaux goulets d'étranglement lors du traitement de grands jeux de données. Que vous génériez des descriptions de produits pour un site e-commerce avec des millions d'articles, résumiez des milliers de tickets de support ou effectuiez une analyse de sentiments sur des journaux historiques, effectuer des appels API individuels pour chaque élément est inefficace et coûteux.

Entrez dans l'univers de l'API Batch d'OpenAI. Cette fonctionnalité vous permet d'envoyer de grands volumes de requêtes non urgentes à OpenAI avec une réduction significative. En regroupant les requêtes dans un seul travail traité de manière asynchrone, vous pouvez réduire vos coûts de jetons d'entrée et de sortie de jusqu'à 50 % par rapport aux appels API en temps réel. Dans ce guide, nous explorerons comment mettre en œuvre efficacement l'API Batch pour le traitement de données à grande échelle.

Pourquoi utiliser l'API Batch ?

L'avantage principal de l'API Batch est l'efficacité des coûts. OpenAI offre une réduction de 50 % sur tous les jetons utilisés dans les requêtes par lot. Cependant, cela implique des compromis :

Traitement asynchrone : Vous ne pouvez pas vous attendre à des réponses immédiates. Les travaux sont généralement terminés dans les 24 heures.
Absence d'interactivité en temps réel : Elle n'est pas adaptée aux chatbots ou aux applications nécessitant un retour utilisateur immédiat.
Taille de lot fixe : Vous devez préparer toutes les requêtes à l'avance avant la soumission.

Si votre cas d'utilisation implique des analyses hors ligne, la génération de contenu ou l'étiquetage de données, l'API Batch est un choix idéal.

Étape 1 : Préparez votre fichier JSONL

La première étape de tout travail par lot est la création d'un fichier JSON Lines (JSONL). Chaque ligne de ce fichier représente une seule requête API. La structure doit correspondre au format standard de requête de l'API OpenAI, y compris un `custom_id` unique pour le suivi.

import json

def create_batch_file(data_points, output_filename="batch_input.jsonl"):
    """
    Génère un fichier JSONL pour l'API Batch d'OpenAI.
    
    Args:
        data_points: Liste de chaînes de caractères à traiter.
        output_filename: Nom du fichier JSONL de sortie.
    """
    with open(output_filename, 'w') as f:
        for i, text in enumerate(data_points):
            request = {
                "custom_id": f"request-{i}",
                "method": "POST",
                "url": "/v1/chat/completions",
                "body": {
                    "model": "gpt-4o",
                    "messages": [
                        {
                            "role": "user",
                            "content": f"Summarize the following text in one sentence: {text}"
                        }
                    ],
                    "max_tokens": 100
                }
            }
            f.write(json.dumps(request) + '\n')
    print(f"Batch file '{output_filename}' created successfully.")

# Exemple d'utilisation
sample_data = [
    "The quick brown fox jumps over the lazy dog.",
    "Python is a high-level programming language known for its readability.",
    "Machine learning is a subset of artificial intelligence."
]

create_batch_file(sample_data)

Étape 2 : Téléversez le fichier et créez le travail par lot

Une fois le fichier JSONL prêt, vous devez le téléverser vers OpenAI et initier le travail par lot à l'aide du SDK Python.

from openai import OpenAI
import time

client = OpenAI()  # La clé API est lue depuis la variable d'environnement

# 1. Téléverser le fichier
file_response = client.files.create(
    file=open("batch_input.jsonl", "rb"),
    purpose="batch"
)
file_id = file_response.id
print(f"File uploaded with ID: {file_id}")

# 2. Créer le travail par lot
batch = client.batches.create(
    input_file_id=file_id,
    endpoint="/v1/chat/completions",
    completion_window="24h"
)
batch_id = batch.id
print(f"Batch job created with ID: {batch_id}")

Étape 3 : Interroger pour la complétion

Étant donné que le travail par lot est asynchrone, vous devez interroger le statut jusqu'à ce qu'il soit terminé. Selon la taille du travail, cela peut prendre plusieurs heures. Pour les systèmes de production, il est recommandé de mettre en œuvre un travail cron ou un système de notification basé sur des webhooks plutôt que de bloquer dans un script.

def poll_batch_status(batch_id, client, interval=60):
    """
    Interroge le statut du travail par lot jusqu'à ce qu'il soit terminé ou échoué.
    """
    while True:
        batch = client.batches.retrieve(batch_id=batch_id)
        status = batch.status
        
        print(f"Current status: {status}")
        
        if status in ["completed", "failed", "expired", "cancelled"]:
            return batch
        
        time.sleep(interval)

# Attendre la complétion
final_batch = poll_batch_status(batch_id, client, interval=300)  # Vérifier toutes les 5 minutes

Étape 4 : Récupérer et traiter les résultats

Une fois que le lot est marqué comme `completed`, vous pouvez télécharger le fichier de sortie. Ce fichier contient les réponses correspondant à chaque `custom_id` du fichier d'entrée.

def download_and_process_results(batch_id, client):
    """
    Télécharge le fichier de sortie du lot et analyse les résultats.
    """
    # Récupérer l'ID du fichier de sortie
    output_file_id = client.batches.retrieve(batch_id).output_file_id
    
    if output_file_id is None:
        raise ValueError("No output file found. Job may have failed.")
    
    # Récupérer le contenu du fichier
    file_content = client.files.content(output_file_id)
    
    # Analyser le contenu JSONL
    results = []
    for line in file_content.splitlines():
        if line.strip():
            data = json.loads(line)
            # Extraire le contenu de la réponse
            response_content = data.get('response', {}).get('body', {}).get('choices', [{}])[0].get('message', {}).get('content', '')
            results.append({
                'id': data.get('custom_id'),
                'summary': response_content
            })
            
    return results

# Traiter les résultats
results = download_and_process_results(batch_id, client)
for r in results:
    print(f"{r['id']}: {r['summary']}")

Meilleures pratiques pour le traitement par lot

Gérer les erreurs avec élégance : Certaines requêtes peuvent échouer en raison de violations des politiques de contenu ou de problèmes de formatage. Vérifiez toujours le champ `error` dans les lignes JSONL de sortie.
Diviser les grands jeux de données : Bien qu'il n'y ait pas de limite stricte, les fichiers extrêmement grands peuvent être plus difficiles à déboguer. Envisagez de diviser des millions de lignes en blocs de 10 000 à 50 000 pour une gestion plus facile.
Contrôler les versions de vos invites : Étant donné que les travaux par lot prennent du temps à s'exécuter, assurez-vous que votre modèle d'invite est stable. Si vous devez modifier l'invite, vous devez créer un nouveau fichier et un nouveau travail par lot.
Surveiller les coûts : Utilisez le Tableau de bord OpenAI pour suivre l'utilisation des jetons et les coûts de chaque travail par lot afin de vous assurer de maximiser vos économies.

Conclusion

L'API Batch d'OpenAI est un outil puissant pour les développeurs souhaitant faire évoluer leurs applications IA sans faire évoluer leurs coûts de manière linéaire. En exploitant le traitement asynchrone, vous pouvez relever des défis de données à grande échelle qui seraient prohibitifs en termes de coût ou de lenteur avec des appels API en temps réel. À mesure que les applications LLM continuent d'évoluer, la maîtrise du traitement par lot sera une compétence essentielle pour construire des systèmes IA efficaces et économiques.

Commencez petit avec un jeu de données de test, vérifiez vos résultats, puis passez à l'échelle supérieure. La réduction de 50 % rend l'API Batch une partie indispensable de toute pile de développement LLM sérieuse.