Construindo uma UI de Chat com Deltas de Token Incrementais

Uma interface de chat que espera a resposta completa de Claude antes de mostrar qualquer coisa parece lenta, mesmo quando o tempo total de resposta é o mesmo.

Renderizar eventos text_delta à medida que chegam - atualizando uma bolha de mensagem token por token - é o que faz uma UI de chat parecer viva.

Esta página constrói esse loop de renderização desde um protótipo de terminal até um padrão que você pode conectar a um endpoint de backend real.

Resumo

Uma UI de chat construída em streaming tem três camadas: um consumidor de stream que lê eventos, um acumulador que rastreia o texto crescente da mensagem e um renderizador que redesenha a UI a cada atualização.

O iterador text_stream do SDK Python anthropic lida com a primeira camada para o caso comum de texto puro.

O acumulador é apenas uma string mutável (ou lista de fragmentos) que sua aplicação atualiza a cada delta.

O renderizador depende da sua stack - um print de terminal, um send de WebSocket, ou uma resposta Server-Sent Events de volta para um navegador - mas o padrão alimenta todos eles com o mesmo texto acumulado.

Esta página mostra primeiro a versão de terminal (mais rápida para verificar), depois um endpoint FastAPI que retransmite a saída de Claude para um navegador.

Receita

import anthropic
 
client = anthropic.Anthropic()
 
def stream_chat_message(prompt: str) -> str:
    full_text = ""
    with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        for delta in stream.text_stream:
            full_text += delta
            render(full_text)  # sua função de atualização de UI
    return full_text
 
def render(current_text: str) -> None:
    print(f"\r{current_text}", end="", flush=True)

Quando usar isso:

• Qualquer interface de chat, assistente ou copiloto onde um usuário está observando ativamente a geração da resposta.
• Painéis de geração de documentos ou código ao vivo que devem ser preenchidos progressivamente.
• Pipelines de voz ou narração que desejam iniciar a conversão de texto em fala na primeira frase, em vez da resposta completa.
• Respostas longas onde mostrar progresso parcial reduz a latência percebida, mesmo que o tempo total não seja alterado.

Exemplo Funcional

Um backend FastAPI mínimo que retransmite um turno de chat para o navegador via Server-Sent Events, e um consumidor correspondente no lado do navegador.

# server.py
import json
import anthropic
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
 
app = FastAPI()
client = anthropic.Anthropic()
 
class ChatRequest(BaseModel):
    message: str
 
def event_generator(user_message: str):
    with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": user_message}],
    ) as stream:
        for delta in stream.text_stream:
            payload = json.dumps({"delta": delta})
            yield f"data: {payload}\n\n"
    yield "data: [DONE]\n\n"
 
@app.post("/chat")
def chat(request: ChatRequest):
    return StreamingResponse(
        event_generator(request.message),
        media_type="text/event-stream",
    )

// client.js (navegador)
async function sendMessage(message, onDelta) {
  const response = await fetch("/chat", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ message }),
  });
  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 lines = buffer.split("\n\n");
    buffer = lines.pop();
    for (const line of lines) {
      if (!line.startsWith("data: ")) continue;
      const data = line.slice(6);
      if (data === "[DONE]") return;
      onDelta(JSON.parse(data).delta);
    }
  }
}

O que isso demonstra:

• O stream SSE de Claude é reempacotado em um segundo stream SSE de nível de aplicação para que o navegador nunca fale diretamente com a API da Anthropic (mantendo a chave de API no lado do servidor).
• Cada text_delta se torna um pequeno payload JSON que o navegador anexa a uma bolha de mensagem crescente.
• Um sentinela [DONE] diz ao navegador quando parar de ler, separado do próprio message_stop de Claude.
• O navegador lê bytes brutos e divide manualmente em linhas em branco - este é o mesmo enquadramento SSE descrito em Como Server-Sent Events Potencializam as Respostas de Streaming de Claude, apenas implementado manualmente no lado do cliente.

Análise Detalhada

Como Funciona

