Volver al blog

API y SDKGuía

Respuestas en streaming en Python

Imprime la respuesta del modelo a medida que se genera con el paquete oficial openai — ejemplos sync y async, salida de razonamiento, uso de tokens y manejo de errores.

Por CheapTokens TeamPublicado el 3 min de lectura

Código Python con un bucle for sobre chunks del stream y texto apareciendo token a token
En esta página
  1. Configuración
  2. Ejemplo mínimo
  3. Modelos de razonamiento
  4. Uso de tokens
  5. asyncio
  6. Detener antes y timeouts
  7. Errores
  8. Qué hay en el cable

Con streaming, la respuesta llega pieza a pieza mientras el modelo aún la está generando, así que tus usuarios ven el texto de inmediato en lugar de esperar la respuesta completa. La API usa el formato de streaming estándar de OpenAI (server-sent events), y el paquete oficial openai de Python se encarga de ello por ti.

Todos los ejemplos a continuación son completos: cópialos en un archivo y ejecútalos. Los ejemplos se verificaron con Python 3.10.12 y openai 3.31.0 contra un servidor de prueba local que reproduce el formato de streaming de la API (sintaxis y ejecución).

Configuración#

bash
pip install openai
export CHEAPTOKENS_API_KEY="paste-your-key-here"

Ejemplo mínimo#

python
import os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
)

stream = client.chat.completions.create(
    model="deepseek-ai/DeepSeek-V4-Flash-0731",
    messages=[{"role": "user", "content": "Write a haiku about latency"}],
    stream=True,
)
for chunk in stream:
    if not chunk.choices:  # service chunk without text, e.g. the final usage chunk
        continue
    text = chunk.choices[0].delta.content
    if text:
        print(text, end="", flush=True)
print()

Dos detalles importan:

  • Omite los chunks sin choices. Algunos chunks no llevan texto — por ejemplo el chunk final con el uso de tokens.
  • delta.content puede estar vacío. El primer chunk a menudo solo lleva el rol, y el último solo finish_reason. Imprime texto solo cuando esté ahí.

Modelos de razonamiento#

zai-org/GLM-5.3-Flash razona antes de responder. El razonamiento llega en un campo separado del delta, reasoning_content, y la respuesta en sí en content. El SDK conserva los campos desconocidos, así que puedes leerlo con getattr. Este es un ejemplo completo con su propio cliente y su propio stream:

python
import os
import sys
from openai import OpenAI

client = OpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
)

stream = client.chat.completions.create(
    model="zai-org/GLM-5.3-Flash",
    messages=[{"role": "user", "content": "Is 1013 a prime number?"}],
    stream=True,
)
for chunk in stream:
    if not chunk.choices:
        continue
    delta = chunk.choices[0].delta
    thinking = getattr(delta, "reasoning_content", None)
    if thinking:
        print(thinking, end="", flush=True, file=sys.stderr)  # reasoning → stderr
    if delta.content:
        print(delta.content, end="", flush=True)              # answer → stdout
print()

El razonamiento puede tardar un rato antes de que aparezca la primera palabra de la respuesta — otra razón para usar streaming.

Uso de tokens#

Pide el uso con stream_options; llega en el último chunk, que no tiene choices:

python
import os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
)

stream = client.chat.completions.create(
    model="deepseek-ai/DeepSeek-V4-Flash-0731",
    messages=[{"role": "user", "content": "Explain idempotency in one paragraph"}],
    stream=True,
    stream_options={"include_usage": True},
)
usage = None
for chunk in stream:
    if chunk.usage:
        usage = chunk.usage
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

if usage:
    print(f"\ninput: {usage.prompt_tokens}, output: {usage.completion_tokens}")

Multiplica estos números por las tarifas del modelo para obtener el costo de la solicitud — consulta Cómo estimar el costo de un agente de IA.

asyncio#

En código async usa AsyncOpenAI y async for. El cliente tiene su propio nombre, async_client, para que no choque con el síncrono client de los otros ejemplos:

python
import asyncio
import os
from openai import AsyncOpenAI

async_client = AsyncOpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
)

async def main() -> None:
    stream = await async_client.chat.completions.create(
        model="deepseek-ai/DeepSeek-V4-Flash-0731",
        messages=[{"role": "user", "content": "Name three uses of a message queue"}],
        stream=True,
    )
    async for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)
    print()

asyncio.run(main())

Detener antes y timeouts#

Sal del bucle para detener la generación. Abrir el stream con with cierra la conexión cuando termina el bloque. El timeout del cliente se establece explícitamente; el SDK también reintenta algunas solicitudes fallidas por su cuenta (max_retries, 2 por defecto).

python
import os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
    timeout=120,  # seconds; long answers take time
)

received = ""
with client.chat.completions.create(
    model="deepseek-ai/DeepSeek-V4-Flash-0731",
    messages=[{"role": "user", "content": "Write a long story about a lighthouse"}],
    stream=True,
) as stream:
    for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            received += chunk.choices[0].delta.content
        if len(received) > 300:  # stop early: enough text received
            break
print(received)

Errores#

Los errores que ocurren antes del primer chunk — clave incorrecta, saldo vacío, modelo desconocido — se lanzan como excepciones cuando creas el stream:

python
import os
import openai
from openai import OpenAI

client = OpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
)

try:
    stream = client.chat.completions.create(
        model="deepseek-ai/DeepSeek-V4-Flash-0731",
        messages=[{"role": "user", "content": "Hello"}],
        stream=True,
    )
    for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)
    print()
except openai.AuthenticationError:          # 401: wrong or rotated key
    print("Check the API key")
except openai.BadRequestError as e:         # 400: e.g. unknown model ID
    print("Bad request:", e.message)
except openai.APIStatusError as e:          # other statuses, e.g. 402 when the balance is used up
    print(e.status_code, e.message)

Una lista completa de códigos de estado y soluciones está en Errores de conexión comunes y cómo solucionarlos.

Qué hay en el cable#

Para depurar, mira el stream en bruto con curl (-N desactiva el buffering):

bash
curl -N https://api.cheaptokens.dev/v1/chat/completions \
  -H "Authorization: Bearer $CHEAPTOKENS_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"model": "deepseek-ai/DeepSeek-V4-Flash-0731", "stream": true,
       "messages": [{"role": "user", "content": "Hi"}]}'

Cada evento es una línea data: {…} con un chunk JSON; el stream termina con data: [DONE]. Las líneas que empiezan con dos puntos son comentarios keep-alive enviados durante pausas largas — los clientes los ignoran.

Etiquetas: