Crea un chatbot de IA con una sola API Key: de cero a producción en 30 minutos

Puedes construir un chatbot de IA que transmita una respuesta en una tarde. Mantenerlo vivo después del primer 429 es el verdadero trabajo. Construiremos un pequeño servicio FastAPI con streaming SSE, memoria de conversación en el servidor y una lista de modelos permitidos, todo bajo una sola clave de TokenLab. El objetivo es un backend que puedas colocar detrás de una interfaz de producto real y cambiar de modelos sin tener que reescribir la integración. Si aún no has realizado el cambio de URL base, lee primero la guía de migración. Si tu principal preocupación es el backoff bajo carga, combina esta guía con la guía de limitación de tasa 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 servidor, una lista de modelos permitidos, manejo de errores real y una ruta de actualización de almacenamiento.
• Carga la clave desde TOKENLAB_API_KEY en cada proceso y mantenla en el servidor. La documentación de autenticación (observada el 03-10-2026) indica que nunca se deben exponer las claves en el código del lado del cliente ni subirlas al repositorio.
• GET /v1/models es público, por lo que una respuesta exitosa no prueba que tu clave funcione. Pruébala con una solicitud de chat real (Quickstart, observado el 03-10-2026).
• SSE cubre la mayoría de los productos de chat. Trata una transmisión que se interrumpe después de que llegó la salida como incompleta, no como una respuesta finalizada.
• La lista de permitidos en esta guía asigna niveles a deepseek-v4-flash, gpt-5.5, claude-sonnet-5 y gemini-3.5-flash. Cada uno enumera openai_chat_completions como un formato aceptado (observado el 03-10-2026).
• Como estimación basada en los precios a continuación, un turno de 2,000 tokens de entrada y 500 de salida cuesta entre $0.0006 y $0.0075, dependiendo del modelo.

Construye un chatbot de IA: configuración y servicio principal

El servicio terminado tiene seis partes móviles:

1. Un endpoint /chat síncrono para pruebas de humo.
2. Un endpoint /chat/stream de streaming para la interfaz real.
3. Estado de conversación identificado por conversation_id.
4. Una lista de modelos permitidos, para que el frontend no pueda solicitar IDs arbitrarios.
5. Manejo de errores que no colapsa ante el primer 429.
6. Una ruta desde el prototipo en memoria hacia Redis o PostgreSQL.

Eso es suficiente para un bot de soporte, un asistente interno o la primera versión de un widget de chat integrado. Cada bloque de código a continuación se añade a un archivo, main.py, en el orden mostrado.

Paso 1: Configura la clave e instala el stack

Crea una clave en API Keys en tu espacio de trabajo. El valor completo se muestra solo una vez, y el prefijo en la lista de claves no puede autenticar solicitudes. Todas las claves de TokenLab comienzan con sk-. Verifica tu saldo en Billing también, porque la primera solicitud se factura al precio actual del modelo (Quickstart, observado el 03-10-2026).

Almacena la clave en una variable de entorno, nunca en el código fuente. El prompt a continuación la mantiene fuera de la línea de comandos y del historial del shell, y se aplica solo a la terminal actual y sus procesos hijos. Usa un gestor de secretos para aplicaciones desplegadas. El Quickstart también tiene una versión para PowerShell.

printf 'TokenLab API key: '
read -r -s TOKENLAB_API_KEY
printf '\n'
export TOKENLAB_API_KEY

pip install fastapi uvicorn openai pydantic

No necesitas un SDK específico de TokenLab. El SDK oficial de OpenAI funciona una vez que cambias la clave y la URL base (guía de SDKs, observado el 03-10-2026).

Paso 2: Define el esquema, la lista de permitidos y la memoria

Comenzamos con las piezas compartidas. El esquema de solicitud utiliza tier, no una cadena de model cruda, y cada ejemplo de curl más adelante en esta guía envía los mismos campos.

