9. Production Patterns

# Parameters
BATCH_MODE = "true"

Patterns de Production : APIs Avancées OpenAI

Navigation : Index | << Précédent | Suivant >>

Ce notebook couvre les fonctionnalités avancées nécessaires pour des applications en production : - Conversations API : Persistance d’état entre sessions - Background Mode : Tâches asynchrones longues - Rate Limiting : Gestion des limites d’API - Optimisation : Réduction des coûts

Objectifs : - Gérer des conversations multi-sessions - Exécuter des tâches en arrière-plan - Implémenter des patterns de résilience - Optimiser les coûts d’API

Prérequis : Notebooks 1-4

Durée estimée : 70 minutes

from pathlib import Path
%pip install -q openai python-dotenv tenacity

import os
import time
from openai import OpenAI
from dotenv import load_dotenv
from tenacity import retry, stop_after_attempt, wait_exponential

# Chargement robuste de la configuration .env
from dotenv import load_dotenv
import os
# Recherche du .env dans tous les parents (pour Papermill qui change le cwd)
current_path = Path.cwd()
env_loaded = False
for _ in range(10):
    env_path = current_path / ".env"
    if env_path.exists():
        load_dotenv(env_path)
        print(f".env charge depuis: {env_path.name}")
        env_loaded = True
        break
    if current_path.name == "GenAI" or len(current_path.parts) <= 1:
        break
    current_path = current_path.parent
if not env_loaded:
    print("WARNING: .env non trouve, utilisation variables environnement")
client = OpenAI()

# Modèle par défaut depuis .env
DEFAULT_MODEL = os.getenv("OPENAI_MODEL", "gpt-5-mini")
BATCH_MODE = os.getenv("BATCH_MODE", "false").lower() == "true"

print("Client OpenAI initialisé !")
print(f"Modèle par défaut: {DEFAULT_MODEL}")
print(f"Mode batch: {BATCH_MODE}")

Note: you may need to restart the kernel to use updated packages.
.env charge depuis: .env
Client OpenAI initialisé !
Modèle par défaut: gpt-5-mini
Mode batch: False

Vérification de l’Environnement

Composants installés et chargés :

Bibliothèques Python :
- openai : SDK officiel pour l’API OpenAI
- python-dotenv : Chargement sécurisé des variables d’environnement
- tenacity : Gestion avancée des retry avec backoff exponentiel
Configuration extraite du fichier .env :
- OPENAI_MODEL : Modèle par défaut (ex: gpt-5-mini)
- BATCH_MODE : Active le mode batch pour tests automatisés (skip les inputs interactifs)

Sortie attendue :

Client OpenAI initialisé !
Modèle par défaut: gpt-5-mini
Mode batch: False

Points de validation :

Élément	Validation	Action si erreur
Client initialisé	✅ Pas d’exception	Vérifier `OPENAI_API_KEY` dans `.env`
Modèle détecté	✅ Affiche un nom valide	Définir `OPENAI_MODEL` dans `.env`
Mode batch	✅ `True` ou `False`	Optionnel, par défaut `False`

Sécurité : Ne JAMAIS afficher la clé API dans les logs. Le SDK utilise automatiquement la variable d’environnement OPENAI_API_KEY sans exposition dans le code.

1. Gestion du Contexte avec Chat Completions

La gestion manuelle du contexte avec Chat Completions permet de : - Maintenir l’historique : Conserver les messages précédents dans une liste - Chaîner les conversations : Chaque requête inclut tout le contexte - Contrôle total : Gestion explicite de ce qui est envoyé

Cas d’usage : - Conversations multi-tours - Chatbots avec mémoire - Systèmes de Q&A contextuels

Architecture :

messages = [
    {"rôle": "system", "content": "..."},  # Optionnel
    {"rôle": "user", "content": "..."},
    {"rôle": "assistant", "content": "..."},  # Réponse précédente
    {"rôle": "user", "content": "..."}  # Nouvelle question
]

Note importante : La Responses API avec store=True n’est pas disponible dans cette version. Nous utilisons Chat Completions avec gestion manuelle de l’historique.

# Premier message - utilisation de Chat Completions standard
# Note: La Responses API avec store n'est pas disponible dans cette version
# Nous utilisons Chat Completions avec gestion manuelle du contexte

from openai import OpenAI

# Budget max_completion_tokens relevé à 4000 : gpt-5-mini est un modèle de
# raisonnement dont les tokens de raisonnement sont déduits du budget. Avec
# 200, le raisonnement pouvait consommer tout le budget sans laisser de texte
# final visible (issue #3571). 1000 laisse une marge confortable pour le contenu.

# Conversation avec historique géré manuellement
conversation_history = []

# Premier message
conversation_history.append({
    "role": "user", 
    "content": "Je m'appelle Jean et j'habite à Paris. Retiens ces informations."
})

response1 = client.chat.completions.create(
    model=DEFAULT_MODEL,
    messages=conversation_history,
    max_completion_tokens=4000
)

content1 = response1.choices[0].message.content
print(f"Response 1 ID: {response1.id}")
print(f"Contenu: {content1[:200]}...")

# Ajouter la réponse à l'historique
conversation_history.append({
    "role": "assistant",
    "content": content1
})

# Deuxième message (contexte préservé via l'historique)
conversation_history.append({
    "role": "user",
    "content": "Quel est mon nom et où j'habite?"
})

response2 = client.chat.completions.create(
    model=DEFAULT_MODEL,
    messages=conversation_history,
    max_completion_tokens=4000
)

content2 = response2.choices[0].message.content
print(f"\nResponse 2 ID: {response2.id}")
print(f"Contenu: {content2}")

# Vérifier les tokens
print(f"\nTokens utilisés: {response2.usage.prompt_tokens} input / {response2.usage.completion_tokens} output")

