Creación de una interfaz de chat con deltas de tokens incrementales

Una interfaz de chat que espera la respuesta completa de Claude antes de mostrar algo se siente lenta, incluso cuando el tiempo total de respuesta es el mismo.

Renderizar eventos text_delta a medida que llegan —actualizando una burbuja de mensaje token por token— es lo que hace que una interfaz de chat se sienta viva.

Esta página construye ese bucle de renderizado desde un prototipo de terminal hasta un patrón que puedes integrar en un endpoint de backend real.

Resumen

Una interfaz de chat construida sobre streaming tiene tres capas: un consumidor de stream que lee eventos, un acumulador que rastrea el texto creciente del mensaje y un renderizador que redibuja la interfaz de usuario con cada actualización.

El iterador text_stream del SDK de Python anthropic maneja la primera capa para el caso común de texto plano.

El acumulador es solo una cadena mutable (o lista de fragmentos) que tu aplicación actualiza en cada delta.

El renderizador depende de tu stack —un print de terminal, un send de WebSocket o una respuesta de Server-Sent Events de vuelta a un navegador— pero el patrón les proporciona a todos el mismo texto acumulado.

Esta página muestra primero la versión de terminal (la más rápida de verificar) y luego un endpoint de FastAPI que retransmite la salida de Claude a un navegador.

Receta

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)  # tu función de actualización de UI
    return full_text
 
def render(current_text: str) -> None:
    print(f"\r{current_text}", end="", flush=True)

Cuándo usar esto:

• Cualquier interfaz de chat, asistente o copiloto donde un usuario esté observando activamente la generación de la respuesta.
• Paneles de generación de documentos o código en vivo que deberían completarse progresivamente.
• Pipelines de voz o narración que desean iniciar la conversión de texto a voz en la primera oración en lugar de la respuesta completa.
• Respuestas largas donde mostrar progreso parcial reduce la latencia percibida, incluso si el tiempo total no cambia.

Ejemplo funcional

Un backend mínimo de FastAPI que retransmite un turno de chat al navegador a través de Server-Sent Events, y un consumidor del lado del navegador que coincide.

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

Lo que esto demuestra:

• El stream SSE de Claude se reenvuelve en un segundo stream SSE a nivel de aplicación para que el navegador nunca hable directamente con la API de Anthropic (manteniendo la clave API en el lado del servidor).
• Cada text_delta se convierte en una pequeña carga útil JSON que el navegador agrega a una burbuja de mensaje en crecimiento.
• Un centinela [DONE] le dice al navegador cuándo dejar de leer, separado del propio message_stop de Claude.
• El navegador lee bytes sin procesar y divide manualmente en líneas en blanco; este es el mismo enmarcado SSE descrito en Cómo Server-Sent Events potencia las respuestas en streaming de Claude, implementado manualmente en el lado del cliente.

Análisis detallado

Cómo funciona

• stream.text_stream filtra el stream de eventos sin procesar para obtener solo fragmentos text_delta, en orden, descartando marcadores de bloque y eventos de metadatos.
• El patrón acumulador —agregar cada delta a una cadena en ejecución— es lo que permite que una interfaz de usuario redibuje el mensaje completo en cada actualización en lugar de solo agregar, lo cual es importante para interfaces de usuario que se vuelven a renderizar desde el estado (como React) en lugar de mutar el DOM directamente.
• Re-transmitir a través de tu propio backend (en lugar de exponer la clave API de Anthropic al navegador) es una práctica estándar; tu endpoint se convierte en un delgado relé que remodela el stream SSE de Claude en el transporte que tu frontend espera.
• Terminar tu propio stream con un centinela explícito ([DONE]) desacopla la señal de "stream finalizado" de tu frontend de message_stop de Claude, lo cual es útil si alguna vez deseas enviar metadatos de seguimiento (como el uso) después de que el texto finalice.

Estrategias de renderizado de un vistazo

Notas de Python

# Agrupa deltas pequeños en menos actualizaciones de UI usando un simple vaciado basado en 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

Errores comunes

• Redibujar toda la cadena del mensaje en cada delta individual — está bien en una demostración de terminal, pero es costoso en una interfaz de usuario de navegador con muchos nodos DOM. Solución: agrupa los deltas durante un corto intervalo (25-50 ms) antes de activar un re-renderizado.
• Enviar la clave API al navegador para llamar a Claude directamente — expone tus credenciales a cualquiera que abra las herramientas de desarrollador. Solución: siempre retransmite a través de tu propio endpoint de backend, como se muestra arriba.
• Asumir que text_delta se divide en límites de palabras o frases — no lo hace, y el código que intenta detectar "fin de frase" a mitad de delta fallará. Solución: acumula en la cadena acumulada y ejecuta la detección de frases sobre eso, no sobre deltas individuales.
• Olvidar cerrar la respuesta SSE cuando el cliente se desconecta — una pestaña del navegador que se cierra a mitad de stream deja a tu backend consumiendo el stream de Claude y pagando por tokens que nadie ve. Solución: verifica request.is_disconnected() en FastAPI (o el equivalente para tu framework) y sal del generador.
• Mezclar text_stream con la iteración manual de eventos en el mismo objeto de stream — el stream del SDK es un iterador de un solo paso; consumirlo de dos maneras diferentes produce resultados incompletos. Solución: elige un estilo de consumo (text_stream o for event in stream sin procesar) por stream y cíñete a él.
• No manejar message_stop antes de llamar a get_final_message() — llamarlo demasiado pronto puede devolver un mensaje parcialmente construido. Solución: solo llama a get_final_message() después de que la iteración del stream se haya completado por completo.

Alternativas

Preguntas frecuentes

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

Relacionado

• Cómo Server-Sent Events potencia las respuestas en streaming de Claude - el modelo de eventos que esta página renderiza.
• Conceptos básicos de respuestas en streaming - el bucle de streaming a nivel de terminal en el que se basa esto.
• Mejores prácticas de streaming - buffering, reconexiones y manejo de errores para interfaces de chat de producción.
• Transmisión de bloques de pensamiento extendidos al frontend - renderizado de un segundo stream (razonamiento) junto con el texto del chat.
• Respuestas en streaming con el SDK de Python - cobertura más profunda del asistente de streaming del SDK.

Versiones de stack: Escrito contra la línea de modelos de Claude actual a partir de ~junio de 2026 — Claude Fable 5, Claude Opus 4.8, Claude Sonnet 5 (el predeterminado) y Claude Haiku 4.5 — y el SDK oficial de Python anthropic (última versión 0.x). Los nombres de modelos, versiones de SDK y precios cambian rápidamente; verifica los detalles actuales en platform.claude.com/docs antes de confiar en ellos.