La primera demo de chatbot generalmente mantiene la transcripción en el navegador y la reenvía en cada turno. Por ejemplo, eso funciona para un prototipo. Se vuelve complicado una vez que necesitas reintentos, sesiones reanudables o herramientas en el servidor. Por lo tanto, el servidor posee el historial desde el principio.

import json
import logging
import os
import time
import uuid
from collections import defaultdict

from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from openai import APIConnectionError, APIStatusError, OpenAI, RateLimitError
from pydantic import BaseModel, Field

log = logging.getLogger("chatbot")
app = FastAPI()

client = OpenAI(
    api_key=os.environ["TOKENLAB_API_KEY"],
    base_url="https://api.tokenlab.sh/v1",
    timeout=30.0,
    max_retries=0,
)

ALLOWED_MODELS = {
    "default": "deepseek-v4-flash",
    "flagship": "gpt-5.5",
    "balanced": "claude-sonnet-5",
    "budget": "gemini-3.5-flash",
}
SYSTEM_PROMPT = "You are a helpful assistant. Be concise and direct."
MAX_HISTORY = 20          # mensajes guardados por conversación (a tu elección)
MAX_REPLY_TOKENS = 1000   # límite de salida por turno (a tu elección)

class ChatRequest(BaseModel):
    message: str = Field(min_length=1, max_length=4000)
    tier: str = "default"
    conversation_id: str | None = None

def resolve_model(tier: str) -> str:
    model = ALLOWED_MODELS.get(tier)
    if model is None:
        raise HTTPException(status_code=400, detail=f"unknown tier: {tier}")
    return model

conversations: dict[str, list[dict]] = defaultdict(list)

def build_messages(conv_id: str, user_msg: str) -> list[dict]:
    history = conversations[conv_id][-MAX_HISTORY:]
    return [
        {"role": "system", "content": SYSTEM_PROMPT},
        *history,
        {"role": "user", "content": user_msg},
    ]

def remember(conv_id: str, user_msg: str, reply: str) -> None:
    conversations[conv_id].append({"role": "user", "content": user_msg})
    conversations[conv_id].append({"role": "assistant", "content": reply})

El cliente establece max_retries=0 y un tiempo de espera de 30 segundos, coincidiendo con el Quickstart. Tomamos nuestras propias decisiones de reintento, porque un reintento ciego del SDK en un turno transmitido puede repetir un cargo.

Paso 3: Mapea errores y añade el endpoint síncrono

Mantenemos una pequeña función que convierte cualquier excepción del SDK en un código de estado y un mensaje seguro para el usuario. El detalle va a tus registros, no al usuario.

def describe_error(exc: Exception) -> tuple[int, str]:
    if isinstance(exc, RateLimitError):  # verificar primero: es subclase de APIStatusError
        return 429, "The assistant is busy. Please retry shortly."
    if isinstance(exc, APIConnectionError):
        return 502, "Connection problem. Please retry."
    if isinstance(exc, APIStatusError):
        log.error("upstream status=%s code=%s", exc.status_code, exc.code)
        if exc.status_code in (401, 402, 403):  # clave, saldo o permiso: problema del operador
            return 503, "The assistant is unavailable right now."
        if exc.status_code in (400, 404):  # campo incorrecto o ID de modelo: nuestro error
            return 500, "Something went wrong on our side."
        return 502, "The assistant is unavailable right now."
    return 500, "Something went wrong."

def retry_delay(exc: RateLimitError) -> float:
    try:
        return min(float(exc.response.headers.get("retry-after", "2")), 10.0)
    except ValueError:
        return 2.0

def create_with_one_retry(**kwargs):
    try:
        return client.chat.completions.create(**kwargs)
    except RateLimitError as exc:
        time.sleep(retry_delay(exc))
        return client.chat.completions.create(**kwargs)

