Streaming de Claude, Gemini y GPT con el SDK de OpenAI (SSE)

El streaming es la diferencia entre un cuadro de chat que se queda muerto durante ocho segundos y otro que empieza a escribir en menos de un segundo. La mecánica es la misma tanto si el modelo detrás de tu base_url es Claude, Gemini o GPT: activa stream=True, itera los chunks, lee el delta de cada uno y detente en el centinela [DONE]. Como Brievio habla el protocolo Chat Completions de OpenAI, el mismo bucle exacto funciona en todos los modelos first-party genuinos: cambias la cadena model y nada más.

Este artículo cubre cómo funciona realmente el streaming con Server-Sent Events a través de un endpoint compatible con OpenAI, cómo obtener el uso de tokens exacto en el chunk final con stream_options, el patrón idéntico en Python y Node, y los dos modos de fallo silencioso que hacen que un flujo parezca correcto mientras te traiciona en silencio: el streaming falso (con búfer) y el usage ausente.

Qué significa «streaming» sobre HTTP

Una llamada sin streaming es una petición y una respuesta: el servidor piensa unos segundos y luego te entrega la completación entera de golpe. El streaming mantiene la conexión HTTP abierta y va empujando la respuesta por partes a medida que el modelo la genera, usando Server-Sent Events (SSE). En el cable, cada parte llega como una línea que empieza con data: seguido de un objeto JSON, y el flujo termina con una línea literal data: [DONE].

Casi nunca parseas ese texto tú mismo: lo hace el SDK. Lo que recibes en el código es un iterable de chunks. Cada chunk se parece a un objeto de completación normal, salvo que el contenido vive en choices[0].delta en lugar de choices[0].message, y solo contiene el fragmento generado desde el chunk anterior. Concatena cada delta.content en orden y habrás reconstruido el mensaje completo. La única métrica que importa aquí es el tiempo hasta el primer token (TTFB): cuánto tarda en aparecer el primer delta no vacío. Ese número es toda la razón para hacer streaming.

El patrón en Python

Aquí tienes todo el asunto: un bucle de streaming real que además captura el usage. La única línea específica de Brievio es el base_url:

from openai import OpenAI

client = OpenAI(
    api_key="sk-brievio-...",
    base_url="https://api.brievio.com/v1",   # un solo base_url, modelos first-party genuinos
)

# stream=True convierte la respuesta en un flujo de Server-Sent Events.
# Iteras el objeto; cada elemento es un chunk con un "delta" parcial.
stream = client.chat.completions.create(
    model="claude-sonnet-4-6",               # o gemini-2.5-flash, gpt-..., etc.
    messages=[{"role": "user", "content": "Explain Raft consensus in 200 words."}],
    stream=True,
    stream_options={"include_usage": True},  # pide el usage en el chunk FINAL
)

usage = None
for chunk in stream:
    # El último evento de datos antes de [DONE] lleva el usage y una lista de choices vacía.
    if chunk.usage is not None:
        usage = chunk.usage
        continue
    delta = chunk.choices[0].delta.content
    if delta:
        print(delta, end="", flush=True)     # renderiza los tokens conforme llegan

print()
# usage solo está poblado porque se activó include_usage. Estos son conteos reales.
print(usage.prompt_tokens, usage.completion_tokens, usage.total_tokens)

Tres detalles con los que la gente tropieza. Primero, el delta de contenido puede ser None o estar vacío en algunos chunks (el chunk inicial a menudo solo fija el rol), así que protégete antes de imprimir. Segundo, el chunk que lleva el usage llega después de que el contenido haya terminado y trae una lista de choices vacía — por eso el ejemplo comprueba chunk.usage primero y hace continue. Tercero, no buscas [DONE] tú mismo; el SDK consume ese centinela y cierra el iterador por ti. Si llamaras al endpoint con requests o fetch en crudo, tendrías que dividir por saltos de línea y cortar en [DONE] a mano.

El mismo bucle en Node

El SDK de Node expone el flujo como un iterable asíncrono, así que la estructura es idéntica — for await ... of en lugar de for, y process.stdout.write en lugar de un print con flush:

import OpenAI from "openai";

const client = new OpenAI({
  apiKey: "sk-brievio-...",
  baseURL: "https://api.brievio.com/v1",
});

// El mismo contrato en Node: stream=true devuelve un iterable asíncrono de chunks.
const stream = await client.chat.completions.create({
  model: "claude-sonnet-4-6",
  messages: [{ role: "user", content: "Explain Raft consensus in 200 words." }],
  stream: true,
  stream_options: { include_usage: true },
});

