Créer un chatbot IA avec une seule clé API : de zéro à la production en 30 minutes
Ce tutoriel construit un backend de chatbot IA prêt pour la production avec des réponses en streaming, un historique de conversation, le changement de modèle et une gestion correcte des erreurs. Nous utiliserons Python, FastAPI et le SDK OpenAI pointé vers un agrégateur d’API afin que vous puissiez utiliser n’importe quel modèle.
Prérequis
pip install fastapi uvicorn openai
Étape 1 : Endpoint de chat basique
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import OpenAI
from pydantic import BaseModel
app = FastAPI()
client = OpenAI(
api_key="sk-lemon-xxx",
base_url="https://api.lemondata.cc/v1"
)
class ChatRequest(BaseModel):
message: str
model: str = "gpt-4.1-mini"
conversation_id: str | None = None
@app.post("/chat")
async def chat(req: ChatRequest):
response = client.chat.completions.create(
model=req.model,
messages=[{"role": "user", "content": req.message}]
)
return {"reply": response.choices[0].message.content}
Cela fonctionne mais sans streaming, sans historique et sans gestion des erreurs. Corrigeons cela.
Étape 2 : Ajouter le streaming
Le streaming envoie les tokens au fur et à mesure de leur génération au lieu d’attendre la réponse complète. Les utilisateurs voient la réponse se former en temps réel.
@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
def generate():
stream = client.chat.completions.create(
model=req.model,
messages=[{"role": "user", "content": req.message}],
stream=True
)
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
yield f"data: {delta.content}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(
generate(),
media_type="text/event-stream"
)
Étape 3 : Historique de conversation
Stockez l’historique des conversations en mémoire (remplacez par Redis ou une base de données en production).
from collections import defaultdict
import uuid
conversations: dict[str, list] = defaultdict(list)
SYSTEM_PROMPT = "Vous êtes un assistant utile. Soyez concis et direct."
@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
conv_id = req.conversation_id or str(uuid.uuid4())
# Construire l’historique des messages
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
messages.extend(conversations[conv_id])
messages.append({"role": "user", "content": req.message})
# Stocker le message utilisateur
conversations[conv_id].append(
{"role": "user", "content": req.message}
)
def generate():
full_response = []
stream = client.chat.completions.create(
model=req.model,
messages=messages,
stream=True
)
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
full_response.append(delta.content)
yield f"data: {delta.content}\n\n"
# Stocker la réponse de l’assistant
conversations[conv_id].append(
{"role": "assistant", "content": "".join(full_response)}
)
yield f"data: [DONE]\n\n"
return StreamingResponse(
generate(),
media_type="text/event-stream",
headers={"X-Conversation-ID": conv_id}
)
Étape 4 : Gestion des erreurs
Les appels à l’API IA peuvent échouer pour plusieurs raisons : limites de taux, solde insuffisant, modèle indisponible. Gérez chaque cas :
from openai import (
APIError,
RateLimitError,
APIConnectionError
)
@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
conv_id = req.conversation_id or str(uuid.uuid4())
messages = build_messages(conv_id, req.message)
def generate():
try:
full_response = []
stream = client.chat.completions.create(
model=req.model,
messages=messages,
stream=True
)
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
full_response.append(delta.content)
yield f"data: {delta.content}\n\n"
conversations[conv_id].append(
{"role": "assistant", "content": "".join(full_response)}
)
except RateLimitError as e:
yield f"data: [ERROR] Limite de débit atteinte. Veuillez patienter un instant.\n\n"
except APIConnectionError:
yield f"data: [ERROR] Échec de la connexion. Nouvelle tentative...\n\n"
except APIError as e:
yield f"data: [ERROR] {e.message}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")
def build_messages(conv_id: str, user_msg: str) -> list:
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
# Conserver les 10 derniers tours pour gérer la longueur du contexte
history = conversations[conv_id][-20:]
messages.extend(history)
messages.append({"role": "user", "content": user_msg})
conversations[conv_id].append({"role": "user", "content": user_msg})
return messages
Étape 5 : Changement de modèle
Laissez les utilisateurs changer de modèle en cours de conversation. Différents modèles pour différents besoins :
AVAILABLE_MODELS = {
"fast": "gpt-4.1-mini",
"smart": "claude-sonnet-4-6",
"reasoning": "o3",
"budget": "deepseek-chat",
"creative": "claude-sonnet-4-6",
}
@app.get("/models")
async def list_models():
return {"models": AVAILABLE_MODELS}
Le frontend peut présenter ces options. Puisque tous les modèles utilisent le même format compatible OpenAI via l’agrégateur, changer de modèle revient simplement à modifier le paramètre model.
Étape 6 : Gestion de la fenêtre de contexte
Les longues conversations dépassent les limites de contexte du modèle. Implémentez une fenêtre glissante :
def trim_history(messages: list, max_tokens: int = 8000) -> list:
"""Conserver le prompt système + les messages récents dans le budget de tokens."""
# Estimation approximative : 1 token ≈ 4 caractères
system = messages[0] # Toujours conserver le prompt système
history = messages[1:]
total_chars = len(system["content"])
trimmed = []
for msg in reversed(history):
msg_chars = len(msg["content"])
if total_chars + msg_chars > max_tokens * 4:
break
trimmed.insert(0, msg)
total_chars += msg_chars
return [system] + trimmed
Application complète
# Lancer avec : uvicorn main:app --reload --port 8000
# Tester : curl -N -X POST http://localhost:8000/chat/stream \
# -H "Content-Type: application/json" \
# -d '{"message": "Bonjour !", "model": "gpt-4.1-mini"}'
Le code complet fait moins de 100 lignes. À partir d’ici, vous pouvez ajouter :
- Authentification (clés API ou JWT)
- Stockage persistant (PostgreSQL ou Redis pour les conversations)
- Limitation de débit par utilisateur
- Suivi d’utilisation et facturation
- Support WebSocket pour streaming bidirectionnel
- Frontend (React, Vue ou JS vanilla avec EventSource)
Estimation des coûts
Pour un chatbot gérant 1 000 conversations/jour (en moyenne 5 tours chacune) :
| Modèle | Coût journalier | Coût mensuel |
|---|---|---|
| GPT-4.1-mini | ~2,40 $ | ~72 $ |
| GPT-4.1 | ~12,00 $ | ~360 $ |
| Claude Sonnet 4.6 | ~18,00 $ | ~540 $ |
| DeepSeek V3 | ~1,68 $ | ~50 $ |
Utiliser GPT-4.1-mini pour la plupart des conversations et passer à Claude Sonnet 4.6 uniquement sur demande permet de maintenir les coûts sous 100 $/mois pour la majorité des applications.
Obtenez votre clé API : lemondata.cc propose plus de 300 modèles via un seul endpoint. 1 $ de crédit gratuit pour commencer à construire.