Ce tutoriel permet de créer un service de chatbot simple mais prêt pour la production avec FastAPI, le streaming SSE, la mémoire de conversation et le changement de modèle. L'objectif n'est pas de faire une démonstration gadget. L'objectif est de créer un backend que vous pouvez placer derrière une interface produit réelle et sur lequel vous pouvez itérer en toute sécurité, sans avoir à réécrire votre intégration à chaque changement de modèle.
Si vous avez déjà configuré un SDK compatible OpenAI avec TokenLab, cet article prend le relais à partir de là. Si vous n'avez pas encore effectué le changement d'URL de base, lisez d'abord le guide de migration. Si votre préoccupation principale est le formatage des requêtes et le backoff sous charge, associez ce guide au guide de limitation de débit de l'API IA.
Points clés à retenir
- Un chatbot prêt pour la production nécessite six éléments : un endpoint synchrone, un endpoint de streaming, un état de conversation côté serveur, une liste blanche de modèles, une gestion réelle des erreurs et un chemin de mise à niveau clair pour le stockage.
- Validez votre clé, votre URL de base et votre routage avec un petit endpoint de chat avant d'ajouter le streaming, la mémoire ou les appels d'outils.
- Le streaming SSE couvre la plupart des produits de chat et comporte moins de frais opérationnels que les websockets.
- Exposez les modèles via une liste blanche backend, et non un champ de texte libre, afin que le frontend ne puisse pas demander des IDs de modèles arbitraires ou obsolètes.
- La disponibilité et la gamme des modèles changent souvent. Consultez le répertoire des modèles de TokenLab (observé le 07/07/2026) avant de verrouiller votre liste blanche en production.
Ce que nous construisons
Le service final comporte six parties mobiles :
- Un endpoint
/chatsynchrone pour les tests de base. - Un endpoint
/chat/streamen streaming pour l'interface utilisateur réelle. - Un état de conversation identifié par
conversation_id. - Une liste blanche de modèles pour empêcher le frontend de demander des IDs arbitraires.
- Une gestion des erreurs qui ne s'effondre pas à la première erreur 429.
- Un chemin clair du prototype en mémoire vers Redis ou PostgreSQL.
C'est suffisant pour alimenter un bot de support, un assistant interne ou la première version d'un widget de chat intégré.
Installer la stack minimale
pip install fastapi uvicorn openai pydantic redis
Vous pouvez ignorer redis pour la première étape, mais intégrer l'importation dès maintenant fait de la mise à niveau ultérieure une simple formalité plutôt qu'une refonte.
Étape 1 : Commencer avec un petit endpoint de chat basique
La façon la plus rapide de se perdre dans le développement d'un chatbot est de commencer par les websockets, l'utilisation d'outils et l'orchestration d'agents avant que le chemin de requête de base ne soit stable. Commencez par un petit endpoint qui prouve que votre clé, votre URL de base et votre routage de modèle fonctionnent.
from fastapi import FastAPI
from openai import OpenAI
from pydantic import BaseModel
app = FastAPI()
client = OpenAI(
api_key="sk-tokenlab-xxx",
base_url="https://api.tokenlab.sh/v1"
)
class ChatRequest(BaseModel):
message: str
model: str = "deepseek-v4-flash"
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}
Effectuez un test de base. Si cela échoue, ne continuez pas à construire par-dessus.
Étape 2 : Ajouter le streaming car les utilisateurs ressentent la latence avant de la mesurer
La plupart des produits de chatbot semblent lents, non pas parce que le modèle est lent, mais parce que l'interface reste vide jusqu'à ce que la réponse complète arrive. Le SSE est suffisant pour la plupart des produits de chat et a une charge opérationnelle plus faible que les websockets.
from fastapi.responses import StreamingResponse
@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")
Sur le frontend, un simple lecteur fetch suffit toujours :
async function sendMessage(payload) {
const response = await fetch('/chat/stream', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(payload),
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
while (true) {
const { done, value } = await reader.read();
if (done) break;
const chunk = decoder.decode(value, { stream: true });
console.log(chunk);
}
}
Si votre produit exécute déjà un client navigateur via HTTP standard, le SSE maintient l'architecture plus simple qu'une couche websocket.
Étape 3 : Déplacer l'état de la conversation hors du corps de la requête
La première démo de chatbot conserve généralement la transcription complète dans le navigateur et la renvoie à chaque tour. Cela fonctionne pour les prototypes. Cela devient vite compliqué dès que vous avez besoin de tentatives, de sessions reprenables ou d'outils côté serveur.
Un stockage en mémoire est suffisant pour commencer :
from collections import defaultdict
import uuid
conversations: dict[str, list] = defaultdict(list)
SYSTEM_PROMPT = "You are a helpful assistant. Be concise and direct."
def build_messages(conv_id: str, user_msg: str) -> list:
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
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
Le chemin de mise à niveau vers Redis concerne principalement la plomberie du stockage, pas les changements de logique :
import json
import redis
redis_client = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
def load_history(conv_id: str) -> list:
raw = redis_client.get(f"chat:{conv_id}")
return json.loads(raw) if raw else []
def save_history(conv_id: str, history: list) -> None:
redis_client.setex(f"chat:{conv_id}", 60 * 60 * 24, json.dumps(history))
Utilisez Redis lorsque les conversations ont besoin de TTL, de reprenabilité ou d'un déploiement multi-instances. Utilisez PostgreSQL lorsque la transcription elle-même est une donnée produit que vous devez interroger, auditer ou rapporter plus tard.
Étape 4 : Traiter les erreurs comme un comportement produit, pas seulement comme des exceptions
Si votre chatbot est destiné aux clients, le chemin d'échec compte autant que le chemin nominal. Un utilisateur ne se soucie pas de savoir si l'échec provient d'une limitation de débit, d'un solde épuisé ou d'une panne de modèle en amont. Il se soucie de savoir si l'interface se fige.
from openai import APIConnectionError, APIError, RateLimitError
@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():
full_response = []
try:
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"
except RateLimitError:
yield "data: [error: rate limited, please retry shortly]\n\n"
except APIConnectionError:
yield "data: [error: connection issue, please retry]\n\n"
except APIError:
yield "data: [error: something went wrong on our end]\n\n"
finally:
if full_response:
conversations[conv_id].append(
{"role": "assistant", "content": "".join(full_response)}
)
yield "data: [DONE]\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")
Étape 5 : Verrouiller les modèles que le frontend peut demander
Ne laissez jamais le frontend transmettre une chaîne de modèle arbitraire directement à l'API. Un champ de texte libre invite à demander des modèles obsolètes, des fautes de frappe ou des modèles que vous n'aviez jamais l'intention d'exposer. Routez plutôt via une liste blanche backend.
ALLOWED_MODELS = {
"default": "deepseek-v4-flash",
"flagship": "gpt-5.5",
"balanced": "claude-sonnet-5",
}
class ChatRequest(BaseModel):
message: str
tier: str = "default"
conversation_id: str | None = None
@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
model = ALLOWED_MODELS.get(req.tier, ALLOWED_MODELS["default"])
# le reste de la logique de streaming utilise `model` au lieu d'une chaîne brute fournie par le client
Cela vous donne un endroit unique pour changer de modèle lorsqu'un fournisseur en déprécie un, sans toucher au code frontend ni publier une mise à jour client.
Étape 6 : Gérer le reste de la production, pas seulement le chemin nominal
Un backend de chatbot est considéré comme prêt pour la production lorsque les aspects périphériques sont gérés, et non lorsque l'appel de chat principal devient intelligent.
La liste de contrôle est courte :
- ajoutez des IDs de requête pour pouvoir connecter les échecs frontend aux logs backend
- limitez la concurrence par utilisateur et la taille des requêtes
- élaguez les historiques longs avant qu'ils n'explosent votre budget de tokens
- loggez le modèle, la latence, la taille d'entrée et la raison de la fin
- séparez les messages d'erreur visibles par l'utilisateur des détails d'erreur internes
- testez un modèle alternatif pour savoir que le repli fonctionne avant la première panne réelle
L'élagage de l'historique peut rester simple :
def trim_history(messages: list, max_tokens: int = 8000) -> list:
system = messages[0]
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
Le but n'est pas une comptabilité parfaite des tokens. Le but est d'arrêter les explosions de contexte évidentes avant qu'elles n'affectent votre facture ou votre latence.
De la démo au produit
Une fois ce backend stable, la prochaine mise à niveau est rarement « plus d'IA ». C'est généralement de l'infrastructure ennuyeuse :
- l'authentification pour qu'un utilisateur ne puisse pas lire la conversation d'un autre
- la persistance pour que les sessions survivent aux déploiements
- la limitation de débit pour qu'un utilisateur bruyant ne puisse pas épuiser votre quota
- la facturation ou l'attribution d'utilisation si le chatbot est destiné aux clients
- la synthèse en arrière-plan si les conversations nécessitent une mémoire à long terme
Une passerelle unifiée aide pour la plupart de ces points. Une fois la migration de l'URL de base effectuée, les changements de modèle cessent d'être une réécriture de plateforme pour devenir une modification de configuration.
Test de base
uvicorn main:app --reload --port 8000
curl -N -X POST http://localhost:8000/chat/stream \
-H "Content-Type: application/json" \
-d '{"message": "Hello!", "model": "deepseek-v4-flash"}'
Si vous pouvez streamer un tour, préserver une conversation et renvoyer une erreur propre lors d'une défaillance forcée, vous avez la bonne base.
Estimation des coûts
Créez une clé API sur TokenLab, pointez votre SDK OpenAI vers https://api.tokenlab.sh/v1, et vous pourrez expédier la première version de production de votre chatbot sans gérer de comptes séparés chez différents fournisseurs.
| Modèle | Niveau typique | Notes |
|---|---|---|
| DeepSeek V4 Flash | Rapide / par défaut | Bon choix par défaut pour les tours à haut volume et faible latence |
| GPT-5.5 | Flagship | À utiliser pour les tours nécessitant un raisonnement plus fort |
| Claude Sonnet 5 | Équilibré | Excellent choix pour le codage et les réponses de type révision |
| Gemini 3.5 Flash | Budget / alt rapide | Alternative rapide et peu coûteuse pour le routage à haut volume |
La tarification exacte par token change fréquemment selon les fournisseurs et n'est pas reproduite ici sous forme de chiffres fixes. Vérifiez les tarifs actuels sur le répertoire des modèles (observé le 07/07/2026) avant d'établir votre budget. En pratique, router la plupart des conversations vers un niveau rapide/par défaut comme DeepSeek V4 Flash ou Gemini 3.5 Flash et réserver GPT-5.5 ou Claude Sonnet 5 pour les tours qui en ont besoin maintient la plupart des applications sur une facture mensuelle basse, mais confirmez les tarifs réels par million de tokens pour votre compte avant de vous engager sur un budget.
FAQ
Ai-je besoin de websockets pour construire un chatbot IA ? Non. Le streaming SSE, présenté à l'étape 2, couvre la grande majorité des produits de chat. Les websockets apportent une réelle valeur ajoutée lorsque vous avez besoin d'un push bidirectionnel en dehors du cycle requête/réponse, comme pour la collaboration en direct ou les événements initiés par le serveur. Pour une interface de chat standard, le SSE est plus simple à déployer, à déboguer et à mettre à l'échelle.
Comment savoir quel modèle utiliser par défaut ? Commencez avec un modèle rapide et peu coûteux comme DeepSeek V4 Flash ou Gemini 3.5 Flash pour le niveau par défaut, et ajoutez un niveau équilibré ou de raisonnement sur Claude Sonnet 5 ou GPT-5.5 derrière la liste blanche présentée à l'étape 5. Consultez le répertoire des modèles (observé le 07/07/2026) pour les options actuelles, car de nouveaux modèles sortent et les anciens sont dépréciés selon un calendrier hors de votre contrôle.
Qu'est-ce qui casse en premier lorsqu'un chatbot passe de la démo au trafic réel ? Presque toujours le chemin d'erreur, pas le chemin nominal. Les tentatives illimitées, l'absence de limites de concurrence par utilisateur et l'historique de conversation illimité sont les trois causes les plus fréquentes de la chute d'un backend de chatbot sous une charge réelle. Les étapes 4 et 6 ci-dessus traitent ces trois points directement.
Démarrez avec votre clé API : tokenlab.sh fournit plus de 300 modèles via un seul endpoint. 1 $ de crédit gratuit pour commencer à construire.
Sources
Prix observé le 2026-07-07
- TokenLab model directoryObservé le 2026-07-07



