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 lanzar una demo de juguete. El objetivo es llegar a un backend que realmente puedas colocar detrás de la interfaz de un producto e iterar de forma segura.
Si ya has apuntado un SDK compatible con OpenAI a LemonData, este artículo continúa desde ahí. Si aún no has realizado el cambio de la base URL, lee primero la guía de migración. Si tu principal preocupación es el modelado de peticiones y el backoff bajo carga, combina esta guía con la guía de rate limiting de API de IA.
Qué vamos a construir
El servicio terminado tiene seis partes móviles:
- Un endpoint
/chatsíncrono para pruebas de humo (smoke tests). - Un endpoint
/chat/streamde streaming para la UI real. - Estado de la conversación identificado por
conversation_id. - Una lista de permitidos (allowlist) de modelos para que el frontend no pueda solicitar IDs arbitrarios.
- Manejo de errores que no colapse ante el primer 429.
- Un camino claro desde un prototipo en memoria hacia Redis o PostgreSQL.
Eso es suficiente para alimentar 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 para la primera pasada, pero es útil configurar el import ahora para que el camino de actualización sea obvio.
Paso 1: Comienza con un endpoint de chat pequeño y sencillo
La forma más rápida de perderse en el desarrollo de chatbots es empezar con websockets, tool use y orquestación de agentes antes de que la ruta de petición básica sea estable. Comienza con un endpoint pequeño que demuestre que tu key, base URL y enrutamiento de modelos son correctos.
from fastapi import FastAPI
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}
Ejecuta una prueba de humo. Si esto falla, no sigas añadiendo capas de funcionalidades encima.
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 UI permanece en blanco hasta que llega la respuesta completa. SSE es suficiente para muchos 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, el cliente de navegador más simple 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 utiliza un cliente de navegador y HTTP estándar, SSE mantiene la arquitectura más simple.
Paso 3: Mueve el estado de la conversación fuera del cuerpo de la petición
La primera demo de un chatbot suele mantener la transcripción completa en el navegador y enviarla en cada turno. Eso funciona para prototipos. Se vuelve complicado en el momento en 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
El camino de actualización a Redis es principalmente fontanería de almacenamiento:
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))
Usa Redis si las conversaciones necesitan TTL, capacidad de reanudación o despliegue multi-instancia. Usa PostgreSQL si la propia transcripción son datos del producto.
Paso 4: Trata los errores como comportamiento del producto, no solo como excepciones
Si tu chatbot está de cara al cliente, la ruta de fallo importa tanto como la ruta de éxito. A un usuario no le importa si el fallo provino del rate limiting, del saldo o de una caída del modelo. Les importa si la UI 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] The model is busy. Please retry in a few seconds.\n\n"
except APIConnectionError:
yield "data: [ERROR] Temporary network issue. Please retry.\n\n"
except APIError as error:
yield f"data: [ERROR] {error.message}\n\n"
else:
conversations[conv_id].append(
{"role": "assistant", "content": "".join(full_response)}
)
finally:
yield "data: [DONE]\n\n"
return StreamingResponse(
generate(),
media_type="text/event-stream",
headers={"X-Conversation-ID": conv_id}
)
Si estás sirviendo una carga significativa, también deberías modelar las peticiones antes de que lleguen al upstream. Los patrones detallados están en la guía de rate limiting, pero la versión corta es: usa reintentos limitados, usa jitter y nunca hagas except Exception: time.sleep(1).
Paso 5: El cambio de modelo necesita una lista de permitidos, no un cuadro de texto libre
Una sola API key puede acceder a cientos de modelos. Eso no significa que tu UI deba exponer cientos de modelos. El backend debería publicar una pequeña lista de permitidos (allowlist) adaptada a tu caso de uso.
AVAILABLE_MODELS = {
"fast": "gpt-4.1-mini",
"balanced": "claude-sonnet-4-6",
"reasoning": "o3",
"budget": "deepseek-chat",
}
@app.get("/models")
async def list_models():
return {"models": AVAILABLE_MODELS}
Esto hace tres cosas útiles:
- evita que el frontend solicite IDs de modelos inválidos o obsoletos
- te permite remapear un nivel (tier) más tarde sin tener que volver a desplegar cada cliente
- te da un lugar único para aplicar controles de costes
Si tu equipo aún está decidiendo en qué proveedores estandarizarse, la comparativa de precios y la comparativa OpenRouter vs LemonData son las dos páginas que vale la pena leer antes de cerrar la allowlist.
Paso 6: Añade los detalles de producción antes de que llegue el tráfico
Un backend de chatbot alcanza el nivel de producción cuando se gestionan los detalles periféricos, no cuando la llamada de chat principal es ingeniosa.
La lista de verificación es corta:
- añade IDs de petición para que puedas conectar los fallos del frontend con los logs del backend
- limita la concurrencia por usuario y el tamaño de la petición
- recorta los historiales largos antes de que disparen tu presupuesto de tokens
- registra el modelo, la latencia, el tamaño de la entrada y el motivo de finalización (finish reason)
- separa los mensajes de error visibles para el usuario de los detalles técnicos internos
- prueba un modelo alternativo para saber que el fallback funciona antes de la primera caída
El recorte del historial puede ser 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 llevar una contabilidad de tokens perfecta. El punto es detener las explosiones obvias de contexto.
De la demo al 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
- rate limiting para que un usuario ruidoso no agote tu cuota
- facturación o atribución de uso si el chatbot está de cara al cliente
- resumen en segundo plano si las conversaciones necesitan memoria a largo plazo
Por eso ayuda un gateway unificado. Una vez que has realizado la migración de la base URL, los cambios de modelo dejan de ser una reescritura de la plataforma y se convierten en configuración.
Prueba de humo (Smoke Test)
uvicorn main:app --reload --port 8000
curl -N -X POST http://localhost:8000/chat/stream \
-H "Content-Type: application/json" \
-d '{"message": "Hello!", "model": "gpt-4.1-mini"}'
Si puedes transmitir un turno, preservar una conversación y devolver un error limpio en un fallo forzado, tienes la base adecuada.
Estimación de costes
Crea una API key en LemonData, apunta tu SDK de OpenAI a https://api.lemondata.cc/v1 y podrás lanzar la primera versión de producción de tu chatbot sin gestionar cuentas de proveedores por separado.
| Modelo | Coste diario | Coste mensual |
|---|---|---|
| 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 |
Usar GPT-4.1-mini para la mayoría de las conversaciones y actualizar a Claude Sonnet 4.6 solo cuando los usuarios lo soliciten mantiene los costes por debajo de los 100 $/mes para la mayoría de las aplicaciones.
Obtén tu API key: lemondata.cc proporciona más de 300 modelos a través de un único endpoint. 1 $ de crédito gratuito para empezar a construir.