Relais API LLM

Relais API LLM : Centraliser Claude, OpenAI et Gemini

Non classé

Référence pratique PythonAvancé

Relais API LLM : Centraliser Claude, OpenAI et Gemini

Gérer des clés API distinctes pour Claude, OpenAI et Gemini est une aberration opérationnelle. Le Relais API LLM résout la fragmentation des accès en offrant une interface unique et compatible OpenAI pour tous vos modèles.

L’enjeu est double : l’unification du code client et la réduction drastique des coûts via le partage de quotas. En utilisant un système de proxying, il est possible de mutualiser les abonnements (pinchen) et de distribuer l’usage de manière granulaire.

Après cette lecture, vous saurez implémenter un proxy capable de transformer des payloads hétérogènes et de gérer un pool de tokens partagé.

Relais API LLM

🛠️ Prérequis

Environnement Linux (Debian 12+ ou Ubuntu 22.04+) et les composants suivants :

  • Python 3.12+ (pour l’asynchronisme performant)
  • Docker 24.0+ ou Docker Compose
  • Redis 7.0+ (pour la gestion des quotas et le state)
  • Un accès à des clés API (OpenAI, Anthropic, Google Gemini)

📚 Comprendre Relais API LLM

Le Relais API LLM repose sur le pattern ‘Adapter’. Il agit comme une couche d’abstraction entre le client (format OpenAI) et les fournisseurs (formats propriétaires).

Client (OpenAI Format) 
      | 
      v
[ Relais API LLM ] <--- Adapter Pattern (Transformation JSON) 
      | 
      +--> Provider A (Anthropic Claude API)
      +--> Provider B (OpenPrime/OpenAI API)
      +--> Provider C (Google Gemini API)

Contrairement à un simple reverse proxy (Nginx), ce Relais API LLM effectue une mutation profonde du payload (deep inspection/mutation) et gère la logique de répartition des tokens entre utilisateurs via Redis.

🐍 Le code — Relais API LLM

Python
import asyncio
import httpx
from fastapi import FastAPI, Request, HTTPException
from typing import Dict, Any

app = FastAPI()

# Configuration des fournisseurs (simulée)
PROVIDERS = {
    "claude": "https://api.anthropic.com/v1/messages",
    "openai": "https://api.openai.com/v1/chat/completions",
    "gemini": "https://generativelanguage.googleapis.com/v1beta/models/gemini-pro:generateContent"
}

class LLMProxy:
    def __init__(self, client: httpx.AsyncClient):
        self.client = client

    async def proxy_request(self, provider_name: str, payload: Dict[str, Any]) -> Dict[str, Any]:
        url = PROVIDERS.get(provider_name)
        if not url:
            raise HTTPException(status_code=404, detail="Provider not found")

        # Ici, la transformation du format OpenAI vers le format cible se produit
        # Exemple simplifié pour la démonstration
        headers = {"Content-Type": "application/json", "x-api-key": "secret"}
        
        try:
            response = await self.client.post(url, json=payload, headers=headers, timeout=30.0)
            response.raise_for_status()
            return response.json()
        except httpx.HTTPStatusError as e:
            raise HTTPException(status_code=e.response.status_code, detail=str(e))

proxy_service = None

@app.on_event("startup")
async def startup():
    global proxy_service
    # Utilisation d'un client unique pour le connection pooling (essentiel)
    client = httpx.AsyncClient(limits=httpx:Limits(max_connections=100, max_keepalive_connections=20))
    proxy_service = LLMProxy(client)

@app.post("/v1/chat/completions/{provider}")
async def chat_proxy(provider: str, request: Request):
    payload = await request.json()
    return await proxy_service.proxy_request(provider, payload)

📖 Explication

Dans le premier snippet, l’utilisation de httpx.AsyncClient dans l’événement startup est cruciale. Créer un client par requête est une erreur classique qui épuise les descripteurs de fichiers (FD) du système. L’utilisation de limits permet de contrôler la pression sur les API distantes.

Dans le second snippet, la gestion du quota utilise redis.watch. C’est l’implémentation du pattern Optimistic Concurrency Control. Si deux requêtes tentent de modifier le même quota simultanément, la transaction échoue au lieu de permettre un dépassement de budget (over-spending). C’est indispensable pour un Relais API LLM utilisé en mode partage (pinchen).