@app.post("/chat")
def chat(req: ChatRequest):
    model = resolve_model(req.tier)
    conv_id = req.conversation_id or str(uuid.uuid4())
    messages = build_messages(conv_id, req.message)
    try:
        response = create_with_one_retry(
            model=model, messages=messages, max_tokens=MAX_REPLY_TOKENS
        )
    except (APIConnectionError, APIStatusError) as exc:
        status, message = describe_error(exc)
        raise HTTPException(status_code=status, detail=message)
    reply = response.choices[0].message.content or ""
    log.info("model=%s usage=%s", model, response.usage)
    remember(conv_id, req.message, reply)
    return {"conversation_id": conv_id, "reply": reply}

Ejecuta una solicitud antes de construir cualquier otra cosa. Si esto falla, no sigas construyendo sobre ello.

Transmite respuestas y mantén el chatbot de IA receptivo

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. Añadimos SSE porque cubre la mayoría de los productos de chat y conlleva menos peso operativo que los websockets. Si tu producto ya ejecuta un cliente de navegador sobre HTTP simple, SSE mantiene la arquitectura más simple.

Paso 4: Añade el endpoint de streaming

La documentación de streaming (Guía de streaming, observada el 03-10-2026) mantiene el streaming de Chat Completions válido para frameworks que ya esperan fragmentos SSE de /v1/chat/completions. Nos mantenemos en ese camino porque los cuatro modelos permitidos lo aceptan. Codificamos cada evento como JSON, por lo que las nuevas líneas dentro de una respuesta no pueden romper el marco.

def sse(payload: dict) -> str:
    return f"data: {json.dumps(payload)}\n\n"

@app.post("/chat/stream")
def chat_stream(req: ChatRequest):
    model = resolve_model(req.tier)
    conv_id = req.conversation_id or str(uuid.uuid4())
    messages = build_messages(conv_id, req.message)

    def generate():
        parts: list[str] = []
        finish_reason = None
        try:
            with client.chat.completions.create(
                model=model,
                messages=messages,
                max_tokens=MAX_REPLY_TOKENS,
                stream=True,
                stream_options={"include_usage": True},
            ) as stream:
                for chunk in stream:
                    if chunk.usage:
                        log.info("model=%s usage=%s", model, chunk.usage)
                    if not chunk.choices:
                        continue
                    choice = chunk.choices[0]
                    if choice.delta.content:
                        parts.append(choice.delta.content)
                        yield sse({"delta": choice.delta.content})
                    if choice.finish_reason:
                        finish_reason = choice.finish_reason
        except (APIConnectionError, APIStatusError) as exc:
            _, message = describe_error(exc)
            yield sse({"error": message})
            return
        if finish_reason != "stop":
            yield sse({"error": f"Reply incomplete ({finish_reason})."})
            return
        remember(conv_id, req.message, "".join(parts))
        yield sse({"done": True, "conversation_id": conv_id})

    return StreamingResponse(generate(), media_type="text/event-stream")

Dos detalles importan aquí. Primero, include_usage: true añade el uso de tokens a los fragmentos de la transmisión, y el fragmento de uso final no tiene choices, por lo que el bucle lo omite. Segundo, guardamos el turno solo cuando finish_reason es stop. Un final por length significa que el límite de tokens cortó la respuesta, y la documentación enumera ambos como condiciones finales normales.

Debido a que los errores surgen dentro del generador, el estado HTTP ya es 200 cuando ocurre una falla. El cliente debe leer el evento error. El bloque with cierra la transmisión ascendente cuando el generador sale. La documentación aconseja cancelar la transmisión ascendente cuando el usuario se detiene o se desconecta, así que prueba que tu servidor ASGI realmente detenga el generador al desconectarse.

Paso 5: Lee la transmisión en el navegador

El navegador debe llamar a tu propio backend autenticado y nunca directamente a TokenLab. Analiza por límite de evento, porque un fragmento de red no es necesariamente un evento SSE completo.