let usage = null;
for await (const chunk of stream) {
  // Evento final: choices está vacío, usage está presente.
  if (chunk.usage) {
    usage = chunk.usage;
    continue;
  }
  const delta = chunk.choices[0]?.delta?.content;
  if (delta) process.stdout.write(delta); // vuelca cada token a la terminal
}

console.log();
console.log(usage?.prompt_tokens, usage?.completion_tokens, usage?.total_tokens);

Fíjate en el encadenamiento opcional (chunk.choices[0]?.delta?.content). En el chunk final que lleva el usage, choices está vacío, así que indexar [0] sin la protección lanzaría justo en la línea de meta. Esta es la razón más común, con diferencia, de que un manejador de streaming en Node se caiga en el último evento tras haber funcionado perfectamente durante toda la respuesta.

Obtener el usage real en el último chunk

Por defecto, una respuesta en streaming no incluye conteos de tokens — no hay ningún objeto usage en ningún chunk. Eso es una parte deliberada del protocolo de OpenAI, y muerde a los equipos que hacen streaming en producción y luego no pueden cuadrar su factura. El arreglo es un solo parámetro:

• Activa stream_options={"include_usage": True} (Python) o stream_options: { include_usage: true } (Node).
• El servidor emite entonces un chunk extra justo antes de [DONE] cuyo choices está vacío y cuyo usage contiene prompt_tokens, completion_tokens y total_tokens.
• En Brievio esos son los conteos genuinos reportados por el modelo — los mismos números que obtendrías de una llamada sin streaming, facturados a aproximadamente un 15% por debajo de la tarifa oficial. No hay un objeto de usage inflado ni un system prompt inyectado que hinche el lado del prompt.

Si te saltas include_usage y aun así necesitas una estimación de tokens, tu única opción es contar en local con el tokenizador del modelo — algo aproximado y una carga de mantenimiento. Mejor activa el flag.

Los fallos silenciosos: streaming falso y usage ausente

Dos modos de fallo pasan un vistazo superficial y solo afloran bajo escrutinio. Ambos merecen una comprobación de 20 segundos antes de confiarle tráfico real a un gateway.

• Streaming «falso» con búfer. Algunos gateways aceptan stream=True, esperan a la completación upstream entera y luego te la reproducen como una ráfaga de chunks al final. Tu bucle se ejecuta, los deltas llegan, todo parece transmitido — pero el TTFB es idéntico al de una llamada sin streaming porque no se envió nada hasta que el modelo terminó. La señal es simple: mide el intervalo entre enviar la petición y el primer delta no vacío. En el streaming genuino aterriza muy por debajo de un segundo; en una reproducción con búfer equivale al tiempo de generación completo. Si la latencia del primer token sigue a la latencia total, no estás haciendo streaming, estás viendo una grabación.
• Usage ausente o fabricado. Un gateway que no respeta include_usage te deja sin conteos de tokens en las llamadas con streaming — así que cuadras tu factura contra el aire. Peor aún, uno deshonesto puede adjuntar un objeto usage con números inflados, porque en un flujo el cliente rara vez vuelve a contar. Verifícalo de la forma aburrida: ejecuta el mismo prompt una vez con streaming y otra sin él, y confirma que el usage del chunk final con streaming coincide con el usage sin streaming. Deberían ser idénticos.
• Errores a mitad de flujo que parecen un final limpio. Si el modelo upstream falla a medio camino, un gateway correcto lo expone como una excepción en tu bucle, no como un truncamiento silencioso. Comprueba siempre que recibiste una razón de finalización (o el chunk de usage) antes de tratar el texto como completo — un flujo que simplemente se detiene no es lo mismo que un flujo que terminó.

La conclusión

El streaming sobre un endpoint compatible con OpenAI son cuatro piezas en movimiento: stream=True, iterar los chunks, leer cada delta y dejar que el SDK gestione [DONE]. Añade stream_options={"include_usage": True} y además obtienes conteos de tokens honestos en el chunk final. Las mismas quince líneas funcionan sin cambios en Claude Sonnet 4.6, Gemini 2.5 Flash y la familia GPT detrás de un solo base_url — cambia la cadena del modelo, conserva el bucle.

Antes de desplegar, mide el tiempo hasta el primer token y compara el usage con streaming frente al de sin streaming. El streaming real te da un TTFB por debajo del segundo y conteos que coinciden; una reproducción con búfer se delata por la latencia. En Brievio, las llamadas fallidas 4xx/5xx no se facturan, así que puedes ejecutar estas comprobaciones gratis. Consulta la referencia de Chat Completions para la lista completa de parámetros, el resto de la documentación de la API para herramientas y visión sobre el mismo flujo, la guía para llamar a Claude con el SDK de OpenAI para lo básico sin streaming, y el catálogo de modelos para cada slug al que puedes apuntar este bucle.