Documentation officielle Python

🔄 Second exemple

Python
import redis.asyncio as redis
from typing import Optional

class QuotaManager:
    """Gestionnaire de quotas pour le partage de tokens (pinchen)."""
    
    def __init__(self, redis_url: str):
        self.redis = redis.from_url(redis_url, decode_responses=True)

    async def check_and_consume(self, user_id: str, tokens_requested: int) -> bool:
        """Vérifie si l'utilisateur a assez de tokens et décrémente le pool."""
        key = f"quota:{user_ide}"
        
        # Utilisation d'une transaction (WATCH/MULTI) pour l'atomicité
        async with self.redis.pipeline(transaction=True) as pipe:
            try:
                await pipe.watch(key)
                current_quota = await self.redis.get(key)
                
                if current_quota is None:
                    return False
                
                current_quota = int(current_quota)
                if current_quota < tokens_requested:
                    return False
                
                # Mise à jour atomique
                new_quota = current_quota - tokens_requested
                await pipe.multi()
                await pipe.set(key, new_quota)
                await pipe.execute()
                return True
            except redis.WatchError:
                # Concurrence détectée, on échoue pour éviter le dépassement
                return False

# Exemple d'usage technique
# quota_manager = QuotaManager("redis://localhost")
# success = await quota_lar.check_and_consume("user_123", 500)

Référence pratique

Le déploiement d’un Relais API LLM nécessite une configuration rigoureuse pour garantir la stabilité du flux. Voici les recettes essentielles pour une mise en production.

1. Configuration du routage multi-fournisseur

Pour transformer un appel OpenAI en appel Claude, vous devez mapper le champ messages vers le format anthropic. Dans votre configuration Sub2API-CRS2, définissez des règles de transformation JSON. Utilisez des expressions régulières pour injecter les headers d’authentification dynamiquement selon le provider cible.

2. Implémentation du mode ‘Fallback’ (Repli)

Si le fournisseur OpenAI renvoie une erreur 429 (Too Many Requests), le relais doit automatiquement basculer sur Gemini. Voici la logique à implémenter dans votre boucle asynchrone :

async def smart_forward(providers: list, payload: dict):
    for p in providers:
        try:
            return await proxy_service.proxy_request(p, payload)
        except Exception:
            continue
    raise Exception("Tous les providers sont indisponibles")

3. Gestion du streaming (Server-Sent Events)

Le streaming est critique pour l’expérience utilisateur. Ne lisez pas le corps de la réponse avec await response.json(). Utilisez response.aiter_bytes() pour streamer les chunks directement vers le client. Cela évite de saturer la RAM du serveur avec des buffers massifs lors de longues générations.

4. Stratégie de partage de coûts (Pinchen)

Pour le partage d’abonnement, utilisez Redis pour stocker un compteur global de tokens. Chaque requête au Relais API LLM doit décrémenter ce compteur. Si le compteur atteint zéro, le relais doit renvoyer une erreur 402 (Payment Required) ou basculer sur un autre pool de clés.

▶️ Exemple d’utilisation

Exemple d’appel au Relais API LLM pour interroger Claude via l’interface OpenAI :

curl http://localhost:8000/v1/chat/completions/claude \\
  -H "Content-Type: application/json" \\
  -d '{
    "model": "claude-3-opus",
    "messages": [{"role": "user", "content": "Hello!"}]
  }'

Sortie attendue (format OpenAI standard) :

{
  "id": "chatcmpl-123",
  "object": "chat.completion",
  "created": 1677652288,
  "model": "claude-3-opus",
  "choices": [{
    "index": 0,
    "message": {
      "role": "assistant",
      "content": "Hello! How can I assist you today?"
    },
    "finish_reason": "stop"
  }]
}

🚀 Cas d’usage avancés

1. Observabilité via Prometheus : Intégrez un middleware pour exposer le nombre de tokens consommés par provider. Cela permet de monitorer la dérive des coûts en temps réel.
2. A/B Testing de modèles : Dirigez 10% du trafic vers Gemini Pro et 90% vers GPT-4o pour comparer la latence et la qualité des réponses via le Relais API LLM.
3. Auto-rotation des clés : Implémentez un script Python qui surveille les erreurs 401 et met à jour la configuration du relais via un signal SIGHUP ou une API de gestion.