async function sendMessage(payload, onDelta) {
  const response = await fetch('/chat/stream', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify(payload), // { message, tier, conversation_id }
  });
  if (!response.ok || !response.body) throw new Error(`HTTP ${response.status}`);

  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let buffer = '';

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buffer += decoder.decode(value, { stream: true });
    const events = buffer.split('\n\n');
    buffer = events.pop();
    for (const raw of events) {
      if (!raw.startsWith('data: ')) continue;
      const event = JSON.parse(raw.slice(6));
      if (event.error) throw new Error(event.error);
      if (event.delta) onDelta(event.delta);
    }
  }
}

Actualiza la memoria a Redis cuando lo necesites

El diccionario en memoria está bien para un proceso. Busca Redis cuando las conversaciones necesiten un TTL, capacidad de reanudación o múltiples instancias. Busca PostgreSQL cuando la transcripción sea datos de producto que debas consultar, auditar o sobre los cuales informar. El cambio a Redis es fontanería de almacenamiento, no lógica. Ejecuta pip install redis, luego haz que build_messages y remember lean y escriban a través de estos ayudantes.

import redis

redis_client = redis.Redis(
    host=os.environ.get("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))

Maneja errores cuando construyas un chatbot de IA para producción

A un usuario final no le importa si una falla provino de una limitación de tasa, un saldo vacío o una interrupción ascendente. Le importa si la interfaz se congela. Imagina un widget de soporte cuya transmisión muere después de cuarenta palabras. La documentación de TokenLab dice que una transmisión interrumpida después de que llegó la salida no es una respuesta completa, y repetirla puede generar una salida diferente y un segundo cargo (Manejo de errores de API, observado el 03-10-2026). Es por eso que el Paso 4 nunca guarda un turno inacabado.

Paso 6: Aplica las reglas de reintento

Maneja los errores por estado HTTP y code, no por el message legible por humanos, que puede cambiar sin previo aviso. La tabla combina la documentación con lo que hace nuestro servicio.

Estado	Significado (según docs)	Lo que hace este servicio
400	Campo, ID de modelo o entrada no válida	Devolver 500, registrarlo, nunca reintentar
401	Clave faltante, inválida, expirada o revocada	Devolver 503, alertar al operador
402	Saldo o límite de API-key demasiado bajo (insufficient_balance, quota_exceeded)	Devolver 503, alertar al operador
403	La clave no puede usar este modelo	Devolver 503, verificar la lista de modelos de la clave
404	Recurso o modelo no encontrado	Devolver 500, verificar el ID contra /v1/models
429	Límite de solicitud alcanzado	Esperar por Retry-After, reintentar una vez (solo síncrono)
500–504	Servicio no disponible o falla de red	Reintentar solo cuando retryable sea true; respetar retry_after

Dos advertencias de la documentación. Un 503 all_channels_failed no siempre significa una interrupción temporal. Si retryable es false, no repitas la misma solicitud y verifica GET /v1/models antes de elegir otro modelo. Además, Chat Completions y Responses usan un objeto error al estilo OpenAI, pero Anthropic Messages y Gemini mantienen sus propios formatos. Nuestro servicio llama a Chat Completions en cada modelo permitido, por lo que un analizador funciona. Si mueves un nivel a un formato nativo, necesitas un segundo analizador.

Guarda el ID de solicitud de los encabezados de respuesta con el endpoint, modelo, tiempo y tu propio ID de usuario o trabajo. La documentación no nombra el encabezado en las páginas que usamos, así que confírmalo contra la ruta de solución de problemas antes de conectar tus registros.

El resto de la lista de verificación de producción es corta:

• Añade tus propios IDs de solicitud para que puedas conectar fallas del frontend a registros del backend.
• Limita la concurrencia por usuario y el tamaño de la solicitud. El max_length=4000 en message es un punto de partida que deberías ajustar.
• Limita el historial por conteo de mensajes, como hace MAX_HISTORY. La documentación de facturación dice que la tokenización difiere según el modelo, el idioma y la entrada multimedia, por lo que una fórmula de conteo de caracteres no es una estimación de facturación confiable.
• Registra el modelo, la latencia, el uso y la razón de finalización en cada turno.
• Mantén los mensajes visibles para el usuario separados del detalle de error interno.
• Envía un turno de prueba a un nivel alternativo, para que sepas que la alternativa funciona antes de la primera interrupción real.

Elige modelos y estima el costo para tu chatbot de IA

Nunca permitas que el frontend pase una cadena de modelo arbitraria a la API. Un campo de texto libre invita a errores tipográficos, modelos retirados y modelos que nunca quisiste exponer. El mapa ALLOWED_MODELS te da un lugar para cambiar un modelo cuando un proveedor lo depreca, sin actualización del cliente. Los IDs de modelo desconocidos devuelven model_not_found, y la documentación te dirige a /v1/models para un ID actual. Consulta el directorio de modelos antes de bloquear la lista de permitidos.

Precios actuales por 1M de tokens

Estos son precios de TokenLab en USD de la entrada de API en vivo de cada modelo, observados el 03-10-2026. La fijación de precios se actualizó por última vez el 02-10-2026T16:53:30.068Z. Los cuatro modelos mostraron la etapa de ciclo de vida active.

Nivel	ID de modelo	Entrada / 1M	Salida / 1M	Lectura caché / 1M	Fuente
default	deepseek-v4-flash (fuera de pico)	$0.15	$0.60	$0.003	model API
default	deepseek-v4-flash (pico)	$0.30	$1.20	$0.006	model API
flagship	gpt-5.5	$1.50	$9.00	$0.15	model API
balanced	claude-sonnet-5	$0.90	$4.50	$0.09	model API
budget	gemini-3.5-flash	$0.75	$4.50	$0.075	model API

Para deepseek-v4-flash, la entrada enumera ventanas de pico en días laborables, excluyendo los días festivos de China. Fuera de pico incluye fines de semana y esos días festivos. claude-sonnet-5 también enumera un precio de escritura en caché de $1.125 por 1M de tokens.

Una solicitud trabajada

Supongamos un turno con 2,000 tokens de entrada y 500 tokens de salida. Ese tamaño es nuestra suposición, no una medición. Para gpt-5.5:

• Entrada: 2,000 × $1.50 / 1,000,000 = $0.0030
• Salida: 500 × $9.00 / 1,000,000 = $0.0045
• Total: $0.0075 por turno (estimación, sin aciertos de caché)

Usando la misma aritmética, obtenemos estas estimaciones: deepseek-v4-flash cuesta $0.0006 fuera de pico y $0.0012 en pico, claude-sonnet-5 cuesta $0.00405 y gemini-3.5-flash cuesta $0.00375. Durante 1,000 turnos de este tipo, eso es aproximadamente $7.50 en gpt-5.5 frente a $0.60 a $1.20 en deepseek-v4-flash. Los turnos reales crecen a medida que se acumula el historial, así que lee usage de la respuesta o del fragmento de transmisión final en lugar de confiar en este dimensionamiento.

En la práctica, dirigimos la mayoría de las conversaciones al nivel predeterminado y reservamos gpt-5.5 o claude-sonnet-5 para turnos que los necesitan. La documentación de facturación advierte que el precio más bajo por token no siempre es el costo más bajo por tarea completada. Ejecuta los mismos prompts a través de dos niveles y compara el costo final en Usage.

Dos comprobaciones antes de enviar. La documentación de facturación dice que no se debe codificar una tabla de precios copiada, así que lee GET /v1/models/:model/pricing si tu aplicación muestra costos (Facturación y precios, observado el 03-10-2026). Además, la entrada de deepseek-v4-flash enumera json-mode, prompt-cache y tool-use pero no vision. Si tu chatbot acepta imágenes, dirige esos turnos a un nivel cuya entrada enumere vision.

Prueba de humo y envío de tu chatbot de IA

Paso 7: Ejecútalo de extremo a extremo

Inicia el servidor en una terminal donde TOKENLAB_API_KEY esté exportada. Las cargas útiles de curl usan los mismos campos message, tier y conversation_id que ChatRequest.

uvicorn main:app --reload --port 8000

# 1. Endpoint síncrono: espera una respuesta corta y un conversation_id
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "Reply only with OK.", "tier": "default"}'

# 2. Streaming: espera líneas data: {"delta": ...}, luego {"done": true, ...}
curl -N -X POST http://localhost:8000/chat/stream \
  -H "Content-Type: application/json" \
  -d '{"message": "Hello!", "tier": "default"}'

# 3. Memoria: reutiliza un conversation_id devuelto arriba
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "What did I just say?", "tier": "default", "conversation_id": "PASTE_ID_FROM_STEP_1"}'

