Construye un Chatbot de IA con Una Clave API: De Cero a Producción en 30 Minutos
Este tutorial crea un backend de chatbot de IA listo para producción con respuestas en streaming, historial de conversación, cambio de modelo y manejo adecuado de errores. Usaremos Python, FastAPI y el SDK de OpenAI apuntando a un agregador de API para que puedas usar cualquier modelo.
Requisitos Previos
pip install fastapi uvicorn openai
Paso 1: Endpoint Básico de Chat
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}
Esto funciona pero no tiene streaming, ni historial, ni manejo de errores. Vamos a corregir eso.
Paso 2: Añadir Streaming
El streaming envía tokens a medida que se generan en lugar de esperar la respuesta completa. Los usuarios ven la respuesta formándose en tiempo real.
@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"
)
Paso 3: Historial de Conversación
Almacena el historial de conversación en memoria (cámbialo por Redis o una base de datos en producción).
from collections import defaultdict
import uuid
conversations: dict[str, list] = defaultdict(list)
SYSTEM_PROMPT = "Eres un asistente útil. Sé conciso y directo."
@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
conv_id = req.conversation_id or str(uuid.uuid4())
# Construir historial de mensajes
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
messages.extend(conversations[conv_id])
messages.append({"role": "user", "content": req.message})
# Guardar mensaje del usuario
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"
# Guardar respuesta del asistente
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}
)
Paso 4: Manejo de Errores
Las llamadas a la API de IA pueden fallar por varias razones: límites de tasa, saldo insuficiente, modelo no disponible. Maneja cada caso:
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] Límite de tasa alcanzado. Por favor espera un momento.\n\n"
except APIConnectionError:
yield f"data: [ERROR] Falló la conexión. Reintentando...\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}]
# Mantener los últimos 10 turnos para gestionar la longitud del contexto
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
Paso 5: Cambio de Modelo
Permite a los usuarios cambiar de modelo en medio de la conversación. Diferentes modelos para diferentes necesidades:
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}
El frontend puede presentar estas opciones. Dado que todos los modelos usan el mismo formato compatible con OpenAI a través del agregador, cambiar es solo modificar el parámetro model.
Paso 6: Gestión de la Ventana de Contexto
Las conversaciones largas exceden los límites de contexto del modelo. Implementa una ventana deslizante:
def trim_history(messages: list, max_tokens: int = 8000) -> list:
"""Mantener el prompt del sistema + mensajes recientes dentro del presupuesto de tokens."""
# Estimación aproximada: 1 token ≈ 4 caracteres
system = messages[0] # Siempre mantener el prompt del sistema
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
Aplicación Completa
# Ejecutar con: uvicorn main:app --reload --port 8000
# Probar: curl -N -X POST http://localhost:8000/chat/stream \
# -H "Content-Type: application/json" \
# -d '{"message": "¡Hola!", "model": "gpt-4.1-mini"}'
El código completo tiene menos de 100 líneas. Desde aquí puedes añadir:
- Autenticación (claves API o JWT)
- Almacenamiento persistente (PostgreSQL o Redis para conversaciones)
- Limitación de tasa por usuario
- Seguimiento de uso y facturación
- Soporte WebSocket para streaming bidireccional
- Frontend (React, Vue o JS puro con EventSource)
Estimación de Costos
Para un chatbot que maneje 1,000 conversaciones/día (promedio 5 turnos cada una):
| Modelo | Costo Diario | Costo 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 costos por debajo de $100/mes para la mayoría de las aplicaciones.
Obtén tu clave API: lemondata.cc ofrece más de 300 modelos a través de un solo endpoint. $1 de crédito gratis para comenzar a construir.