Proxy API LLM

Proxy API LLM : implémenter l’unification CCX

Référence pratique PythonAvancé

Proxy API LLM : implémenter l'unification CCX

Gérer plusieurs SDK propriétaires pour Claude, Gemini et Codex transforme rapidement un projet en usine à spaghetti. Le Proxy API LLM résout ce problème en imposant une interface unique et typée via le pattern Adapter.

L’enjeu est de réduire la surface de maintenance de 60% en centralisant la logique de retry, de timeout et de formatage. Un Proxy API LLement bien conçu permet de changer de fournisseur sans modifier une seule ligne de code client.

Après ce tour d’horizon, vous saurez implémenter un middleware asynchrone capable de router des requêtes vers n’importe quel fournisseur d’IA en utilisant Python 3.12 et Pydantic v2.

Proxy API LLM

🛠️ Prérequis

Installation de l’environnement de développement nécessaire :

  • Python 3.12+ (indispensable pour les améliorations de performance du loop asyncio)
  • pip install fastapi httpx pydantic-settings pydantic-ai
  • Un accès aux clés API de Anthropic, Google et OpenAI

📚 Comprendre Proxy API LLM

Le Proxy API LLM repose sur le pattern Adapter. Au lieu de consommer les types spécifiques de chaque SDK, nous définissons un contrat de base (Interface) que chaque fournisseur doit implémenter.

Schéma de flux :
Client (Standard OpenAI Format) -> CCX Proxy (Unification/Routing) -> [Adapter Claude | Adapter Gemini | Adapter Codex]

Contrairement à une simple redirection Nginx, le Proxy API LLM effectue une transformation de payload (payload transformation). Il doit mapper le champ messages de l’un vers le format contents de l’autre, tout en gérant la conversion des tokens. En Python, l’utilisation de Protocol de la bibliothèque typing est préférable à l’héritage classique pour une approche plus duck-typed et flexible.

🐍 Le code — Proxy API LLM

Python
from typing import Protocol, Any
import httpx
from pydantic import BaseModel

class LLMResponse(BaseModel):
    content: str
    usage: dict[str, int]

class LLMProvider(Protocol):
    """Définition de l'interface pour chaque fournisseur."""
    async def generate(self, prompt: str) -> LLMResponse:
        ...

class ClaudeAdapter:
    def __init__(self, api_key: str, client: httpx.AsyncClient):
        self.api_key = api_key
        self.client = client
        self.url = "https://api.anthropic.com/v1/messages"

    async def generate(self, prompt: str) -> LLResponse:
        # Adaptation du format vers le standard Anthropic
        headers = {"x-api-key": self.api_key, "anthropic-version": "2023-06-01"}
        payload = {
            "model": "claude-3-opus-20240229",
            "messages": [{"role": "user", "content": prompt}]
        }
        resp = await self.client.post(self.url, json=payload, headers=headers)
        data = resp.json()
        # Extraction et normalisation du résultat
        return LLMResponse(
            content=data['content'][0]['text'],
            usage=data['usage']
        )

📖 Explication

Dans code_source, l’utilisation de Protocol est cruciale. Contra-réirement à ABC, cela permet de valider la conformité des adaptateurs sans qu’ils partagent une hiérarchie de classes lourde. C’est le principe du duck-typing statique via mypy.

Le choix de httpx.AsyncClient plutôt que requests est dicté par la nécessité de ne pas bloquer l’event loop de l’application FastAPI lors des appels réseau. Un blocage ici paralyserait toutes les autres requêtes en cours du Proxy API LLM.

Attention au piège de la gestion des sessions : ne créez jamais un AsyncClient à l’intérieur de la méthode generate. Cela provoquerait une fuite de sockets et une surcharge de la pile TCP. Le client doit être injecté et partagé (Singleton pattern) durant toute la durée de vie de l’application.

Documentation officielle Python

🔄 Second exemple

Python
from pydantic import BaseModel, Field
from typing import List, Optional

class UnifiedMessage(BaseModel):
    role: str = Field(..., pattern="^(user|assistant|system)$")
    content: str

class UnifiedRequest(BaseModel):
    model: str
    messages: List[UnifiedMessage]
    temperature: float = 0.7
    max_tokens: Optional[int] = 1024

class UnifiedResponse(BaseESSBaseModel):
    id: str
    choices: List[dict[str, Any]]
    usage: dict[str, int]

Référence pratique

Voici les recettes essentielles pour transformer un simple proxy en un Proxy API LLM de production.

1. Implémentation d’un mécanisme de Fallback

Si le fournisseur principal (ex: Claude) renvoie une erreur 500 ou un 429 (Rate Limit), le Proxy API LLM doit basculer automatiquement sur Gemini. Voici la logique à implémenter dans votre routeur :

async def smart_route(prompt: str, providers: list[LLMProvider]) -> LLMResponse:
    for provider in providers:
        try:
            # Tentative avec timeout strict pour éviter de bloquer la file
            return await asyncio.wait_for(provider.generate(prompt), timeout=30.0)
        except (httpx.HTTPStatusError, asyncio.TimeoutError):
            continue # Passage au fournisseur suivant
    raise Exception("Tous les fournisseurs ont échoué")

2. Calcul de coût en temps réel

Le Proxy API LLM peut injecter des métadonnées de coût. En utilisant Pydantic, vous pouvez enrichir la réponse avec un champ estimated_cost basé sur le nombre de tokens détectés dans le payload retourné par le fournisseur.

3. Mise en cache par Hash de Prompt