# 4. Falla forzada: espera HTTP 400 y "unknown tier: nope"
curl -i -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "Hello!", "tier": "nope"}'

Para forzar una falla ascendente, reinicia el servidor con una TOKENLAB_API_KEY incorrecta. Tu cliente debería recibir un mensaje 503 limpio, y tus registros deberían mostrar el 401 ascendente con invalid_api_key. Luego abre Requests para confirmar el modelo, el estado y el uso de las llamadas exitosas. Si puedes transmitir un turno, preservar una conversación y devolver un error limpio en una falla forzada, tienes la base correcta.

Después de eso, las siguientes actualizaciones son infraestructura aburrida en lugar de más IA:

• Autenticación, para que un usuario no pueda leer la conversación de otro usuario.
• Persistencia, para que las sesiones sobrevivan a los despliegues.
• Limitación de tasa por usuario, para que un usuario ruidoso no pueda quemar tu cuota.
• Atribución de uso, si el chatbot está orientado al cliente.
• Resumen en segundo plano, si las conversaciones necesitan memoria a largo plazo.

Una vez que la migración de URL base quede atrás, un cambio de modelo es una edición de configuración en ALLOWED_MODELS.

Preguntas frecuentes

¿Necesito websockets para construir un chatbot de IA?

No. SSE, mostrado en el Paso 4, cubre la mayoría de los productos de chat. Los websockets añaden valor cuando necesitas push bidireccional fuera de solicitud y respuesta, como colaboración en vivo o eventos iniciados por el servidor. Para una interfaz de chat estándar, SSE es más simple de desplegar y depurar.