• stream.text_stream filtra o stream de eventos brutos para incluir apenas fragmentos text_delta, em ordem, descartando marcadores de bloco e eventos de metadados.
• O padrão acumulador - anexar cada delta a uma string em execução - é o que permite que uma UI redesenhe a mensagem completa a cada atualização, em vez de apenas anexar, o que é importante para UIs que re-renderizam a partir do estado (como React) em vez de mutar o DOM diretamente.
• Retransmitir através do seu próprio backend (em vez de expor a chave da API da Anthropic ao navegador) é uma prática padrão; seu endpoint se torna um simples retransmissor que remodela o stream SSE de Claude para qualquer transporte que seu frontend espera.
• Terminar seu próprio stream com um sentinela explícito ([DONE]) desacopla o sinal de "stream finalizado" do seu frontend do evento message_stop de Claude, o que é útil se você quiser enviar metadados de acompanhamento (como uso) após o término do texto.

Estratégias de Renderização em Resumo

Notas Python

# Agrupar deltas pequenos em menos atualizações de UI usando um simples flush baseado em temporizador.
import time
 
def stream_with_batched_updates(prompt: str, flush_interval: float = 0.05):
    buffer = []
    last_flush = time.monotonic()
    full_text = ""
    with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        for delta in stream.text_stream:
            buffer.append(delta)
            if time.monotonic() - last_flush >= flush_interval:
                full_text += "".join(buffer)
                render(full_text)
                buffer.clear()
                last_flush = time.monotonic()
        if buffer:
            full_text += "".join(buffer)
            render(full_text)
    return full_text

Armadilhas

• Redesenhar toda a string da mensagem a cada delta individual - aceitável em uma demonstração de terminal, mas caro em uma UI de navegador com muitos nós DOM. Correção: agrupe deltas em um curto intervalo (25-50ms) antes de acionar um redesenho.
• Enviar a chave de API para o navegador para chamar Claude diretamente - expõe suas credenciais a qualquer pessoa que abra as ferramentas de desenvolvedor. Correção: sempre retransmita através do seu próprio endpoint de backend, como mostrado acima.
• Assumir que text_delta quebra em limites de palavras ou frases - não o faz, e o código que tenta detectar "fim de frase" no meio de um delta falhará. Correção: acumule na string acumulada e execute a detecção de frases nela, não em deltas individuais.
• Esquecer de fechar a resposta SSE quando o cliente se desconecta - o fechamento de uma aba do navegador no meio de um stream deixa seu backend ainda consumindo o stream de Claude e pagando por tokens que ninguém vê. Correção: verifique request.is_disconnected() no FastAPI (ou o equivalente para seu framework) e saia do gerador.
• Misturar text_stream com iteração manual de eventos no mesmo objeto de stream - o stream do SDK é um iterador de passagem única; consumi-lo de duas maneiras diferentes produz resultados incompletos. Correção: escolha um estilo de consumo (text_stream ou for event in stream bruto) por stream e mantenha-o.
• Não lidar com message_stop antes de chamar get_final_message() - chamá-lo muito cedo pode retornar uma mensagem parcialmente construída. Correção: chame get_final_message() apenas após a iteração do stream ter sido totalmente concluída.

Alternativas

FAQs

with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=200,
    messages=[{"role": "user", "content": "Oi"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

Relacionados

• Como Server-Sent Events Potencializam as Respostas de Streaming de Claude - o modelo de eventos que esta página renderiza.
• Noções Básicas de Respostas de Streaming - o loop de streaming em nível de terminal no qual este se baseia.
• Melhores Práticas de Streaming - buffer, reconexões e tratamento de erros para UIs de chat de produção.
• Transmitindo Blocos de Pensamento Estendidos para o Frontend - renderizando um segundo stream (raciocínio) ao lado do texto do chat.
• Respostas de Streaming com o SDK Python - cobertura mais aprofundada do helper de streaming do SDK.

Versões da Stack: Escrito contra a linha de modelos Claude atual em ~junho de 2026 - Claude Fable 5, Claude Opus 4.8, Claude Sonnet 5 (o padrão), e Claude Haiku 4.5 - e o SDK oficial anthropic para Python (última versão 0.x). Nomes de modelos, versões de SDK e preços mudam rapidamente - verifique as especificações atuais em platform.claude.com/docs antes de confiar nelas.