Este tutorial cria um serviço de chatbot pequeno, porém pronto para produção, com FastAPI, streaming SSE, memória de conversão e troca de modelos. O objetivo não é uma demonstração simples. O objetivo é um backend que você possa colocar atrás de uma interface de produto real e iterar com segurança, sem reescrever sua integração toda vez que mudar de modelo.
Se você já apontou um SDK compatível com OpenAI para o TokenLab, este artigo continua a partir daí. Se você ainda não fez a troca da base URL, leia o guia de migração primeiro. Se sua principal preocupação é a formatação de requisições e backoff sob carga, combine este guia com o guia de limitação de taxa (rate limiting) 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 (allowlist) de modelos, tratamento de erros real e um caminho claro para upgrade de armazenamento.
- Valide sua chave, base URL e roteamento com um pequeno endpoint de chat antes de adicionar streaming, memória ou chamadas de ferramentas.
- O streaming SSE atende à maioria dos produtos de chat e traz menos sobrecarga operacional do que websockets.
- Exponha modelos através de uma allowlist no backend, não em um campo de texto livre, para que o frontend não possa solicitar IDs de modelos arbitrários ou descontinuados.
- A disponibilidade e a linha de modelos mudam com frequência. Verifique o diretório de modelos do TokenLab (observado em 07/07/2026) antes de fixar sua allowlist em produção.
O que estamos construindo
O serviço final possui seis partes móveis:
- Um endpoint
/chatsíncrono para testes básicos. - Um endpoint
/chat/streamde streaming para a interface real. - Estado de conversação indexado por
conversation_id. - Uma allowlist de modelos para que o frontend não possa solicitar IDs arbitrários.
- Tratamento de erros que não colapsa no primeiro 429.
- Um caminho claro de um protótipo em memória para Redis ou PostgreSQL.
Isso é o suficiente para alimentar um bot de suporte, um assistente interno ou a primeira versão de um widget de chat integrado.
Instale a Stack Mínima
pip install fastapi uvicorn openai pydantic redis
Você pode pular o redis na primeira etapa, mas configurar a importação agora torna o upgrade posterior um evento simples em vez de uma refatoração.
Passo 1: Comece com um endpoint de chat pequeno e simples
A maneira mais rápida de se perder no desenvolvimento de um chatbot é começar com websockets, uso de ferramentas e orquestração de agentes antes que o caminho básico da requisição esteja estável. Comece com um pequeno endpoint que prove que sua chave, base URL e roteamento de modelo funcionam.
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}
Execute um teste básico. Se isso falhar, não continue construindo sobre ele.
Passo 2: Adicione streaming, pois os usuários sentem a latência antes de medi-la
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. O SSE é suficiente para a maioria dos produtos de chat e tem uma carga operacional menor do que 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")
No frontend, um leitor fetch simples ainda é 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);
}
}
Se o seu produto já executa um cliente de navegador sobre HTTP padrão, o SSE mantém a arquitetura mais simples do que uma camada de websocket.
Passo 3: Mova o estado da conversação para fora do corpo da requisição
A primeira demonstração de chatbot geralmente mantém a transcrição completa no navegador e a reenvia a cada turno. Isso funciona para protótipos. Fica bagunçado rapidamente quando você precisa de novas tentativas, sessões retomáveis ou ferramentas no lado do servidor.
Um armazenamento em memória é bom para começar:
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
O caminho de upgrade para o Redis é basicamente uma questão de infraestrutura de armazenamento, não de mudanças na 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))
Use o Redis quando as conversas precisarem de TTL, retomabilidade ou implantação em múltiplas instâncias. Use o PostgreSQL quando a própria transcrição for um dado do produto que você precisa consultar, auditar ou gerar relatórios posteriormente.
Passo 4: Trate erros como comportamento do produto, não apenas exceções
Se o seu chatbot é voltado para o cliente, o caminho de falha é tão importante quanto o caminho de sucesso. O usuário não se importa se a falha veio de limitação de taxa, saldo esgotado ou uma interrupção do modelo no upstream. Eles se importam se a interface congelar.
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")
Passo 5: Bloqueie quais modelos o frontend pode solicitar
Nunca deixe o frontend passar uma string de modelo arbitrária diretamente para a API. Um campo de texto livre convida solicitações de modelos descontinuados, erros de digitação ou modelos que você nunca pretendeu expor. Roteie através de uma allowlist no 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"])
# o restante da lógica de streaming usa `model` em vez de uma string bruta fornecida pelo cliente
Isso lhe dá um único lugar para trocar modelos quando um provedor descontinua um, sem tocar no código do frontend ou enviar uma atualização do cliente.
Passo 6: Lide com o restante da produção, não apenas o caminho feliz
Um backend de chatbot é considerado pronto para produção quando as bordas ao redor são tratadas, não quando a chamada central de chat se torna inteligente.
A lista de verificação é curta:
- adicione IDs de requisição para que você possa conectar falhas do frontend aos logs do backend
- limite a concorrência por usuário e o tamanho da requisição
- corte históricos longos antes que eles estourem seu orçamento de tokens
- registre o modelo, latência, tamanho da entrada e motivo da finalização
- separe mensagens de erro visíveis ao usuário de detalhes de erro internos
- teste um modelo alternativo para saber se o fallback funciona antes da primeira interrupção real
O corte de histórico pode ser simples:
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
O objetivo não é uma contabilidade perfeita de tokens. O objetivo é impedir estouros óbvios de contexto antes que eles atinjam sua fatura ou sua latência.
Da demonstração ao produto
Uma vez que este backend esteja estável, o próximo upgrade raramente é "mais IA". Geralmente é infraestrutura básica:
- autenticação para que um usuário não possa ler a conversa de outro
- persistência para que as sessões sobrevivam a deploys
- limitação de taxa para que um usuário barulhento não consuma sua cota
- faturamento ou atribuição de uso se o chatbot for voltado para o cliente
- sumarização em segundo plano se as conversas precisarem de memória de longo prazo
Um gateway unificado ajuda com a maior parte disso. Uma vez que a migração da base URL esteja concluída, as mudanças de modelo deixam de ser uma reescrita da plataforma e passam a ser uma edição de configuração.
Teste básico
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"}'
Se você consegue transmitir um turno, preservar uma conversa e retornar um erro limpo em uma falha forçada, você tem a base correta.
Estimativa de custo
Crie uma API key no TokenLab, aponte seu SDK OpenAI para https://api.tokenlab.sh/v1, e você poderá enviar a primeira versão de produção do seu chatbot sem gerenciar contas separadas entre provedores.
| Modelo | Nível Típico | Notas |
|---|---|---|
| DeepSeek V4 Flash | Rápido / padrão | Bom padrão para turnos de alto volume e baixa latência |
| GPT-5.5 | Flagship | Use para turnos que precisam de raciocínio mais forte |
| Claude Sonnet 5 | Equilibrado | Ótima escolha para codificação e respostas de revisão |
| Gemini 3.5 Flash | Econômico / alt rápida | Alternativa rápida e de baixo custo para roteamento de alto volume |
O preço exato por token muda frequentemente entre os provedores e não é reproduzido aqui como números fixos. Verifique as taxas atuais no diretório de modelos (observado em 07/07/2026) antes de fazer um orçamento. Na prática, rotear a maioria das conversas para um nível rápido/padrão como DeepSeek V4 Flash ou Gemini 3.5 Flash e reservar o GPT-5.5 ou Claude Sonnet 5 para turnos que precisam disso mantém a maioria das aplicações com uma fatura mensal baixa, mas confirme as taxas reais por milhão de tokens para sua conta antes de se comprometer com um orçamento.
FAQ
Preciso de websockets para construir um chatbot de IA? Não. O streaming SSE, mostrado no Passo 2, cobre a grande maioria dos produtos de chat. Websockets agregam valor real quando você precisa de push bidirecional fora do padrão requisição/resposta, como colaboração ao vivo ou eventos iniciados pelo servidor. Para uma interface de chat padrão, o SSE é mais simples de implantar, depurar e escalar.
Como sei qual modelo definir como padrão? Comece com um modelo rápido e de baixo custo, como o DeepSeek V4 Flash ou Gemini 3.5 Flash para o nível padrão, e adicione um nível equilibrado ou de raciocínio no Claude Sonnet 5 ou GPT-5.5 atrás da allowlist mostrada no Passo 5. Verifique o diretório de modelos (observado em 07/07/2026) para opções atuais, já que novos modelos são lançados e os antigos são descontinuados em um cronograma fora do seu controle.
O que quebra primeiro quando um chatbot passa de uma demonstração para tráfego real? Quase sempre o caminho de erro, não o caminho feliz. Novas tentativas ilimitadas, falta de limites de concorrência por usuário e histórico de conversação ilimitado são as três causas mais comuns de um backend de chatbot cair sob carga real. Os Passos 4 e 6 acima abordam todos os três diretamente.
Comece com sua API key: tokenlab.sh fornece mais de 300 modelos através de um único endpoint. $1 de crédito gratuito para começar a construir.
Fontes
Preço observado em 2026-07-07
- TokenLab model directoryObservado em 2026-07-07