Évitez de payer deux fois pour la même requête. Utilisez hashlib.sha256 sur le contenu du prompt pour créer une clé de cache Redis. Si la clé existe, le Proxy API LLM renvoie la réponse stockée sans appeler l’API externe.

4. Gestion du Streaming (SSE)

Pour ne pas dégrader l’expérience utilisateur, le Proxy API LLM doit supporter le streaming. Utilisez httpx.AsyncClient.stream("POST", ...) et ré-émettez les chunks via une StreamingResponse de FastAPI. Attention : la transformation du format de stream est la partie la plus complexe car chaque fournisseur utilise un format de chunk différent (ex: data: {...} vs chunks bruts).

▶️ Exemple d’utilisation

Exemple d’appel au Proxy API LLM avec une bibliothèque client standard :

import httpx

async def main():
    async with httpx.AsyncClient() as client:
        payload = {"model": "claunode-3", "messages": [{"role": "user", "content": "Hello"}]}
        response = await client.post("http://localhost:8000/v1/chat/completions", json=payload)
        print(response.json()["choices"][0]["message"]["content"])

# Sortie attendue :
# "Bonjour ! Comment puis-je vous aider aujourd'hui ?"

🚀 Cas d’usage avancés

1. A/B Testing de modèles : Répartissez 10% du trafic vers un nouveau modèle (ex: Gemini 1.5 Pro) via le Proxy API LLM pour comparer la latence et la qualité des réponses sans changer le code client.

2. Audit de sécurité (PII Masking) : Interceptez les UnifiedMessage dans le middleware pour détecter et masquer les données sensibles (numéros de CB, emails) avant qu’elles ne quittent votre infrastructure vers les serveurs d’Anthropic ou OpenAI.

3. Injection de System Prompt global : Forcez une consigne de sécurité (ex: « Réponds toujours en français ») en injectant systématiquement un message de rôle system au début de la liste messages du Proxy API LLM.

🐛 Erreurs courantes

⚠️ Fuite de sockets

Instanciation d’un client HTTP à chaque requête.

✗ Mauvais

async with httpx.AsyncClient() as client: await client.post(...)
✓ Correct

client: AsyncClient (injecté via dépendance FastAPI)

⚠️ Blocage de l'Event Loop

Utilisation de la bibliothèque ‘requests’ (synchrone) dans une fonction async.

✗ Mauvais

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

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

⚠️ Erreur de typage Pydantic

Oubli de la validation du pattern sur les rôles de messages.

✗ Mauvais

role: str
✓ Correct

role: str = Field(..., pattern="^(user|assistant|system)$")

⚠️ Timeout mal configuré

Ne pas définir de timeout sur les appels vers des LLM lents.

✗ Mauvais

await client.post(url, json=payload)
✓ Correct

await client.post(url, json=payload, timeout=60.0)

✅ Bonnes pratiques

Pour un Proxy API LLM de niveau production, respectez ces principes :

  • Utilisez le typage statique strict : Configurez pyright ou mypy en mode strict pour intercepter les erreurs de mapping de payload avant l’exécution.
  • Implémentez le pattern Circuit Breaker : Si un fournisseur échoue 5 fois de suite, stoppez les appels vers lui pendant 30 secondes pour laisser le service récupérer.
  • Loggez la latence par fournisseur : Utilisez des middleware Prometheus pour suivre la distribution des temps de réponse (P95, P99).
  • Standardisez les erreurs : Le Proxy API LLM doit toujours retourner un format d’erreur compatible avec l’API OpenAI, même si le fournisseur original renvoie un format bizarre.
  • Gestion des secrets : Ne jamais coder les clés API en dur. Utilisez pydantic-settings pour charger les variables d’environnement de manière sécurisée.
Points clés

  • Le Proxy API LLM unifie les interfaces disparates (Claude, Gemini, Codex).
  • Utilisez le pattern Adapter pour isoler la logique de chaque fournisseur.
  • L'utilisation de httpx.AsyncClient est impérative pour la performance.
  • Pydantic v2 permet une validation ultra-rapide des payloads entrants.
  • Le pattern Fallback garantit la haute disponibilité du service.
  • Le cache par hash de prompt réduit drastiquement les coûts d'API.
  • Le streaming via SSE est complexe mais nécessaire pour l'UX.
  • L'injection de dépendances permet de tester les adaptateurs isolément.

❓ Questions fréquentes

Est-ce que ce proxy augmente la latence ?

L’overhead est négligeable (souvent < 5ms) par rapport au temps de génération du LLM. Le gain en fiabilité compense largement ce coût.

Peut-on utiliser ce proxy avec l'OpenAI SDK officiel ?

Oui, si votre Proxy API LLM respecte scrupuleusement le format de réponse de l’API OpenAI (le format ‘chat/completions’).

Comment gérer le streaming des réponses ?

Il faut utiliser les streams de httpx et renvoyer un générateur asynchrone via FastAPI pour maintenir une connexion ouverte.

Est-ce sécurisé pour les données d'entreprise ?

Le proxy est l’endroit idéal pour implémenter un filtrage de données sensibles (DLP) avant l’envoi vers les API tierces.

📚 Sur le même blog

🔗 Le même sujet sur nos autres blogs

📝 Conclusion

Le Proxy API LLM est l’infrastructure indispensable pour toute application multi-LLM sérieuse. Il transforme une architecture fragile et dépendante en un système résilient et interchangeable. Le coût de l’abstraction est une latence minime face au gain de maintenabilité. Pour approfondir la gestion des types en Python, consultez la documentation Python officielle. N’oubliez pas : la complexité ne doit jamais être cachée, elle doit être encapsulée.

Laisser un commentaire

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