Response 1 ID: chatcmpl-ELTuaoVuyXYsHUQ6NckYTl9Sgp9Yj
Contenu: D'accord, Jean — j'ai noté que vous vous appelez Jean et que vous habitez à Paris. Je peux conserver et utiliser ces informations pendant cette conversation pour personnaliser mes réponses. 

Je ne pe...

Response 2 ID: chatcmpl-ELTugKaBsAFmdWQzKmNj2N93Etm5i
Contenu: Votre nom est Jean et vous habitez à Paris.

Tokens utilisés: 133 input / 148 output

Interprétation des Résultats

Observations importantes :

Gestion manuelle du contexte : L’historique est conservé dans une liste conversation_history
Contexte préservé : En passant tout l’historique à chaque appel, le modèle a accès aux échanges précédents. La réponse 2 restitue correctement les informations mémorisées (“Vous vous appelez Jean et vous habitez à Paris.”), preuve que le contexte est bien transmis entre les tours.
Consommation de tokens : Les tokens d’entrée augmentent avec la taille de l’historique (110 tokens input au second tour, qui inclut tout le contexte accumulé).

Note technique (issue #3571) : gpt-5-mini est un modèle de raisonnement : les tokens de raisonnement sont déduits de max_completion_tokens. Le budget ici est relevé à 4000 pour garantir une marge confortable de contenu final après le raisonnement — un budget trop bas (≤500) pouvait auparavant être entièrement consommé par le raisonnement sans produire de texte visible.

Pattern utilisé :

# Ajouter question utilisateur
conversation_history.append({"rôle": "user", "content": question})

# Appel API avec tout l'historique
response = client.chat.completions.create(
    model=DEFAULT_MODEL,
    messages=conversation_history
)

# Ajouter réponse au contexte
conversation_history.append({"rôle": "assistant", "content": response.choices[0].message.content})

Optimisation : Pour les longues conversations, envisagez de : - Limiter l’historique aux N derniers échanges - Résumer le contexte périodiquement - Utiliser un système RAG pour les contextes très longs

2. Conversations Multi-Tours

La gestion des conversations multi-tours avec Chat Completions repose sur : - Historique cumulatif : Chaque message s’ajoute à la liste - Contexte automatique : Le modèle voit tout l’historique à chaque appel - Flexibilité : Possibilité de modifier l’historique (résumé, filtrage)

Architecture :

messages = [system, user1, assistant1, user2, assistant2, user3, ...]
              ↓
        Chat Completions API
              ↓
         assistant3

Avantages : - Simple à implémenter - Contrôle total sur le contexte - Compatible avec tous les modèles

# Simulation de conversation multi-turn avec Chat Completions
# Le contexte est géré en conservant l'historique des messages

print("=== Conversation multi-turn ===\n")

# Budget max_completion_tokens relevé à 4000 : gpt-5-mini est un modèle de
# raisonnement dont les tokens de raisonnement sont déduits du budget. Avec
# 200, le raisonnement pouvait consommer tout le budget sans texte final
# visible (issue #3571).

# Historique de conversation
messages = [
    {"role": "system", "content": "Tu es un assistant de voyage expert."},
    {"role": "user", "content": "Je planifie un voyage au Japon."}
]

# Premier échange
resp1 = client.chat.completions.create(
    model=DEFAULT_MODEL,
    messages=messages,
    max_completion_tokens=4000
)
assistant_reply1 = resp1.choices[0].message.content
messages.append({"role": "assistant", "content": assistant_reply1})
print(f"Assistant: {assistant_reply1[:200]}...")

# Deuxième échange (contexte automatiquement préservé via messages)
messages.append({"role": "user", "content": "Quels sont les meilleurs endroits à Tokyo?"})
resp2 = client.chat.completions.create(
    model=DEFAULT_MODEL,
    messages=messages,
    max_completion_tokens=4000
)
assistant_reply2 = resp2.choices[0].message.content
messages.append({"role": "assistant", "content": assistant_reply2})
print(f"\nAssistant: {assistant_reply2[:200]}...")

# Troisième échange
messages.append({"role": "user", "content": "Quelle est la meilleure période pour y aller?"})
resp3 = client.chat.completions.create(
    model=DEFAULT_MODEL,
    messages=messages,
    max_completion_tokens=4000
)
assistant_reply3 = resp3.choices[0].message.content
messages.append({"role": "assistant", "content": assistant_reply3})
print(f"\nAssistant: {assistant_reply3[:200]}...")

print(f"\n{len([m for m in messages if m['role'] != 'system'])} messages dans la conversation")
print(f"  Contexte: Japon → Tokyo → Meilleure période")

=== Conversation multi-turn ===

Assistant: Super — je peux t’aider à tout planifier. Pour commencer, peux-tu me dire quelques infos rapides ?
- Dates (ou saison) et durée du voyage ?
- Ville d’arrivée/départ (Tokyo, Osaka, autre) ?
- Voyage so...

Assistant: Voici une sélection organisée des meilleurs endroits à Tokyo, avec pour chacun un court descriptif et un conseil pratique — utile pour préparer ton itinéraire.

Incontournables (top 10)
- Shibuya (Shi...

Assistant: Bonne question — la “meilleure” période dépend vraiment de ce que tu veux voir et faire. Voici un résumé saison par saison avec avantages/inconvénients et conseils pratiques pour t’aider à choisir.

P...

6 messages dans la conversation
  Contexte: Japon → Tokyo → Meilleure période

Analyse de la Conversation Multi-Tour

Points clés observés :

Historique manuel : Le contexte est conservé dans une liste conversation_history (Chat Completions), chaque tour y est ajouté avant l’appel API. Il n’y a pas de previous_response_id (mécanisme de la Responses API, non utilisée ici – OpenRouter ne la supporte pas).
Contexte cumulatif : Le modèle comprend le contexte complet :
- Message 1 : Établit le rôle (assistant de voyage) + destination (Japon)
- Message 2 : Peut répondre spécifiquement sur Tokyo (contexte préservé)
- Message 3 : Comprend qu’on parle toujours du Japon
Structure de données :

resp1 (Japon)
  └─ resp2 (Tokyo) 
      └─ resp3 (Meilleure période)

Cas d’usage production : - Chatbots de support client (reprendre une conversation après déconnexion) - Assistants multi-sessions (planification, conseil) - Tutoriels interactifs (mémoriser les réponses précédentes)

Comprendre le Polling

Pourquoi le polling ?

En mode background, le serveur ne peut pas “pousser” le résultat vers le client. Le client doit donc interroger régulièrement le serveur pour vérifier le statut.

Pattern typique observé :

t=0s   : status = "pending"     → Attendre 5s
t=5s   : status = "processing"  → Attendre 5s
t=10s  : status = "processing"  → Attendre 5s
t=15s  : status = "completed"   → Récupérer le résultat

Optimisations possibles : - Backoff progressif : Attendre 2s, puis 5s, puis 10s, etc. (éviter surcharge) - Webhooks : Le serveur appelle votre API quand c’est terminé (pas supporté par OpenAI actuellement) - WebSockets : Connexion persistante pour notifications temps réel

Quand utiliser background mode ? - ✅ Analyses longues (>30s) - ✅ Génération de rapports complexes - ✅ Tâches où l’utilisateur peut attendre - ❌ Chatbot temps réel (utiliser streaming à la place)

3. Streaming pour UX Réactive

Le Streaming permet d’améliorer l’expérience utilisateur : - Feedback immédiat : Les tokens apparaissent progressivement - Latence perçue réduite : Premier token en ~200ms - Annulation possible : L’utilisateur peut arrêter la génération

Quand utiliser le streaming ? - ✅ Réponses longues (>100 tokens) - ✅ Chatbots en temps réel - ✅ Génération de contenu (articles, rapports)

Alternative pour tâches longues : - API Batch OpenAI : Pour les traitements asynchrones (délai 24h) - Architecture async : Celery, Redis Queue pour le polling

Décorticage du Retry

Fonctionnement du décorateur @retry :

Première tentative : Appel normal de la fonction
En cas d’échec : Si l’exception est RateLimitError ou APIError :
- Attendre 2s (multiplier=1, min=2)
- Tentative 2
Nouvel échec : Attendre 4s (backoff exponentiel)
Et ainsi de suite : jusqu’à 5 tentatives max

Exemple de scénario réel :

t=0s    : Requête → RateLimitError (429 Too Many Requests)
t=2s    : Retry #1 → APIError (503 Service Unavailable)
t=6s    : Retry #2 → Succès ✅

Pourquoi exponentiel ? - Éviter l’effet “thundering herd” : Si 1000 clients retentent simultanément après 2s, le serveur reste surchargé - Avec backoff exponentiel, les requêtes sont étalées : 2s, 4s, 8s, 16s, 32s

Limites : - ⚠️ Ne résout pas les erreurs permanentes (authentification, prompt invalide) - ⚠️ Peut augmenter la latence totale (jusqu’à 2+4+8+16+32 = 62s)

# Background Mode non disponible avec Chat Completions standard
# Alternative: utiliser l'API Batch d'OpenAI pour les tâches longues

if not BATCH_MODE:
    print("=== Note sur le Background Mode ===")
    print()
    print("Le Background Mode n'est pas disponible avec l'API Chat Completions standard.")
    print()
    print("Alternatives pour les tâches longues:")
    print("  1. API Batch OpenAI: Pour les traitements asynchrones (24h délai)")
    print("  2. Streaming: Pour les réponses longues avec feedback utilisateur")
    print("  3. Architecture async: Celery, Redis Queue pour le polling")
    print()
    
    # Démonstration de streaming comme alternative
    print("=== Alternative: Streaming pour tâches longues ===")
    stream = client.chat.completions.create(
        model=DEFAULT_MODEL,
        messages=[{
            "role": "user", 
            "content": "Analyse en 3 points les avantages des microservices."
        }],
        stream=True,
        max_completion_tokens=300
    )
    
    print("Réponse progressive: ", end="", flush=True)
    for chunk in stream:
        if chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)
    print("\n")
else:
    print("[BATCH_MODE] Simulation de streaming:")
    print("Réponse progressive: Les microservices offrent 3 avantages majeurs...")

=== Note sur le Background Mode ===

Le Background Mode n'est pas disponible avec l'API Chat Completions standard.

Alternatives pour les tâches longues:
  1. API Batch OpenAI: Pour les traitements asynchrones (24h délai)
  2. Streaming: Pour les réponses longues avec feedback utilisateur
  3. Architecture async: Celery, Redis Queue pour le polling

=== Alternative: Streaming pour tâches longues ===
Réponse progressive: 1) Scalabilité fine et optimisation des ressources  
   Les microservices permettent de faire évoluer indépendamment chaque compos

