Este tutorial construye un servicio de chatbot pequeño pero listo para producción con FastAPI, streaming SSE, memoria de conversación y cambio de modelos. El objetivo no es una demo de juguete. El objetivo es un backend que puedas colocar detrás de una interfaz de producto real e iterar de forma segura, sin tener que reescribir tu integración cada vez que cambies de modelo.
Si ya has apuntado un SDK compatible con OpenAI a TokenLab, este artículo continúa a partir de ahí. Si aún no has realizado el cambio de base URL, lee primero la guía de migración. Si tu principal preocupación es el formato de las peticiones y el backoff bajo carga, combina esta guía con la guía de limitación de tasa (rate limiting) de la API de IA.
Puntos clave
- Un chatbot listo para producción necesita seis partes: un endpoint síncrono, un endpoint de streaming, estado de conversación en el lado del servidor, una lista de modelos permitidos (allowlist), gestión de errores real y una ruta clara para actualizar el almacenamiento.
- Prueba tu clave, base URL y enrutamiento con un pequeño endpoint de chat antes de añadir streaming, memoria o llamadas a herramientas.
- El streaming SSE cubre la mayoría de los productos de chat y conlleva menos carga operativa que los websockets.
- Expón los modelos a través de una lista de permitidos en el backend, no mediante un campo de texto libre, para que el frontend no pueda solicitar IDs de modelos arbitrarios o retirados.
- La disponibilidad y las alineaciones de modelos cambian a menudo. Consulta el directorio de modelos de TokenLab (observado el 07-07-2026) antes de bloquear tu lista de permitidos en producción.
Lo que vamos a construir
El servicio final tiene seis partes móviles:
- Un endpoint
/chatsíncrono para pruebas de humo. - Un endpoint
/chat/streamde streaming para la interfaz de usuario real. - Estado de conversación identificado por
conversation_id. - Una lista de modelos permitidos para que el frontend no pueda solicitar IDs arbitrarios.
- Gestión de errores que no colapse ante el primer 429.
- Una ruta clara desde el prototipo en memoria hacia Redis o PostgreSQL.
Eso es suficiente para impulsar un bot de soporte, un asistente interno o la primera versión de un widget de chat embebido.
Instala el stack mínimo
pip install fastapi uvicorn openai pydantic redis
Puedes omitir redis en la primera pasada, pero incluir la importación ahora hace que la actualización posterior sea un trámite sencillo en lugar de una refactorización.
Paso 1: Empieza con un endpoint de chat pequeño y sencillo
La forma más rápida de perderse trabajando en un chatbot es empezar con websockets, uso de herramientas y orquestación de agentes antes de que la ruta básica de la petición sea estable. Empieza con un pequeño endpoint que demuestre que tu clave, base URL y enrutamiento de modelos funcionan.
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}
Ejecuta una prueba de humo. Si esto falla, no sigas construyendo sobre ello.
Paso 2: Añade streaming porque los usuarios sienten la latencia antes de medirla
La mayoría de los productos de chatbot se sienten lentos no porque el modelo sea lento, sino porque la interfaz permanece en blanco hasta que llega la respuesta completa. SSE es suficiente para la mayoría de los productos de chat y tiene una carga operativa menor que los 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")
En el frontend, un lector fetch estándar sigue siendo suficiente:
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 tu producto ya ejecuta un cliente de navegador sobre HTTP estándar, SSE mantiene la arquitectura más simple que una capa de websocket.
Paso 3: Saca el estado de la conversación del cuerpo de la petición
La primera demo de un chatbot suele mantener la transcripción completa en el navegador y reenviarla en cada turno. Eso funciona para prototipos. Se vuelve complicado rápidamente una vez que necesitas reintentos, sesiones reanudables o herramientas del lado del servidor.
Un almacenamiento en memoria está bien para empezar:
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
La ruta de actualización a Redis es principalmente fontanería de almacenamiento, no cambios de lógica:
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))
Recurre a Redis cuando las conversaciones necesiten TTL, reanudabilidad o despliegue en múltiples instancias. Recurre a PostgreSQL cuando la transcripción en sí sea un dato del producto que necesites consultar, auditar o sobre el cual generar informes más adelante.
Paso 4: Trata los errores como comportamiento del producto, no solo como excepciones
Si tu chatbot está orientado al cliente, la ruta de fallo importa tanto como la ruta de éxito. A un usuario no le importa si el fallo proviene de una limitación de tasa, un saldo agotado o una interrupción del modelo upstream. Le importa si la interfaz se congela.
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")
Paso 5: Bloquea qué modelos puede solicitar el frontend
Nunca permitas que el frontend pase un string de modelo arbitrario directamente a la API. Un campo de texto libre invita a solicitar modelos retirados, errores tipográficos o modelos que nunca tuviste la intención de exponer. Enruta a través de una lista de permitidos en el 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"])
# el resto de la lógica de streaming usa `model` en lugar de un string crudo suministrado por el cliente
Esto te da un lugar único para cambiar modelos cuando un proveedor retira uno, sin tocar el código del frontend ni enviar una actualización del cliente.
Paso 6: Gestiona el resto de la producción, no solo la ruta de éxito
Un backend de chatbot se considera listo para producción cuando los bordes circundantes están gestionados, no cuando la llamada de chat principal se vuelve inteligente.
La lista de verificación es corta:
- añade IDs de petición para que puedas conectar fallos del frontend con logs del backend
- limita la concurrencia por usuario y el tamaño de la petición
- recorta historiales largos antes de que exploten tu presupuesto de tokens
- registra el modelo, la latencia, el tamaño de entrada y la razón de finalización
- separa los mensajes de error visibles para el usuario de los detalles de error internos
- prueba un modelo alternativo para saber que el fallback funciona antes de la primera interrupción real
El recorte del historial puede mantenerse 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
El punto no es una contabilidad perfecta de tokens. El punto es detener explosiones de contexto obvias antes de que afecten tu factura o tu latencia.
De demo a producto
Una vez que este backend es estable, la siguiente actualización rara vez es "más IA". Suele ser infraestructura aburrida:
- autenticación para que un usuario no pueda leer la conversación de otro
- persistencia para que las sesiones sobrevivan a los despliegues
- limitación de tasa para que un usuario ruidoso no pueda quemar tu cuota
- facturación o atribución de uso si el chatbot está orientado al cliente
- resumen en segundo plano si las conversaciones necesitan memoria a largo plazo
Una puerta de enlace unificada ayuda con la mayor parte de esto. Una vez que la migración de base URL queda atrás, los cambios de modelo dejan de ser una reescritura de plataforma y se convierten en una edición de configuración.
Prueba de humo
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 puedes hacer streaming de un turno, preservar una conversación y devolver un error limpio ante un fallo forzado, tienes la base correcta.
Estimación de costes
Crea una API key en TokenLab, apunta tu SDK de OpenAI a https://api.tokenlab.sh/v1, y podrás lanzar la primera versión de producción de tu chatbot sin gestionar cuentas separadas entre proveedores.
| Modelo | Nivel típico | Notas |
|---|---|---|
| DeepSeek V4 Flash | Rápido / por defecto | Buen valor por defecto para turnos de alto volumen y baja latencia |
| GPT-5.5 | Flagship | Úsalo para turnos que necesiten un razonamiento más fuerte |
| Claude Sonnet 5 | Equilibrado | Excelente elección para codificación y respuestas de estilo revisión |
| Gemini 3.5 Flash | Presupuesto / alt rápido | Alternativa rápida y de bajo coste para enrutamiento de alto volumen |
Los precios exactos por token cambian frecuentemente entre proveedores y no se reproducen aquí como números fijos. Consulta las tarifas actuales en el directorio de modelos (observado el 07-07-2026) antes de presupuestar. En la práctica, enrutar la mayoría de las conversaciones a un nivel rápido/por defecto como DeepSeek V4 Flash o Gemini 3.5 Flash y reservar GPT-5.5 o Claude Sonnet 5 para turnos que lo necesiten mantiene la mayoría de las aplicaciones con una factura mensual baja, pero confirma las tarifas reales por millón de tokens para tu cuenta antes de comprometerte con un presupuesto.
Preguntas frecuentes
¿Necesito websockets para construir un chatbot de IA? No. El streaming SSE, mostrado en el Paso 2, cubre la gran mayoría de los productos de chat. Los websockets añaden valor real cuando necesitas push bidireccional fuera de la solicitud/respuesta, como colaboración en vivo o eventos iniciados por el servidor. Para una interfaz de chat estándar, SSE es más sencillo de desplegar, depurar y escalar.
¿Cómo sé qué modelo establecer por defecto? Empieza con un modelo rápido y de bajo coste como DeepSeek V4 Flash o Gemini 3.5 Flash para el nivel por defecto, y añade un nivel equilibrado o de razonamiento en Claude Sonnet 5 o GPT-5.5 detrás de la lista de permitidos mostrada en el Paso 5. Consulta el directorio de modelos (observado el 07-07-2026) para ver las opciones actuales, ya que los nuevos modelos se lanzan y los antiguos se deprecian en un calendario fuera de tu control.
¿Qué es lo primero que falla cuando un chatbot pasa de demo a tráfico real? Casi siempre la ruta de error, no la ruta de éxito. Los reintentos ilimitados, la falta de límites de concurrencia por usuario y el historial de conversación ilimitado son las tres causas más comunes de que un backend de chatbot caiga bajo carga real. Los pasos 4 y 6 anteriores abordan los tres directamente.
Empieza con tu API key: tokenlab.sh proporciona más de 300 modelos a través de un único endpoint. $1 de crédito gratuito para empezar a construir.
Fuentes
Precio observado el 2026-07-07
- TokenLab model directoryObservado el 2026-07-07
