Construa um Chatbot de IA com uma única API Key: Do Zero à Produção em 30 Minutos

Você pode criar um chatbot de IA que transmite uma resposta em uma tarde. Mantê-lo vivo após o primeiro 429 é o verdadeiro trabalho. Criamos um pequeno serviço FastAPI com streaming SSE, memória de conversão no lado do servidor e uma lista de permissões de modelos, tudo sob uma única chave TokenLab. O objetivo é um backend que você possa colocar atrás de uma interface de produto real e trocar de modelos sem reescrever a integração. Se você ainda não fez a troca da URL base, leia o guia de migração primeiro. Se sua principal preocupação é o backoff sob carga, combine este guia com o guia de limitação de taxa da API de IA.

Principais Pontos

• Um chatbot pronto para produção precisa de seis partes: um endpoint síncrono, um endpoint de streaming, estado de conversação no lado do servidor, uma lista de permissões de modelos, tratamento de erros real e um caminho de atualização de armazenamento.
• Carregue a chave de TOKENLAB_API_KEY em cada processo e mantenha-a no servidor. A documentação de autenticação (observada em 03/10/2026) diz para nunca expor chaves em código do lado do cliente ou fazer commit delas.
• GET /v1/models é público, portanto, uma resposta bem-sucedida não prova que sua chave funciona. Prove com uma solicitação de chat real (Quickstart, observado em 03/10/2026).
• O SSE cobre a maioria dos produtos de chat. Trate um stream que morre após a saída chegar como incompleto, não como uma resposta finalizada.
• A lista de permissões neste guia mapeia tiers para deepseek-v4-flash, gpt-5.5, claude-sonnet-5 e gemini-3.5-flash. Cada um lista openai_chat_completions como um formato aceito (observado em 03/10/2026).
• Como uma estimativa baseada nos preços abaixo, um turno de 2.000 tokens de entrada e 500 de saída custa entre US$ 0,0006 e US$ 0,0075, dependendo do modelo.

Construindo um Chatbot de IA: Configuração e o Serviço Principal

O serviço finalizado tem seis partes móveis:

1. Um endpoint /chat síncrono para testes de fumaça.
2. Um endpoint /chat/stream de streaming para a interface real.
3. Estado de conversação indexado por conversation_id.
4. Uma lista de permissões de modelos, para que o frontend não possa solicitar IDs arbitrários.
5. Tratamento de erros que não colapsa no primeiro 429.
6. Um caminho do protótipo em memória para Redis ou PostgreSQL.

Isso é suficiente para um bot de suporte, um assistente interno ou a primeira versão de um widget de chat incorporado. Cada bloco de código abaixo é anexado a um arquivo, main.py, na ordem mostrada.

Passo 1: Defina a chave e instale a stack

Crie uma chave em API Keys no seu workspace. O valor completo é mostrado apenas uma vez, e o prefixo na lista de chaves não pode autenticar solicitações. Todas as chaves TokenLab começam com sk-. Verifique seu saldo em Billing também, pois a primeira solicitação é cobrada pelo preço atual do modelo (Quickstart, observado em 03/10/2026).

Armazene a chave em uma variável de ambiente, nunca no código-fonte. O prompt abaixo a mantém fora da linha de comando e do histórico do shell, e se aplica apenas ao terminal atual e seus processos filhos. Use um gerenciador de segredos para aplicativos implantados. O Quickstart também tem uma versão para PowerShell.

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

pip install fastapi uvicorn openai pydantic

Você não precisa de um SDK específico da TokenLab. O SDK oficial da OpenAI funciona assim que você altera a chave e a URL base (guia de SDKs, observado em 03/10/2026).

Passo 2: Defina o esquema, a lista de permissões e a memória

Começamos com as peças compartilhadas. O esquema de solicitação usa tier, não uma string model bruta, e cada exemplo de curl mais adiante neste guia envia os mesmos campos.

A primeira demonstração de chatbot geralmente mantém a transcrição no navegador e a reenvia a cada turno. Por exemplo, isso funciona para um protótipo. Fica complicado quando você precisa de novas tentativas, sessões retomáveis ou ferramentas no lado do servidor. Portanto, o servidor possui o histórico desde o início.

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          # mensagens mantidas por conversação (sua escolha)
MAX_REPLY_TOKENS = 1000   # limite de saída por turno (sua escolha)

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})