Comportement du Rate Limiter

Analyse de l’exécution :

Dans cet exemple avec max_requests_per_minute=10 : - Les 3 requêtes passent instantanément (aucune attente) - Pourquoi ? Parce que 3 < 10 (limite non atteinte)

Test avec limite serrée :

Si nous avions configuré max_requests_per_minute=2 :

t=0s   : Requête 1 → OK (1/2)
t=0.5s : Requête 2 → OK (2/2)
t=1s   : Requête 3 → ATTENTE de ~59s (fenêtre de 60s)
t=60s  : Requête 3 → OK

Fenêtre glissante vs Fenêtre fixe :

Notre implémentation utilise une fenêtre glissante : - ✅ Plus précis : compte exactement les 60 dernières secondes - ✅ Pas d’effet “reset brutal” à chaque minute

Alternative fenêtre fixe :

# Moins précis mais plus simple
if current_minute != last_minute:
    counter = 0
    last_minute = current_minute
counter += 1
if counter > max_rpm:
    wait()

En production : - Combiner rate limiter côté client (courtoisie) + côté serveur (sécurité) - Adapter la limite à votre tier OpenAI (voir dashboard usage limits)

4. Rate Limiting et Retry

En production, il est essentiel de gérer les erreurs transitoires et les limites d’API :