¿Qué modelo debería usar mi chatbot por defecto?

Comienza con un nivel de bajo precio como deepseek-v4-flash y añade claude-sonnet-5 o gpt-5.5 detrás de la lista de permitidos para turnos más difíciles. Los precios anteriores, observados el 03-10-2026, muestran la brecha. Luego prueba tus propios prompts en dos niveles y compara el costo final en Usage. Consulta el directorio de modelos antes de comprometerte, ya que las alineaciones cambian.

¿Por qué mi chatbot devuelve 402 o 429?

Un 402 significa que el saldo de la cuenta o el límite de uso de la propia clave es demasiado bajo. Los códigos son insufficient_balance y quota_exceeded, y repetir la solicitud no ayudará. Un 429 significa que alcanzaste un límite de solicitud, así que espera por Retry-After y luego reintenta. Nuestro endpoint síncrono reintenta una vez y el endpoint de streaming informa el error al cliente.

¿Puede el navegador llamar a TokenLab directamente?

No de forma segura. La documentación de autenticación dice que nunca se deben exponer las claves de API en el código del lado del cliente. Mantén la clave en el entorno del servidor y haz que el navegador llame a tu propio backend autenticado, como hace el ejemplo de frontend.

---

Crea una clave de API en tokenlab.sh, añade fondos y ejecuta la prueba de humo anterior.