O cliente define max_retries=0 e um timeout de 30 segundos, correspondendo ao Quickstart. Tomamos nossas próprias decisões de nova tentativa, porque uma nova tentativa cega do SDK em um turno transmitido pode repetir uma cobrança.

Passo 3: Mapeie erros e adicione o endpoint síncrono

Mantemos uma pequena função que transforma qualquer exceção do SDK em um código de status e uma mensagem segura para o usuário. Os detalhes vão para seus logs, não para o usuário.

def describe_error(exc: Exception) -> tuple[int, str]:
    if isinstance(exc, RateLimitError):  # verifique primeiro: é uma subclasse 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):  # chave, saldo ou permissão: problema do operador
            return 503, "The assistant is unavailable right now."
        if exc.status_code in (400, 404):  # campo ruim ou ID de modelo: nosso bug
            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}

Execute uma solicitação antes de construir qualquer outra coisa. Se isso falhar, não continue construindo sobre isso.

Transmita Respostas e Mantenha o Chatbot de IA Responsivo

A maioria dos produtos de chatbot parece lenta não porque o modelo é lento, mas porque a interface permanece em branco até que a resposta completa chegue. Adicionamos SSE porque ele cobre a maioria dos produtos de chat e carrega menos peso operacional do que websockets. Se o seu produto já executa um cliente de navegador sobre HTTP simples, o SSE mantém a arquitetura mais simples.

Passo 4: Adicione o endpoint de streaming

A documentação de streaming (Guia de streaming, observado em 03/10/2026) mantém o streaming de Chat Completions válido para frameworks que já esperam chunks SSE de /v1/chat/completions. Permanecemos nesse caminho porque todos os quatro modelos da lista de permissões o aceitam. Codificamos cada evento como JSON, para que novas linhas dentro de uma resposta não possam quebrar o enquadramento.

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")

Dois detalhes importam aqui. Primeiro, include_usage: true adiciona o uso de tokens aos chunks do stream, e o chunk de uso final não tem choices, então o loop o ignora. Segundo, salvamos o turno apenas quando finish_reason é stop. Um término por length significa que o limite de tokens cortou a resposta, e a documentação lista ambos como condições de término normais.

Como os erros surgem dentro do gerador, o status HTTP já é 200 quando uma falha ocorre. O cliente deve ler o evento error. O bloco with fecha o stream upstream quando o gerador sai. A documentação aconselha cancelar o upstream quando o usuário para ou se desconecta, então teste se o seu servidor ASGI realmente para o gerador na desconexão.

Passo 5: Leia o stream no navegador

O navegador deve chamar seu próprio backend autenticado e nunca a TokenLab diretamente. Analise por limite de evento, porque um chunk de rede não é necessariamente um 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);
    }
  }
}

Atualize a memória para Redis quando precisar

O dicionário em memória é bom para um processo. Recorra ao Redis quando as conversas precisarem de um TTL, retomabilidade ou múltiplas instâncias. Recorra ao PostgreSQL quando a transcrição for um dado de produto que você precisa consultar, auditar ou gerar relatórios. A mudança para Redis é encanamento de armazenamento, não lógica. Execute pip install redis, então faça com que build_messages e remember leiam e escrevam através destes auxiliares.

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))

Trate Erros ao Construir um Chatbot de IA para Produção

Um usuário final não se importa se uma falha veio de limitação de taxa, saldo vazio ou uma interrupção upstream. Eles se importam se a interface congela. Imagine um widget de suporte cujo stream morre após quarenta palavras. A documentação da TokenLab diz que um stream interrompido após a saída chegar não é uma resposta completa, e repeti-la pode gerar uma saída diferente e uma segunda cobrança (Tratar erros de API, observado em 03/10/2026). É por isso que o Passo 4 nunca salva um turno inacabado.

Passo 6: Aplique as regras de nova tentativa

Trate erros por status HTTP e code, não pela message legível por humanos, que pode mudar sem aviso. A tabela combina a documentação com o que nosso serviço faz.

Status	Significado (conforme docs)	O que este serviço faz
400	Campo, ID de modelo ou entrada inválida	Retorna 500, loga, nunca tenta novamente
401	Chave ausente, inválida, expirada ou revogada	Retorna 503, alerta o operador
402	Saldo ou limite da chave de API muito baixo (insufficient_balance, quota_exceeded)	Retorna 503, alerta o operador
403	Chave não pode usar este modelo	Retorna 503, verifica a lista de modelos da chave
404	Recurso ou modelo não encontrado	Retorna 500, verifica o ID contra /v1/models
429	Limite de solicitação atingido	Aguarda Retry-After, tenta novamente uma vez (apenas síncrono)
500–504	Serviço indisponível ou falha de rede	Tenta novamente apenas quando retryable é true; respeita retry_after