Retry avec Backoff Exponentiel

La bibliothèque tenacity permet d’implémenter facilement un retry automatique : - Backoff exponentiel : Attendre 2s, puis 4s, puis 8s, etc. - Retry sélectif : Uniquement sur certaines exceptions - Limite de tentatives : Éviter les boucles infinies

Pattern recommandé :

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=60),
    retry=retry_if_exception_type((RateLimitError, APIError))
)

Erreurs à gérer : - RateLimitError : Limite de requêtes dépassée (429) - APIError : Erreur serveur temporaire (500, 502, 503) - Timeout : Délai d’attente dépassé

Comparaison des Résultats

Analyse coût/bénéfice :

Pour ces deux requêtes similaires (résumé en 3 points), observons :

Métrique	Test 1 (Python)	Test 2 (JavaScript)
Tokens input	17	18
Tokens output	311	415
Coût	$0.000189	$0.000252

Analyse des coûts : - optimized_completion est un simple Chat Completions (pas de cache activé) - Le Test 2 (JavaScript) génère davantage de tokens de sortie (415 vs 311), d’où un coût légèrement supérieur - Le prompt caching OpenAI nécessite un préfixe identique répété, ce qui n’est pas le cas ici (“Python” vs “JavaScript”)

Test optimal pour le cache :

# Requête 1
resp1 = optimized_completion("Contexte: Documentation Python. Question: Qu'est-ce qu'une liste?")

# Requête 2 (même préfixe "Contexte: Documentation Python.")
resp2 = optimized_completion("Contexte: Documentation Python. Question: Qu'est-ce qu'un tuple?")
# → Économie de ~50% sur les tokens du contexte

Gains typiques du cache : - Conversations longues : 40-60% économie - RAG avec contexte fixe : 60-80% économie - Requêtes isolées : 0% (aucun préfixe commun)

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from openai import RateLimitError, APIError

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=60),
    retry=retry_if_exception_type((RateLimitError, APIError))
)
def safe_completion(prompt: str, model: str = None) -> str:
    """Appel API avec retry automatique sur erreurs transitoires"""
    model = model or DEFAULT_MODEL
    # Budget relevé à 4000 : gpt-5-mini est un modèle de raisonnement dont les
    # tokens de raisonnement sont déduits de max_completion_tokens. Avec 200,
    # le raisonnement pouvait consommer tout le budget sans texte final visible
    # (issue #3571).
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        max_completion_tokens=4000
    )
    return response.choices[0].message.content

# Test
result = safe_completion("Dis 'Hello World' en 5 langues.")
print(result)

Voici "Hello World" en 5 langues :

- Français : « Bonjour le monde »
- Espagnol : « ¡Hola Mundo! »
- Allemand : « Hallo Welt! »
- Chinois (mandarin) : « 你好，世界 » (Nǐ hǎo, shìjiè)
- Japonais : « こんにちは世界 » (Konnichiwa sekai)

Interprétation des Logs

Format de log structuré :

2026-02-04 15:23:45 | OpenAI_Production | INFO | SUCCESS | Model: gpt-5-mini | Duration: 1.23s | Tokens: 85 (in: 25, out: 60)

Champs clés : - Timestamp : Horodatage précis (important pour corrélation) - Level : INFO (succès) ou ERROR (échec) - Model : Tracer quel modèle a été utilisé - Duration : Latence end-to-end (objectif : <2s pour chatbot) - Tokens : Consommation (surveiller les dépassements)

Analyse des logs :

Log SUCCESS :
- Durée ~1-2s : Normal pour gpt-5-mini
- 85 tokens total : Petit prompt + réponse courte
- ✅ Requête efficace
Log ERROR :
- InvalidRequestError : Modèle invalide
- Capturé et loggé avant propagation
- ✅ Permet diagnostic rapide sans crash

Agrégation recommandée :

# Latence moyenne par modèle (via grep + awk)
grep "SUCCESS" production.log | awk '{print $8, $12}' | sort

# Taux d'erreur sur 1h
grep "ERROR" production.log | wc -l

Outils professionnels : - ELK Stack (Elasticsearch, Logstash, Kibana) - Datadog, Splunk, New Relic - Prometheus + Grafana

Transition vers les Patterns Avancés

Jusqu’ici nous avons couvert les patterns de résilience et d’optimisation : - ✅ Retry automatique - ✅ Rate limiting - ✅ Logging structuré - ✅ Optimisation des coûts

Les deux derniers patterns concernent l’expérience utilisateur et la sécurité :

Streaming : Améliorer la perception de latence
- Au lieu d’attendre 5s pour afficher 500 tokens
- Afficher progressivement (premier token en 200ms)
- Utilisateur voit la “pensée en temps réel”
Modération : Filtrer le contenu inapproprié
- Avant de traiter : vérifier que l’input est acceptable
- Après génération : s’assurer que l’output est sûr
- Éviter les violations de politique d’usage

Ces patterns sont essentiels pour toute application destinée aux utilisateurs finaux.

Analyse des Résultats

Streaming :

Observez le comportement : - Les tokens apparaissent progressivement (un par un ou par petits groupes) - Latence au premier token : ~200-500ms - Latence totale : Identique à une requête non-streamée (~2s) - Gain perçu : L’utilisateur voit le début de la réponse immédiatement

Implémentation UX :

# Dans une vraie app web
async def stream_to_user():
    for chunk in stream:
        content = chunk.choices[0].delta.content
        if content:
            await websocket.send(content)  # Envoyer au navigateur en temps réel

Modération :

Les résultats montrent : 1. Texte neutre : “Bonjour, comment vas-tu?” - flagged = False : Aucun problème détecté

Texte menaçant : “Enfoiré, je vais te tuer!”
- flagged = True : Détecté (harcèlement, menace, violence)

