Este tutorial constrói um serviço de chatbot pequeno, mas pronto para produção, com FastAPI, streaming SSE, memória de conversação e troca de modelos. O objetivo não é entregar uma demo de brinquedo. O objetivo é chegar a um backend que você possa realmente colocar por trás da interface de um produto e iterar com segurança.
Se você já apontou um SDK compatível com OpenAI para a LemonData, este artigo continua de onde você parou. Se você ainda não fez a troca da base URL, leia o guia de migração primeiro. Se sua principal preocupação é o ajuste de requisições e backoff sob carga, combine este guia com o guia de rate limiting de API de IA.
O Que Estamos Construindo
O serviço finalizado tem seis partes móveis:
- Um endpoint
/chatsíncrono para smoke tests. - Um endpoint
/chat/streamde streaming para a UI real. - Estado da conversação identificado 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 do 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 incorporado.
Instale a Stack Mínima
pip install fastapi uvicorn openai pydantic redis
Você pode omitir o redis na primeira passagem, mas é útil configurar a importação agora para que o caminho de upgrade seja óbvio.
Passo 1: Comece com um Endpoint de Chat Pequeno e Simples
A maneira mais rápida de se perder no trabalho com chatbots é começar com websockets, tool use e orquestração de agentes antes que o caminho básico da requisição esteja estável. Comece com um endpoint pequeno que prove que sua API key, base URL e roteamento de modelos estão corretos.
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}
Execute um smoke test. Se isso falhar, não continue adicionando camadas de funcionalidades por cima.
Passo 2: Adicione Streaming porque 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 UI fica em branco até que a resposta completa chegue. SSE é suficiente para muitos 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, o cliente mais simples do lado do navegador ainda é bom o 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á usa um cliente de navegador e HTTP padrão, o SSE mantém a arquitetura mais simples.
Passo 3: Mova o Estado da Conversação para fora do Corpo da Requisição
A primeira demo de um chatbot geralmente mantém a transcrição completa no navegador e a envia a cada turno. Isso funciona para protótipos. Torna-se confuso no momento em que você precisa de retentativas, sessões retomáveis ou ferramentas no lado do servidor.
Um armazenamento em memória é suficiente 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 encanamento de armazenamento:
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 Redis se as conversas precisarem de TTL, retomada ou implantação em múltiplas instâncias. Use PostgreSQL se a própria transcrição for um dado do produto.
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 importa tanto quanto o caminho de sucesso. Um usuário não se importa se a falha veio de rate limiting, saldo ou uma queda do modelo. Ele se importa se a UI trava.
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}
)
Se você estiver atendendo a uma carga significativa, também deve moldar as requisições antes que elas cheguem ao upstream. Os padrões detalhados estão no guia de rate limiting, mas a versão curta é: use retentativas limitadas, use jitter e nunca faça except Exception: time.sleep(1).
Passo 5: A Troca de Modelos Precisa de uma Allowlist, não de uma Caixa de Texto Livre
Uma única API key pode acessar centenas de modelos. Isso não significa que sua UI deva expor centenas de modelos. O backend deve publicar uma pequena allowlist adequada ao seu 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}
Isso faz três coisas úteis:
- impede que o frontend solicite IDs de modelos inválidos ou obsoletos
- permite remapear um nível (tier) mais tarde sem precisar reimplantar cada cliente
- oferece um lugar único para aplicar controles de custo
Se sua equipe ainda está decidindo quais provedores padronizar, a comparação de preços e a comparação OpenRouter vs LemonData são as duas páginas que valem a pena ler antes de fechar a allowlist.
Passo 6: Adicione as Bordas de Produção antes que o Tráfego Chegue
Um backend de chatbot torna-se de nível de produção quando as bordas circundantes são tratadas, não quando a chamada de chat principal é inteligente.
O checklist é curto:
- adicione IDs de requisição para que você possa conectar falhas no 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 modelo, latência, tamanho da entrada e motivo de finalização (finish reason)
- separe as mensagens de erro visíveis ao usuário dos detalhes de erro internos
- teste um modelo alternativo para saber que o fallback funciona antes da primeira interrupção
O corte de histórico pode continuar 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 ponto não é uma contabilidade perfeita de tokens. O ponto é impedir explosões óbvias de contexto.
Da Demo ao Produto
Uma vez que este backend esteja estável, o próximo upgrade raramente é “mais IA”. Geralmente é infraestrutura entediante:
- 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
- rate limiting para que um usuário barulhento não queime 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
É por isso que um gateway unificado ajuda. Depois de fazer a migração da base URL, as mudanças de modelo deixam de ser uma reescrita da plataforma e tornam-se configuração.
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"}'
Se você puder fazer o streaming de um turno, preservar uma conversa e retornar um erro limpo em uma falha forçada, você tem a base certa.
Estimativa de Custo
Crie uma API key na LemonData, aponte seu SDK da OpenAI para https://api.lemondata.cc/v1, e você poderá entregar a primeira versão de produção do seu chatbot sem gerenciar contas de provedores separadas.
| Modelo | Custo Diário | Custo Mensal |
|---|---|---|
| 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 a maioria das conversas e fazer o upgrade para Claude Sonnet 4.6 apenas quando os usuários solicitarem mantém os custos abaixo de $100/mês para a maioria das aplicações.
Obtenha sua API key: lemondata.cc fornece mais de 300 modelos através de um único endpoint. $1 de crédito grátis para começar a construir.