Dois avisos da documentação. Um 503 all_channels_failed nem sempre significa uma interrupção temporária. Se retryable for false, não repita a mesma solicitação e verifique GET /v1/models antes de escolher outro modelo. Além disso, Chat Completions e Responses usam um objeto error estilo OpenAI, mas Anthropic Messages e Gemini mantêm seus próprios formatos. Nosso serviço chama Chat Completions em cada modelo da lista de permissões, então um parser funciona. Se você mover um tier para um formato nativo, precisará de um segundo parser.

Salve o ID da solicitação dos cabeçalhos de resposta com o endpoint, modelo, hora e seu próprio ID de usuário ou trabalho. A documentação não nomeia o cabeçalho nas páginas que usamos, então confirme-o contra o caminho de solução de problemas antes de conectar seus logs.

O restante da lista de verificação de produção é curto:

• Adicione seus próprios IDs de solicitação para que você possa conectar falhas de frontend aos logs de backend.
• Limite a concorrência por usuário e o tamanho da solicitação. O max_length=4000 em message é um ponto de partida que você deve ajustar.
• Limite o histórico por contagem de mensagens, como MAX_HISTORY faz. A documentação de cobrança diz que a tokenização difere por modelo, idioma e entrada de mídia, então uma fórmula de contagem de caracteres não é uma estimativa de cobrança confiável.
• Logue modelo, latência, uso e motivo de término em cada turno.
• Mantenha as mensagens visíveis ao usuário separadas dos detalhes internos de erro.
• Envie um turno de teste para um tier alternativo, para que você saiba que o fallback funciona antes da primeira interrupção real.

Escolha Modelos e Estime o Custo para seu Chatbot de IA

Nunca deixe o frontend passar uma string de modelo arbitrária para a API. Um campo de texto livre convida erros de digitação, modelos aposentados e modelos que você nunca pretendeu expor. O mapa ALLOWED_MODELS lhe dá um lugar para trocar um modelo quando um provedor descontinuar um, sem atualização do cliente. IDs de modelo desconhecidos retornam model_not_found, e a documentação aponta para /v1/models para um ID atual. Verifique o diretório de modelos antes de bloquear a lista de permissões.

Preços atuais por 1 milhão de tokens

Estes são preços da TokenLab em USD da entrada de API ao vivo de cada modelo, observado em 03/10/2026. O preço foi atualizado pela última vez em 02/10/2026 às 16:53:30.068Z. Todos os quatro modelos mostraram estágio de ciclo de vida active.

Tier	ID do Modelo	Entrada / 1M	Saída / 1M	Leitura de Cache / 1M	Fonte
default	deepseek-v4-flash (fora de pico)	$0.15	$0.60	$0.003	API de modelo
default	deepseek-v4-flash (pico)	$0.30	$1.20	$0.006	API de modelo
flagship	gpt-5.5	$1.50	$9.00	$0.15	API de modelo
balanced	claude-sonnet-5	$0.90	$4.50	$0.09	API de modelo
budget	gemini-3.5-flash	$0.75	$4.50	$0.075	API de modelo

Para deepseek-v4-flash, a entrada lista janelas de pico em dias úteis, excluindo feriados públicos da China. Fora de pico inclui fins de semana e esses feriados. claude-sonnet-5 também lista um preço de escrita em cache de $1.125 por 1 milhão de tokens.

Uma solicitação de trabalho

Assuma um turno com 2.000 tokens de entrada e 500 tokens de saída. Esse tamanho é nossa suposição, não uma medição. Para gpt-5.5:

• Entrada: 2.000 × $1.50 / 1.000.000 = $0.0030
• Saída: 500 × $9.00 / 1.000.000 = $0.0045
• Total: $0.0075 por turno (estimativa, sem acertos de cache)

Usando a mesma aritmética, obtemos estas estimativas: deepseek-v4-flash custa $0.0006 fora de pico e $0.0012 no pico, claude-sonnet-5 custa $0.00405 e gemini-3.5-flash custa $0.00375. Ao longo de 1.000 turnos, isso é aproximadamente $7.50 no gpt-5.5 contra $0.60 a $1.20 no deepseek-v4-flash. Turnos reais crescem conforme o histórico se acumula, então leia usage da resposta ou do chunk de stream final em vez de confiar neste dimensionamento.