Cas qui seraient flaggés : - Contenu haineux, violent, sexuel - Harcèlement, menaces - Auto-mutilation

Pattern de sécurité :

# Avant de traiter
if moderate(user_input).flagged:
    return "Désolé, ce contenu viole notre politique."

# Générer
response = optimized_completion(user_input)["content"]

# Après génération
if moderate(response).flagged:
    return "Désolé, impossible de répondre à cette demande."

Rate Limiter Personnalisé

Pour respecter les limites de requêtes par minute (RPM), implémentons un rate limiter : - Fenêtre glissante : Compte les requêtes sur les 60 dernières secondes - Attente automatique : Bloque jusqu’à ce qu’une requête soit autorisée - Configurable : Adapter selon votre tier OpenAI

Limites par tier (exemple) : - Free : 3 RPM - Tier 1 : 500 RPM - Tier 2 : 5000 RPM - Tier 5 : 10000 RPM

import time
from collections import deque

class RateLimiter:
    """Limite le nombre de requêtes par minute"""
    def __init__(self, max_requests_per_minute: int = 60):
        self.max_rpm = max_requests_per_minute
        self.requests = deque()
    
    def wait_if_needed(self):
        now = time.time()
        # Nettoyer les anciennes requêtes (plus de 60s)
        while self.requests and now - self.requests[0] > 60:
            self.requests.popleft()
        
        # Si limite atteinte, attendre
        if len(self.requests) >= self.max_rpm:
            wait_time = 60 - (now - self.requests[0])
            if wait_time > 0:
                print(f"Rate limit: attente de {wait_time:.1f}s")
                time.sleep(wait_time)
        
        self.requests.append(time.time())

# Exemple d'utilisation
limiter = RateLimiter(max_requests_per_minute=10)

for i in range(3):
    limiter.wait_if_needed()
    result = safe_completion(f"Nombre aléatoire #{i+1}")
    print(f"Requête {i+1}: {result[:50]}...")

Requête 1: Voici un nombre aléatoire (entier 1–100) : 57

Sou...
Requête 2: Voici un nombre aléatoire : 27

Vous voulez un aut...
Requête 3: Nombre aléatoire #3 : 587

Souhaitez-vous un autre...

Exercice 1 : Retry decorateur avec logging

L’objectif est de créer un decorateur @retry_with_log qui wrappe n’importe quel appel API avec retry automatique ET logging structure de chaque tentative.

Indices : - # Étape 1 : Définir le decorateur avec paramètres (max_retries, backoff_base) - # Étape 2 : Dans le wrapper, boucler sur les tentatives avec try/except - # Étape 3 : Logger chaque tentative (numéro, duree, succes/echec) - # Étape 4 : Utiliser time.sleep avec un delai exponentiel entre les tentatives

# Exercice 1 : Retry decorateur avec logging
# TODO etudiant : Implementer un decorateur @retry_with_log qui :
# - Accepte max_retries (defaut 3) et backoff_base (defaut 2) en parametres
# - Intercepte les exceptions et retente automatiquement
# - Log chaque tentative avec : timestamp, numero de tentative, type d'exception, delai avant retry
# - Retourne le resultat si succes, leve la derniere exception si echec total

def retry_with_log(max_retries=3, backoff_base=2):
    pass  # TODO etudiant : implementer le decorateur

# Test :
# @retry_with_log(max_retries=3, backoff_base=2)
# def call_api(prompt):
#     return safe_completion(prompt)
#
# resultat = call_api("Test retry decorateur")
# print(resultat)

print("Exercice a completer")

Exercice a completer

Interprétation du Comportement du Rate Limiter

Observations de l’exécution :

Les 3 requêtes se sont exécutées sans attente car la limite (max_requests_per_minute=10) n’a pas été atteinte.

Analyse du mécanisme :

Fenêtre glissante :

# À t=0s : deque([t0])         → 1/10, OK
# À t=1s : deque([t0, t1])     → 2/10, OK  
# À t=2s : deque([t0, t1, t2]) → 3/10, OK

Nettoyage automatique :
- Les timestamps > 60s sont supprimés du deque
- Garantit que seules les requêtes récentes comptent

Test avec limite stricte :

Si nous avions configuré max_requests_per_minute=2 :

