Introduction : Pourquoi Traiter vos Requêtes par Lots ?

Bonjour, je m'appelle Marie et je suis développeuse freelance depuis cinq ans. Quand j'ai découvert les APIs d'IA, je traitais mes demandes une par une, attendant parfois plusieurs minutes pour des batchs de 100 documents. Un jour, j'ai chronométré : 47 secondes pour 10 requêtes individuelles contre seulement 3.2 secondes avec le traitement par lots. Cette différence m'a fait comprendre l'importance cruciale de l'optimisation. Aujourd'hui, je vais vous partager tout ce que j'ai appris pour maîtriser le traitement批量请求 (batch processing) avec DeepSeek via HolySheep AI.

Comprendre les Bases : Qu'est-ce qu'une Requête par Lot ?

Imaginez que vous devez envoyer 100 lettres à la poste. Vous avez deux options :

Une requête par lot fonctionne exactement de la même manière. Au lieu d'envoyer 100 demandes séparées à l'API DeepSeek, vous les regroupez en une seule requête. Le résultat ? Une réduction du temps de traitement de 94% et une économie de 85%+ sur les coûts grâce aux tarifs compétitifs de HolySheep AI.

Configuration Initiale : Votre Premier Environnement

Installation des Outils Nécessaires

Ouvrez votre terminal (ou invite de commandes) et installez les dépendances requise :

pip install requests aiohttp asyncio

Ensuite, créez un fichier Python nommé batch_processor.py. Nous allons construire ensemble votre premier traitement par lots.

Votre Premier Script de Traitement par Lots

La Configuration de Base

import requests
import json
import time
from concurrent.futures import ThreadPoolExecutor, as_completed


Configuration HolySheep API

BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"  # Remplacez par votre clé

def traiter_requete_simple(texte):
    """Envoie une seule requête à DeepSeek V3.2"""
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": "deepseek-v3.2",
        "messages": [
            {"role": "user", "content": texte}
        ],
        "max_tokens": 500,
        "temperature": 0.7
    }
    
    response = requests.post(
        f"{BASE_URL}/chat/completions",
        headers=headers,
        json=payload
    )
    
    return response.json()


Test avec un seul message

resultat = traiter_requete_simple("Explique-moi les requêtes par lots en une phrase")
print(resultat)

Le Traitement Parallèle avec ThreadPoolExecutor

Pour les débutants, le traitement parallèle peut sembler complexe. Laissez-moi vous expliquer simplement : c'est comme avoir plusieurs caissiers qui travaillent simultanément. Voici le code complet :

import requests
import json
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"

def requete_deepseek(texte, index):
    """Traite une requête individuelle DeepSeek V3.2"""
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": "deepseek-v3.2",
        "messages": [{"role": "user", "content": texte}],
        "max_tokens": 300
    }
    
    debut = time.time()
    
    try:
        reponse = requests.post(
            f"{BASE_URL}/chat/completions",
            headers=headers,
            json=payload,
            timeout=30
        )
        latence = time.time() - debut
        
        return {
            "index": index,
            "status": "succès",
            "latence_ms": round(latence * 1000, 2),
            "donnees": reponse.json()
        }
    except Exception as e:
        return {
            "index": index,
            "status": "erreur",
            "erreur": str(e)
        }

def traitement_lot_parallel(texte_list, max_workers=10):
    """Traitement par lots avec parallélisation"""
    resultats = []
    
    debut_global = time.time()
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # Soumettre toutes les requêtes
        futures = {
            executor.submit(requete_deepseek, texte, i): i 
            for i, texte in enumerate(texte_list)
        }
        
        # Collecter les résultats au fur et à mesure
        for future in as_completed(futures):
            resultat = future.result()
            resultats.append(resultat)
            print(f"Requête {resultat['index']} terminée en {resultat.get('latence_ms', 'N/A')} ms")
    
    temps_total = time.time() - debut_global
    
    return {
        "resultats": resultats,
        "temps_total_secondes": round(temps_total, 2),
        "nb_requetes": len(texte_list)
    }


Exemple d'utilisation avec 20 prompts

prompts_test = [
    f"Rédige un titre accrocheur pour un article sur le sujet #{i}"
    for i in range(1, 21)
]

resultat = traitement_lot_parallel(prompts_test, max_workers=10)
print(f"\n=== RÉSUMÉ ===")
print(f"Nombre de requêtes : {resultat['nb_requetes']}")
print(f"Temps total : {resultat['temps_total_secondes']} secondes")
print(f"Moyenne par requête : {resultat['temps_total_secondes'] / resultat['nb_requetes']:.2f} secondes")

Optimisation Avancée : Le Batch Streaming

Pour les volumes importants (plus de 500 requêtes), je vous recommande le mode batch streaming qui traite les données en flux continu. Cette méthode est celle que j'utilise quotidiennement pour mes projets clients.

import aiohttp
import asyncio
import time
import json

BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"

class BatchProcessor:
    def __init__(self, api_key, batch_size=50, max_concurrent=20):
        self.api_key = api_key
        self.batch_size = batch_size
        self.max_concurrent = max_concurrent
        self.results = []
        
    async def requete_async(self, session, texte, index):
        """Requête asynchrone unique"""
        url = f"{BASE_URL}/chat/completions"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type

Ressources connexes
Articles connexes