Na prática, roteamos a maioria das conversas para o tier padrão e reservamos gpt-5.5 ou claude-sonnet-5 para turnos que precisam deles. A documentação de cobrança avisa que o menor preço por token nem sempre é o menor custo por tarefa concluída. Execute os mesmos prompts através de dois tiers e compare o custo final em Usage.

Duas verificações antes de enviar. A documentação de cobrança diz para não codificar uma tabela de preços copiada, então leia GET /v1/models/:model/pricing se seu aplicativo exibir custos (Cobrança e preços, observado em 03/10/2026). Além disso, a entrada deepseek-v4-flash lista json-mode, prompt-cache e tool-use, mas não vision. Se seu chatbot aceita imagens, roteie esses turnos para um tier cuja entrada liste vision.

Teste de Fumaça e Envio do seu Chatbot de IA

Passo 7: Execute de ponta a ponta

Inicie o servidor em um terminal onde TOKENLAB_API_KEY esteja exportada. Os payloads curl usam os mesmos campos message, tier e conversation_id que ChatRequest.

uvicorn main:app --reload --port 8000

# 1. Endpoint síncrono: espere uma resposta curta e um conversation_id
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "Reply only with OK.", "tier": "default"}'

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

# 3. Memória: reutilize um conversation_id retornado acima
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. Falha forçada: espere HTTP 400 e "unknown tier: nope"
curl -i -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "Hello!", "tier": "nope"}'

Para forçar uma falha upstream, reinicie o servidor com uma TOKENLAB_API_KEY errada. Seu cliente deve obter uma mensagem 503 limpa, e seus logs devem mostrar o 401 upstream com invalid_api_key. Em seguida, abra Requests para confirmar o modelo, status e uso das chamadas bem-sucedidas. Se você consegue transmitir um turno, preservar uma conversa e retornar um erro limpo em uma falha forçada, você tem a base certa.

Depois disso, as próximas atualizações são infraestrutura chata em vez de mais IA:

• Autenticação, para que um usuário não possa ler a conversa de outro usuário.
• Persistência, para que as sessões sobrevivam a implantações.
• Limitação de taxa por usuário, para que um usuário barulhento não possa queimar sua cota.
• Atribuição de uso, se o chatbot for voltado para o cliente.
• Resumo em segundo plano, se as conversas precisarem de memória de longo prazo.

Uma vez que a migração de URL base esteja atrás de você, uma mudança de modelo é uma edição de configuração em ALLOWED_MODELS.

FAQ

Preciso de websockets para construir um chatbot de IA?

Não. O SSE, mostrado no Passo 4, cobre a maioria dos produtos de chat. Websockets agregam valor quando você precisa de push bidirecional fora da solicitação e resposta, como colaboração ao vivo ou eventos iniciados pelo servidor. Para uma interface de chat padrão, o SSE é mais simples de implantar e depurar.

Qual modelo meu chatbot deve usar por padrão?

Comece com um tier de baixo preço, como deepseek-v4-flash, e adicione claude-sonnet-5 ou gpt-5.5 atrás da lista de permissões para turnos mais difíceis. Os preços acima, observados em 03/10/2026, mostram a diferença. Em seguida, teste seus próprios prompts em dois tiers e compare o custo final em Usage. Verifique o diretório de modelos antes de se comprometer, já que as linhas mudam.

Por que meu chatbot retorna 402 ou 429?

Um 402 significa que o saldo da conta ou o limite de uso da própria chave está muito baixo. Os códigos são insufficient_balance e quota_exceeded, e repetir a solicitação não ajudará. Um 429 significa que você atingiu um limite de solicitação, então aguarde Retry-After e tente novamente. Nosso endpoint síncrono tenta novamente uma vez e o endpoint de streaming relata o erro ao cliente.

O navegador pode chamar a TokenLab diretamente?

Não com segurança. A documentação de autenticação diz para nunca expor chaves de API em código do lado do cliente. Mantenha a chave no ambiente do servidor e faça com que o navegador chame seu próprio backend autenticado, como faz o exemplo de frontend.

---

Crie uma chave de API em tokenlab.sh, adicione fundos e execute o teste de fumaça acima.