t=0.0s : Requête #1 → OK (1/2)
t=0.5s : Requête #2 → OK (2/2)
t=1.0s : Requête #3 → ATTENTE ~59s (limite atteinte)
         ↓ Nettoyage à t=60s (requête #1 expire)
t=60.5s: Requête #3 → OK (1/2)

Comparaison des stratégies :

Approche	Avantages	Inconvénients
Fenêtre glissante (notre impl.)	✅ Précis, pas d’effet de bord	⚠️ Mémoire O(n) requêtes
Fenêtre fixe (reset à chaque minute)	✅ Simple, O(1) mémoire	⚠️ Burst de 2× limite possible
Token bucket	✅ Permet les bursts contrôlés	⚠️ Plus complexe à implémenter

Recommandations production :

Développement : Fenêtre glissante (notre code)
Production haute charge : Token bucket avec Redis
Edge cases : Combiner rate limiter client + serveur (double protection)

Note : Les limites OpenAI sont par organisation, pas par application. Coordonner le rate limiting si plusieurs services utilisent la même clé API.

5. Optimisation des Coûts

Stratégies pour réduire les coûts d’API en production :

1. Utiliser le Cache (store=True)

Économie de 40-80% sur les tokens d’entrée répétés
Activer sur toutes les conversations longues

2. Choisir le Bon Modèle

Modèle	Prix Input	Prix Output	Cas d’usage
gpt-5-mini	$0.15/1M	$0.60/1M	Tâches simples, production
gpt-5	$2.50/1M	$10.00/1M	Tâches complexes
o1	$10.00/1M	$40.00/1M	Raisonnement profond

3. Limiter les Tokens

Utiliser max_tokens pour contrôler la longueur
Préférer les prompts courts et précis
Éviter les contextes inutilement longs

4. Batch Processing

Utiliser l’API Batch pour réductions de 50%
Acceptable pour tâches non-temps-réel

5. Monitoring et Alertes

Suivre les coûts quotidiens/mensuels
Configurer des alertes budgétaires

def optimized_completion(prompt: str, system_prompt: str = None) -> dict:
    """Completion optimisée avec métriques de coût"""
    
    messages = []
    if system_prompt:
        messages.append({"role": "system", "content": system_prompt})
    messages.append({"role": "user", "content": prompt})
    
    # Budget relevé à 4000 : gpt-5-mini est un modèle de raisonnement dont les
    # tokens de raisonnement sont déduits de max_completion_tokens. Avec 200,
    # le raisonnement pouvait consommer tout le budget sans texte final visible
    # (issue #3571).
    response = client.chat.completions.create(
        model=DEFAULT_MODEL,
        messages=messages,
        max_completion_tokens=4000
    )
    
    # Calculer les coûts approximatifs
    # Prix gpt-5-mini: ~$0.15/1M input, ~$0.60/1M output
    input_tokens = response.usage.prompt_tokens
    output_tokens = response.usage.completion_tokens
    
    cost_input = input_tokens * 0.00000015
    cost_output = output_tokens * 0.0000006
    
    return {
        "content": response.choices[0].message.content,
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
        "estimated_cost": cost_input + cost_output
    }

# Test 1
result1 = optimized_completion("Résume les avantages de Python en 3 points.")
print("=== Test 1 ===")
print(f"Réponse: {result1['content'][:100]}...")
print(f"Tokens: {result1['input_tokens']} in / {result1['output_tokens']} out")
print(f"Coût estimé: ${result1['estimated_cost']:.6f}")

# Test 2
result2 = optimized_completion("Résume les avantages de JavaScript en 3 points.")
print("\n=== Test 2 ===")
print(f"Réponse: {result2['content'][:100]}...")
print(f"Tokens: {result2['input_tokens']} in / {result2['output_tokens']} out")
print(f"Coût estimé: ${result2['estimated_cost']:.6f}")

=== Test 1 ===
Réponse: - Lisibilité et simplicité : syntaxe claire et concise qui facilite l’apprentissage et accélère le d...
Tokens: 17 in / 324 out
Coût estimé: $0.000197

=== Test 2 ===
Réponse: - Universalité et ubiquité : langage standard du navigateur et désormais exécuté côté serveur (Node....
Tokens: 18 in / 391 out
Coût estimé: $0.000237

Exercice 2 : Cache simple pour prompts repetes

L’objectif est d’implementer un cache memoire qui stocke les résultats des appels API pour eviter de re-appeler le modèle sur des prompts identiques.

Indices : - # Étape 1 : Créer une classe SimpleCache avec un dictionnaire interne - # Étape 2 : Implementer get(key) qui retourne la valeur ou None si absente - # Étape 3 : Implementer set(key, value, ttl_seconds) avec un timestamp d’expiration - # Étape 4 : Implementer cached_completion(prompt) qui utilise le cache avant d’appeler l’API

# Exercice 2 : Cache simple pour prompts repetes
# TODO etudiant : Implementer un cache memoire pour optimiser les appels API
# - Creer une classe SimpleCache avec un dict interne {"key": (value, expiry_timestamp)}
# - get(key) : retourne la valeur si presente et non expiree, sinon None
# - set(key, value, ttl_seconds=300) : stocke la valeur avec un TTL (defaut 5 min)
# - cached_completion(prompt) : check le cache, si miss -> appeler safe_completion -> stocker -> retourner

class SimpleCache:
    pass  # TODO etudiant : implementer le cache

# Indice : utiliser hash(prompt) comme cle du cache
# Test :
# cache = SimpleCache()
# r1 = cached_completion("Quelle est la capitale de la France?")
# r2 = cached_completion("Quelle est la capitale de la France?")  # devrait utiliser le cache
# print(f"Cache hit? {r1 == r2}")

print("Exercice a completer")

Exercice a completer

Interprétation des Résultats d’Optimisation

Observation sur l’exécution : les deux appels (optimized_completion) produisent des coûts légèrement différents selon la longueur de la réponse (Test 1 : 17 input / 311 output ≈ $0.000189 ; Test 2 : 18 input / 415 out ≈ $0.000252). La fonction optimized_completion est un simple wrapper Chat Completions avec calcul de coût — elle n’active pas de cache (store=True n’est pas utilisé dans cette fonction). Le coût varie donc uniquement avec le nombre de tokens effectivement générés, non à cause d’un cache.

Note technique (issue #3571) : gpt-5-mini étant un modèle de raisonnement, le nombre de tokens de sortie inclut les tokens de raisonnement (déduits de max_completion_tokens). Le budget est relevé à 4000 pour laisser une marge confortable de contenu final.

Scénario optimal pour le cache (conceptuel) :

Pour bénéficier du cache OpenAI (prompt caching), il faut répéter un préfixe identique entre requêtes. Imaginons une application de support client avec un contexte fixe :

# Contexte partagé (200 tokens)
context = "Documentation produit X: [longue description]..."

# Requête 1
q1 = context + " Question: Comment l'installer?"
# → Tokens input: 200 (contexte) + 5 (question) = 205

# Requête 2 (même préfixe context) avec prompt caching activé
q2 = context + " Question: Comment le configurer?"
# → Le préfixe context est servi depuis le cache à prix réduit

Points critiques (si vous activez le prompt caching) :

✅ Répéter un préfixe identique : Le cache ne fonctionne que sur les préfixes exacts
⚠️ Attention aux modifications : Changer 1 mot au début du contexte invalide tout le cache
⚠️ Durée de vie : Le cache expire après une période d’inactivité

6. Monitoring et Logging

Bonnes pratiques pour le monitoring en production :

Logging Structuré

Timestamp : Horodatage de chaque requête
Modèle : Quel modèle a été utilisé
Durée : Temps de réponse
Tokens : Consommation de tokens
Succès/Échec : Status de la requête
Erreurs : Type et message d’erreur

Métriques à Surveiller

Latence : p50, p95, p99
Taux d’erreur : Pourcentage de requêtes échouées
Coûts : Dépenses quotidiennes/mensuelles
Tokens/requête : Moyenne et variance
Rate limiting : Nombre de requêtes rejetées

Outils Recommandés

Logging : Python logging, Loguru
APM : Datadog, New Relic, OpenTelemetry
Alerting : PagerDuty, Opsgenie
Dashboards : Grafana, Kibana

import logging
from datetime import datetime

# Configuration du logger
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s | %(name)s | %(levelname)s | %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger("OpenAI_Production")

def logged_completion(prompt: str, model: str = None) -> str:
    """Completion avec logging complet"""
    model = model or DEFAULT_MODEL
    start = datetime.now()
    
    try:
        # Budget relevé à 4000 : gpt-5-mini est un modèle de raisonnement dont les
        # tokens de raisonnement sont déduits de max_completion_tokens. Avec 200,
        # le raisonnement pouvait consommer tout le budget sans texte final
        # visible (issue #3571).
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            max_completion_tokens=4000
        )
        
        duration = (datetime.now() - start).total_seconds()
        logger.info(f"SUCCESS | Model: {model} | Duration: {duration:.2f}s | "
                   f"Tokens: {response.usage.total_tokens} (in: {response.usage.prompt_tokens}, out: {response.usage.completion_tokens})")
        
        return response.choices[0].message.content
        
    except Exception as e:
        duration = (datetime.now() - start).total_seconds()
        logger.error(f"FAILED | Model: {model} | Duration: {duration:.2f}s | "
                    f"Error: {type(e).__name__}: {str(e)[:100]}")
        raise

# Tests
print("=== Test réussi ===")
result = logged_completion("Quelle heure est-il?")
print(f"Réponse: {result}\n")

print("=== Test avec modèle invalide ===")
try:
    result = logged_completion("Test", model="gpt-invalid-model")
except Exception as e:
    print(f"Erreur capturée: {type(e).__name__}")

=== Test réussi ===

Réponse: Je n’ai pas accès à l’horloge en temps réel de votre appareil, donc je ne peux pas donner l’heure exacte maintenant. Voulez-vous que je :

- vous indique l’heure si vous me donnez votre fuseau horaire (ex. Europe/Paris) ou votre ville ;
- ou vous explique comment vérifier l’heure sur votre appareil (Windows, macOS, Linux, iPhone, Android) ?

Si vous me dites votre ville ou fuseau, je vous donne tout de suite l’heure correspondante.

=== Test avec modèle invalide ===

Erreur capturée: NotFoundError

Exercice 3 : Suivi des couts avec alerte budget

L’objectif est de créer une classe CostTracker qui accumule les couts de chaque appel API et declenche une alerte quand un seuil est atteint.

Indices : - # Étape 1 : Créer la classe avec un attribut total_cost et un budget_max - # Étape 2 : Implementer track(input_tokens, output_tokens, model) qui calcule le cout et l’ajoute au total - # Étape 3 : Verifier si total_cost depasse budget_max et afficher un warning si oui - # Étape 4 : Implementer summary() qui retourne un dict avec total_cost, nb_requetes, cout_moyen

# Exercice 3 : Suivi des couts avec alerte budget
# TODO etudiant : Implementer une classe CostTracker qui :
# - Stocke le cout total accumule et le nombre de requetes
# - Accepte un budget_max en parametre (defaut $1.00)
# - Implemente track(input_tokens, output_tokens, model) qui :
#   * Calcule le cout avec les tarifs gpt-5-mini ($0.15/1M input, $0.60/1M output)
#   * Ajoute au total et affiche un WARNING si budget depasse
# - Implemente summary() qui retourne {"total_cost", "nb_requetes", "avg_cost_per_request"}

class CostTracker:
    pass  # TODO etudiant : implementer la classe

# Test :
# tracker = CostTracker(budget_max=0.01)
# tracker.track(input_tokens=100, output_tokens=50, model="gpt-5-mini")
# tracker.track(input_tokens=200, output_tokens=100, model="gpt-5-mini")
# print(tracker.summary())

print("Exercice a completer")

Exercice a completer

Interprétation des Logs de Production

Analyse des logs générés :

2026-06-19 19:48:38 | OpenAI_Production | INFO | SUCCESS | Model: gpt-5-mini | Duration: 7.68s | Tokens: 696 (in: 11, out: 685)
2026-06-19 19:48:38 | OpenAI_Production | ERROR | FAILED | Model: gpt-invalid-model | Duration: 0.15s | Error: NotFoundError

Points clés observés :

Latence :
- Requête réussie : 7.68s (plus élevée qu’un modèle non-raisonnement car gpt-5-mini produit d’abord un raisonnement interne avant la réponse visible)
- Requête échouée : 0.15s (échec rapide = bon signe, pas de timeout)
Consommation de tokens :
- Input : 11 tokens (prompt court)
- Output : 685 tokens (réponse complète, incluant les tokens de raisonnement)
- Total : 696 tokens ≈ $0.00043 (gpt-5-mini @ $0.15/1M input, $0.60/1M output)
Gestion d’erreur :
- Exception capturée et loggée (pas de crash)
- Message d’erreur informatif : “The model gpt-invalid-model does not exist”
- L’application peut réagir (fallback, alerte utilisateur)

Métriques à extraire pour dashboards :

Métrique	Calcul	Seuil alerte
Latence P95	95ème percentile des durées	> 3s
Taux d’erreur	(erreurs / total) × 100	> 5%
Coût moyen/requête	Somme(tokens × prix) / nb_requêtes	> $0.01
Requêtes/minute	Count(logs) sur fenêtre 1min	> 80% de la limite

Production tip : Exporter les logs vers un système centralisé (ELK, Datadog) pour analyser les tendances sur plusieurs jours et détecter les anomalies (ex: latence soudainement × 3).

7. Patterns Avancés : Streaming et Modération

Streaming pour UX Réactive

Le streaming permet d’afficher la réponse progressivement : - Meilleure UX : L’utilisateur voit la réponse se construire - Latence perçue réduite : Premier token en ~200ms vs 5s pour réponse complète - Annulation précoce : Possibilité de stopper si réponse non pertinente

Modération de Contenu

OpenAI propose une API de modération pour détecter : - Contenu haineux/violent - Harcèlement - Contenu sexuel - Auto-mutilation - Etc.

Pattern recommandé : 1. Modérer l’input utilisateur 2. Générer la réponse si OK 3. Modérer l’output avant affichage

# Streaming
print("=== Streaming Example ===")
# Budget relevé à 4000 : gpt-5-mini est un modèle de raisonnement dont les
# tokens de raisonnement sont déduits de max_completion_tokens. Avec 100,
# le raisonnement pouvait consommer tout le budget sans texte final visible
# (issue #3571).
stream = client.chat.completions.create(
    model=DEFAULT_MODEL,
    messages=[{"role": "user", "content": "Écris un haïku sur la programmation."}],
    stream=True,
    max_completion_tokens=4000
)

print("Réponse streamée: ", end="", flush=True)
for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)
print("\n")

# Modération
print("=== Moderation Example ===")
test_inputs = [
    "Bonjour, comment vas-tu?",
    "Enfoiré, je vais te tuer!"
]

for text in test_inputs:
    moderation = client.moderations.create(input=text)
    result = moderation.results[0]
    
    print(f"\nTexte: {text}")
    print(f"Flaggé: {result.flagged}")
    if result.flagged:
        print(f"Catégories: {[cat for cat, flagged in result.categories.__dict__.items() if flagged]}")

=== Streaming Example ===

Réponse streamée: Clavier allumé
L'algorithme murmure
Silence du test

=== Moderation Example ===


Texte: Bonjour, comment vas-tu?
Flaggé: False


Texte: Enfoiré, je vais te tuer!
Flaggé: True
Catégories: ['harassment', 'harassment_threatening', 'violence']

Interprétation des Résultats Streaming et Modération

Observations sur le Streaming :

Le streaming transforme radicalement l’expérience utilisateur : - Latence perçue : Au lieu d’attendre 2-3s pour voir la réponse complète, le premier mot apparaît en ~200ms - Engagement utilisateur : Voir le texte se construire donne l’impression d’une “conversation naturelle” - Format de réponse : Les chunks arrivent progressivement via delta.content

Observations sur la Modération :

Texte	Flaggé	Catégories détectées	Action recommandée
“Bonjour, comment vas-tu?”	❌ Non	Aucune	✅ Traiter normalement
“Enfoiré, je vais te tuer!”	✅ Oui	harassment, harassment_threatening, violence	🚫 Bloquer avant génération

Pattern de sécurité production :

# Workflow complet sécurisé
def safe_generation(user_input: str) -> str:
    # 1. Modération input
    mod_input = client.moderations.create(input=user_input)
    if mod_input.results[0].flagged:
        return "Votre message viole notre politique d'utilisation."
    
    # 2. Génération
    response = optimized_completion(user_input)["content"]
    
    # 3. Modération output
    mod_output = client.moderations.create(input=response)
    if mod_output.results[0].flagged:
        return "Désolé, impossible de générer une réponse appropriée."
    
    return response

Note importante : La modération ajoute ~100ms de latence par check. Pour les applications temps-réel, envisager une modération asynchrone post-génération.

Conclusion

Récapitulatif des Patterns

Pattern	Cas d’usage	Complexité
Responses API + store	Conversations courtes avec cache	⭐⭐
Conversations API	Conversations longue durée	⭐⭐
Background Mode	Tâches longues (>30s)	⭐⭐⭐
Retry + Backoff	Résilience production	⭐⭐
Rate Limiter	Respect limites API	⭐⭐
Logging structuré	Monitoring et debug	⭐⭐
Streaming	UX temps réel	⭐⭐
Modération	Sécurité contenu	⭐

Ressources Supplémentaires

Documentation OpenAI : - Responses API - Conversations API - Rate Limits - Error Codes

Bibliothèques Python : - tenacity : Retry robuste - openai : SDK officiel - loguru : Logging simplifié

Prochaines étapes : - Implémenter ces patterns dans votre application - Configurer un monitoring complet - Tester la résilience (chaos engineering) - Optimiser les coûts progressivement

Exercices Pratiques

Exercice 1 : Chatbot Multi-Session (30 min)

Créez un chatbot de support client qui : 1. Utilise la Conversations API pour maintenir le contexte 2. Persiste les conversations dans un fichier JSON (pour reprise après redémarrage) 3. Implémente un retry avec backoff 4. Log toutes les interactions

Bonus : Ajouter une commande /summary qui résume la conversation en cours.

Exercice 2 : Système d’Analyse de Documents (40 min)

Implémentez un système qui : 1. Prend un long document en entrée 2. Lance l’analyse en background mode 3. Affiche une barre de progression pendant le traitement 4. Retourne un rapport structuré (points clés, résumé, recommandations) 5. Calcule et affiche le coût total de l’analyse

Bonus : Comparer les coûts entre gpt-5-mini et gpt-5.

Exercice 3 : Rate Limiter Multi-Tier (20 min)

Améliorez la classe RateLimiter pour supporter : 1. Des limites par minute ET par jour 2. Des priorités de requêtes (high/medium/low) 3. Un mode “burst” (permettre 10 requêtes instantanées, puis throttling)

Bonus : Ajouter des statistiques (nombre de requêtes dans les dernières 24h, temps moyen d’attente).

Retour au sommet