🐛 Erreurs courantes

⚠️

Instancier un client HTTP à l’intérieur de la fonction de route au lieu de le réutiliser.

✗ Mauvais

async def proxy(req): async with httpx.AsyncClient() as client: ...
✓ Correct

async def proxy(req): await client_global.post(...)

⚠️ Blocage de la boucle d'événements

Utiliser une bibliothèque synchrone (comme requests) dans une route FastAPI.

✗ Mauvais

resp = requests.post(url, json=data)
✓ Correct

resp = await httpx_client.post(url, json=data)

⚠️ Non-respect du format SSE

Tenter de parser le JSON d’une réponse en streaming avant de l’avoir reçue entièrement.

✗ Mauvais

data = await response.json()
✓ Correct

async for chunk in response.aiter_bytes(): yield chunk

⚠️ Race condition sur les quotas

Lire et écrire le quota Redis sans utiliser de transaction (WATCH/MULTI).

✗ Mauvais

val = r.get(k); r.set(k, val - 1)
✓ Correct

async with r.pipeline(transaction=True) as pipe: ... execute()

✅ Bonnes pratiques

Pour un Relais API LLM de production, respectez ces principes :

  • Immutabilité des payloads : Ne modifiez jamais l’objet request original, créez une copie pour la transformation vers le provider cible.
  • Timeout strict : Définissez un timeout par provider (ex: 30s pour OpenAI, 60s pour Claude) pour éviter l’accumulation de coroutines suspendues.
  • Typage statique : Utilisez mypy pour valider les schémas de transformation de payloads.
  • Backpressure : Implémentez un limiteur de débit (Rate Limiter) en amont du relais pour protéger vos propres ressources.
  • Logging structuré : Loggez les IDs de corrélation pour tracer une requête du client initial jusqu’au provider final.
Points clés

  • Le Relais API LLM unifie les interfaces Claude, OpenAI et Gemini.
  • L'utilisation de l'Adapter Pattern est indispensable pour la compatibilité.
  • Le partage de coûts (pinchen) repose sur une gestion atomique via Redis.
  • L'asynchronisme (asyncio) est obligatoire pour gérer le streaming SSE.
  • Le pooling de connexions (httpx.AsyncClient) évite l'épuisement des sockets.
  • La transformation de payload doit être testée avec des schémas Pydantic.
  • Le mode fallback améliore la résilience face aux pannes de fournisseurs.
  • Le déploiement via Docker simplifie la gestion des dépendances système.

❓ Questions fréquentes

Est-ce que le relais augmente la latence ?

Oui, un overhead de 10 à 50ms est à prévoir pour la transformation JSON et le parsing des headers. Cela reste négligeable face au temps de génération du LLM.

Comment gérer la sécurité des clés API ?

Les clés ne doivent jamais transiter par le client. Le Relais API LLM doit injecter les clés secrètes depuis des variables d’environnement ou un coffre-fort (Vault).

Peut-on utiliser ce système pour du streaming ?

Oui, à condition d’utiliser des générateurs asynchrones et de ne pas tenter de désérialiser le corps de la réponse en JSON complet.

Le système supporte-t-il le multi-utilisateur ?

Absolument, via l’implémentation d’un middleware d’authentification et d’un gestionnaire de quotas lié à un ID utilisateur dans Redis.

📚 Sur le même blog

🔗 Le même sujet sur nos autres blogs

📝 Conclusion

Le Relais API LLM est l’outil indispensable pour quiconque souhaite orchestrer plusieurs modèles d’IA sans multiplier la complexité du code client. En centralisant la logique de transformation et de quota, vous transformez des abonnements fragmentés en une infrastructure unique et scalable. Pour approfondir la gestion des types dans vos transformations, consultez la documentation Python officielle. Un proxy bien configuré est la base d’un système d’agent IA résilient.

Laisser un commentaire

Votre adresse e-mail ne sera pas publiée. Les champs obligatoires sont indiqués avec *