Rate limiter pour API

Les APIs de LLM ont des limites de debit. OpenAI te limite a un certain nombre de requetes par minute (RPM) et de tokens par minute (TPM). Si tu depasses, tu recois une erreur 429 "Too Many Requests". En production, tu dois gerer ca toi-meme pour eviter de te faire bloquer.

Un rate limiter, c'est un compteur qui surveille combien d'appels tu as fait recemment. Si tu depasses la limite, il bloque l'appel au lieu de laisser passer.

On va simplifier le concept avec des timestamps fictifs. Plutot que d'utiliser le vrai temps (qui rendrait les tests non deterministes), on va passer le timestamp comme argument.

Tu vas creer une classe RateLimiter avec :

__init__(self, max_appels, fenetre) : max_appels est le nombre maximum d'appels autorises dans une fenetre de temps (en secondes).

autoriser(self, timestamp) : verifie si un appel est autorise au moment donne. D'abord, supprime tous les appels enregistres qui sont plus vieux que (timestamp - fenetre). Ensuite, si le nombre d'appels restants est inferieur a max_appels, enregistre le timestamp et renvoie True. Sinon, renvoie False.

appels_restants(self, timestamp) : renvoie le nombre d'appels encore disponibles dans la fenetre actuelle (sans oublier de nettoyer les anciens).

Exemple :

rl = RateLimiter(max_appels=3, fenetre=60)
rl.autoriser(0) renvoie True (1er appel)
rl.autoriser(10) renvoie True (2e appel)
rl.autoriser(20) renvoie True (3e appel)
rl.autoriser(30) renvoie False (limite atteinte)
rl.autoriser(61) renvoie True (le 1er appel est sorti de la fenetre)

Tests (4/5)

Appels dans la limite

rl = RateLimiter(max_appels=2, fenetre=60)
assert rl.autoriser(0) == True
assert rl.autoriser(10) == True

Appel refuse quand limite atteinte

rl = RateLimiter(max_appels=2, fenetre=60)
rl.autoriser(0)
rl.autoriser(10)
assert rl.autoriser(20) == False

Fenetre glissante libere des places

rl = RateLimiter(max_appels=1, fenetre=10)
rl.autoriser(0)
assert rl.autoriser(5) == False
assert rl.autoriser(11) == True

Appels restants corrects

rl = RateLimiter(max_appels=3, fenetre=60)
rl.autoriser(0)
rl.autoriser(10)
assert rl.appels_restants(20) == 1
assert rl.appels_restants(61) == 2

+ 0 tests cachés

Indices (3 disponibles)

Solution officielle

class RateLimiter:
    def __init__(self, max_appels, fenetre):
        self.max_appels = max_appels
        self.fenetre = fenetre
        self.historique = []

    def _nettoyer(self, timestamp):
        seuil = timestamp - self.fenetre
        self.historique = [t for t in self.historique if t > seuil]

    def autoriser(self, timestamp):
        self._nettoyer(timestamp)
        if len(self.historique) < self.max_appels:
            self.historique.append(timestamp)
            return True
        return False

    def appels_restants(self, timestamp):
        self._nettoyer(timestamp)
        return max(0, self.max_appels - len(self.historique))

← Retry avec modele de fallback Cache de reponses LLM →

solution.py

Bravo!

Rate limiter pour API

Tests (4/5)

Indices (3 disponibles)

Solution officielle