Streaming de respostas em tempo real com LLMs: Server-Sent Events para apps responsivas

O problema: respostas que travam a interface

Você já pediu uma resposta de uma IA e ficou olhando a tela trancada por 30 segundos? Esse é o cenário clássico quando você faz uma requisição tradicional (request/response HTTP) para uma LLM. O navegador ou app fica preso esperando a resposta inteira, e nada acontece até ela chegar completa.

Se um usuário está esperando e a primeira palavra demora 5 segundos para aparecer, ele pensa que o app congelou. É uma experiência péssima. A solução é simples: entregar o texto conforme ele é gerado, palavra por palavra, em tempo real.

É aí que Server-Sent Events (SSE) entra como herói.

Por que SSE e não WebSocket?

Primeiro, deixo claro: WebSocket também funciona. Mas para o caso específico de um servidor mandando dados continuamente para o cliente (sem muita interação bidirecional), SSE é mais leve e já vem nativo no navegador.

SSE usa HTTP puro, não precisa de biblioteca pesada no frontend. A conexão fica aberta, e o servidor manda eventos quando quiser. Quando termina a resposta, fecha. Simples.

WebSocket é mais poderoso se você precisa de comunicação de mão dupla pesada (chat em tempo real com muitas trocas rápidas). Para IA gerando resposta? SSE é overkill menor e mais fácil de debugar.

Implementando no backend: Python com FastAPI

Vou mostrar um exemplo real que funciona. Vou usar FastAPI porque é rápido de escrever e suporta SSE nativamente.

Primeiro, você precisa de uma função que chama a LLM (usarei OpenAI aqui, mas vale para Claude, Ollama, etc.) e itera sobre a resposta:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json
import openai

app = FastAPI()
openai.api_key = "sua-chave-aqui"

@app.post("/stream-response")
async def stream_response(user_message: str):
    async def generate():
        response = openai.ChatCompletion.create(
            model="gpt-4",
            messages=[{"role": "user", "content": user_message}],
            stream=True
        )
        
        for chunk in response:
            delta = chunk["choices"][0].get("delta", {})
            if "content" in delta:
                content = delta["content"]
                # Formata como SSE
                yield f"data: {json.dumps({'text': content})}\n\n"
    
    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache"}
    )

O detalhe importante: o header media_type="text/event-stream" avisa ao navegador que é um stream SSE. E o formato data: {json}\n\n (duas quebras de linha no final) é obrigatório para o navegador reconhecer como um evento válido.

Recebendo no frontend: JavaScript puro

Agora o lado do navegador. Nada de npm install gigante:

const button = document.getElementById("send-button");
const output = document.getElementById("response-output");
const inputField = document.getElementById("user-input");

button.addEventListener("click", async () => {
    const userMessage = inputField.value;
    output.innerHTML = ""; // Limpa resposta anterior
    
    const eventSource = new EventSource(
        `/stream-response?user_message=${encodeURIComponent(userMessage)}`
    );
    
    eventSource.onmessage = (event) => {
        const data = JSON.parse(event.data);
        output.innerHTML += data.text;
    };
    
    eventSource.onerror = () => {
        eventSource.close();
        output.innerHTML += "\n[Resposta finalizada]";
    };
});

Vê só? O navegador ouve o stream com EventSource, e cada evento dispara onmessage. Conforme a IA gera a resposta, o texto aparece na tela ao vivo. Nada de travamento.

Gotchas e armadilhas reais

Problema 1: CORS

Se o frontend está em outro domínio (ex.: localhost:3000 chamando localhost:8000), SSE vai dar erro de CORS. Solução: configure CORS no FastAPI:

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Em produção: liste o domínio exato
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Problema 2: Timeout de conexão

Se o server demora muito para responder ou fica muito tempo sem enviar nada, alguns proxies (nginx, cloudflare) podem fechar a conexão. Solução: mande um heartbeat (ping) a cada 15 segundos:

async def generate():
    import asyncio
    
    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[{"role": "user", "content": user_message}],
        stream=True
    )
    
    for chunk in response:
        delta = chunk["choices"][0].get("delta", {})
        if "content" in delta:
            yield f"data: {json.dumps({'text': delta['content']})}\n\n"
        # Heartbeat
        yield ": heartbeat\n\n"
        await asyncio.sleep(0.01)  # Pequeno delay

Problema 3: Errors na resposta

Se a API da LLM retorna erro (quota excedida, rate limit), o stream quebra no meio. Precisa de try/catch e avisar o frontend:

async def generate():
    try:
        response = openai.ChatCompletion.create(..., stream=True)
        for chunk in response:
            delta = chunk["choices"][0].get("delta", {})
            if "content" in delta:
                yield f"data: {json.dumps({'text': delta['content']})}\n\n"
    except Exception as e:
        yield f"data: {json.dumps({'error': str(e)})}\n\n"

E no JavaScript, trate o tipo de evento:

eventSource.onmessage = (event) => {
    const data = JSON.parse(event.data);
    if (data.error) {
        output.innerHTML = `Erro: ${data.error}`;
    } else {
        output.innerHTML += data.text;
    }
};

Dica final: caching e rate limit

Se você tá compilando respostas iguais múltiplas vezes, cache é seu amigo. Redis funciona bem. E rate limit (máximo X requisições por IP por minuto) evita que um usuário malandro queime sua conta de API em segundos.

Na prática, eu faço assim: por requisição, armazeno um hash da mensagem em cache por 1 hora. Se vem a mesma pergunta de novo, retorno a resposta em cache em streaming também (lê do cache e faz yield).

Pronto. Agora seu app vai responder em tempo real, sem travar, e o usuário vai pensar que é magia. A partir daqui, é só iterar: adicione tipagem (TypeScript), melhor CSS para a saída, autosave de histórico, tudo que quiser. A base SSE fica robusta e pronta para escala.