0Pricing
AI Engineering Academy · Lección

Consumo de streams con el SDK de Python

Use el cliente asíncrono de OpenAI con async for para consumir completados en streaming, acumular la respuesta completa y gestionar errores durante la transmisión sin perder la salida parcial.

Consumo de streams con el SDK de Python es una lección gratuita de AI Engineering Academy en CoddyKit. Esta es la lección 2 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de AI Engineering Academy, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de AI Engineering Academy incluye 4 lecciones en total.

Partes de esta lección aún no han sido traducidas y se muestran en inglés.

Sync vs Async Streaming Clients

The OpenAI Python SDK provides both a synchronous OpenAI client and an asynchronous AsyncOpenAI client. For command-line scripts and simple applications, the synchronous client is easier to use. For web servers, APIs, and applications that handle multiple concurrent requests, the async client is essential — it does not block the event loop while waiting for tokens, allowing other requests to be served concurrently.

# Synchronous client (simple scripts)
from openai import OpenAI
client = OpenAI()

# Asynchronous client (web servers, concurrent workloads)
from openai import AsyncOpenAI
async_client = AsyncOpenAI()

# The async client has the same API surface as the sync client
# but all methods are coroutines that must be awaited

Async Streaming with AsyncOpenAI

With the AsyncOpenAI client, the streaming call becomes a coroutine. You use async for to iterate over chunks instead of a regular for loop. The event loop can schedule other coroutines between each chunk arrival, enabling your server to handle other requests while waiting for the next token from the LLM — this is the key advantage over synchronous streaming in a web context.

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI()

async def async_stream_completion(prompt: str) -> str:
    stream = await async_client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
        stream=True,
    )

    full_text = ''
    async for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            print(delta, end='', flush=True)
            full_text += delta
    print()
    return full_text

# Run the coroutine
asyncio.run(async_stream_completion('Explain what async/await does in Python'))

Using the Stream Context Manager

The OpenAI SDK also provides a stream context manager via client.chat.completions.stream(). This approach automatically closes the stream when the context exits and provides convenience methods like stream.text_stream that yield only non-None text deltas and stream.get_final_completion() for post-stream usage statistics without manually accumulating them.

from openai import AsyncOpenAI
import asyncio

async def stream_with_context_manager(prompt: str):
    async with async_client.chat.completions.stream(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
    ) as stream:
        # text_stream filters None deltas automatically
        async for text in stream.text_stream:
            print(text, end='', flush=True)

    # Access final completion after stream ends
    completion = await stream.get_final_completion()
    print(f'\nUsage: {completion.usage}')
    return completion

asyncio.run(stream_with_context_manager('What are the benefits of async I/O?'))

Handling Mid-Stream Errors Gracefully

Errors can occur at any point during a stream: during the initial connection, after the first token, or near the end of a long response. Wrap your stream iteration in try/except blocks and handle openai.APIConnectionError, openai.RateLimitError, and openai.APIStatusError separately, as each requires a different recovery strategy (retry, backoff, or user notification).

import openai

async def resilient_stream(prompt: str):
    try:
        stream = await async_client.chat.completions.create(
            model='gpt-4o-mini',
            messages=[{'role': 'user', 'content': prompt}],
            stream=True,
        )
        accumulated = ''
        async for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                accumulated += delta
                yield delta  # async generator

    except openai.RateLimitError:
        yield '[Rate limit reached — please wait and retry]'
    except openai.APIConnectionError:
        yield '[Connection error — check your network]'
    except openai.APIStatusError as e:
        yield f'[API error {e.status_code}]'
    except Exception as e:
        yield f'[Unexpected error: {type(e).__name__}]'

Async Generator for Streaming

The cleanest async pattern for streaming is an async generator function that yields tokens. Consumers iterate over it with async for. This keeps the streaming logic separate from how the output is used — a FastAPI endpoint, a WebSocket handler, and a test all consume the same generator without knowing about each other.

from typing import AsyncGenerator

async def token_stream(
    messages: list[dict],
    model: str = 'gpt-4o-mini',
) -> AsyncGenerator[str, None]:
    stream = await async_client.chat.completions.create(
        model=model,
        messages=messages,
        stream=True,
    )
    async for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            yield delta

# Consumer 1: print to terminal
async def print_stream(messages):
    async for token in token_stream(messages):
        print(token, end='', flush=True)

# Consumer 2: collect to string
async def collect_stream(messages) -> str:
    return ''.join([t async for t in token_stream(messages)])

Concurrent Streaming Requests

A major benefit of async streaming is the ability to run multiple streams concurrently within a single process. Using asyncio.gather, you can start several LLM streaming requests simultaneously and process their tokens as they arrive. This is useful for fan-out patterns where you want to compare multiple prompt variations or run parallel sub-tasks.

import asyncio

async def run_parallel_streams(queries: list[str]) -> list[str]:
    async def collect(query):
        messages = [{'role': 'user', 'content': query}]
        return ''.join([t async for t in token_stream(messages)])

    results = await asyncio.gather(*[collect(q) for q in queries])
    return results

queries = [
    'What is RAG?',
    'What is a vector database?',
    'What is BM25?',
]

async def main():
    answers = await run_parallel_streams(queries)
    for q, a in zip(queries, answers):
        print(f'Q: {q}\nA: {a[:100]}\n')

asyncio.run(main())

Timeout and Cancellation

Long-running streams should have timeouts to prevent indefinite blocking. Use asyncio.wait_for to apply a coroutine-level timeout or httpx.Timeout to set connection and read timeouts at the HTTP client level. Both approaches ensure that a stalled stream does not hold a request indefinitely. Always cancel streams explicitly when the user disconnects to avoid wasting GPU compute.

import asyncio

async def stream_with_timeout(messages, timeout_seconds: float = 30.0):
    try:
        async with asyncio.timeout(timeout_seconds):
            stream = await async_client.chat.completions.create(
                model='gpt-4o-mini',
                messages=messages,
                stream=True,
                timeout=timeout_seconds,  # HTTP-level timeout
            )
            async for chunk in stream:
                delta = chunk.choices[0].delta.content
                if delta:
                    yield delta
    except asyncio.TimeoutError:
        yield '\n[Stream timed out after {:.0f}s]'.format(timeout_seconds)

Buffering Partial Lines

When streaming to a client that processes complete lines (like a CLI that renders markdown), you may want to buffer tokens until a newline or sentence boundary before forwarding them. This avoids flickering renders of partial sentences. Accumulate tokens in a buffer, flush the buffer to the consumer when you detect a sentence-ending punctuation or a newline character, and always flush the remaining buffer at the end of the stream.

async def buffered_line_stream(messages):
    buffer = ''
    flush_on = {'.', '!', '?', '\n'}

    async for token in token_stream(messages):
        buffer += token
        if any(c in buffer for c in flush_on):
            # Find the last sentence-ending position
            for i, c in enumerate(reversed(buffer)):
                if c in flush_on:
                    split_pos = len(buffer) - i
                    yield buffer[:split_pos]
                    buffer = buffer[split_pos:]
                    break

    if buffer:  # flush remainder
        yield buffer

Recording Stream Latency in Production

In production, instrument every stream to record TTFT and total generation time for monitoring. Store these metrics in a time-series database and alert when TTFT exceeds your SLA threshold (typically 1-2 seconds for interactive applications). Correlate TTFT spikes with prompt length, model load, and time of day to identify root causes of latency degradation.

import time
from dataclasses import dataclass

@dataclass
class StreamMetrics:
    prompt_chars: int
    ttft_ms: float
    total_ms: float
    token_count: int

async def instrumented_stream(messages) -> tuple[str, StreamMetrics]:
    t_start = time.perf_counter()
    t_first = None
    token_count = 0
    full_text = ''

    stream = await async_client.chat.completions.create(
        model='gpt-4o-mini', messages=messages, stream=True
    )
    async for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            if t_first is None:
                t_first = time.perf_counter()
            token_count += 1
            full_text += delta

    t_end = time.perf_counter()
    prompt_len = sum(len(m.get('content', '')) for m in messages)
    metrics = StreamMetrics(
        prompt_chars=prompt_len,
        ttft_ms=(t_first - t_start) * 1000 if t_first else 0,
        total_ms=(t_end - t_start) * 1000,
        token_count=token_count,
    )
    return full_text, metrics

Testing Async Streaming Code

Testing async streaming requires special care. Use pytest-asyncio to run async test functions, and mock the OpenAI client to avoid real API calls in unit tests. Create a fake stream that yields predefined chunks with configurable delays to test both happy-path token processing and error-handling paths without spending API budget.

# pip install pytest pytest-asyncio
import pytest
from unittest.mock import AsyncMock, MagicMock

async def fake_stream(tokens: list[str]):
    for token in tokens:
        chunk = MagicMock()
        chunk.choices[0].delta.content = token
        yield chunk

@pytest.mark.asyncio
async def test_stream_accumulates_correctly(monkeypatch):
    mock_create = AsyncMock(return_value=fake_stream(['Hello', ', ', 'world', '!']))
    monkeypatch.setattr(async_client.chat.completions, 'create', mock_create)

    result = await collect_stream([{'role': 'user', 'content': 'Hi'}])
    assert result == 'Hello, world!'

SDK Helpers: stream.text and stream.final_message

The OpenAI Python SDK's stream context manager provides helper attributes that avoid manual accumulation. stream.text_stream is an async iterable that yields only non-None content strings. After the stream completes, await stream.get_final_message() returns a full ChatCompletionMessage with the complete text and usage data. These helpers reduce boilerplate and handle edge cases like empty deltas automatically.

async def clean_streaming_example(prompt: str):
    async with async_client.chat.completions.stream(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
    ) as stream:
        # Iterate only over text tokens, None deltas filtered automatically
        async for text in stream.text_stream:
            print(text, end='', flush=True)

    # After context exit, get accumulated result
    final = await stream.get_final_completion()
    return final.choices[0].message.content

Quick Check

Test your understanding of async streaming with the OpenAI Python SDK from this lesson.

Lesson Recap

In this lesson you learned: AsyncOpenAI enables non-blocking streaming that lets servers handle concurrent requests, async generators are the cleanest pattern for yielding streaming tokens to downstream consumers, and asyncio.wait_for and timeout parameters prevent indefinitely stalled streams from blocking your server. The stream context manager provides convenience helpers like text_stream and get_final_completion. Next up we expose LLM streaming to browser clients via FastAPI and Server-Sent Events.

Preguntas frecuentes

¿La lección «Consumo de streams con el SDK de Python» es gratis?

Sí — el texto completo de «Consumo de streams con el SDK de Python» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de AI Engineering Academy, actualiza a CoddyKit PRO. El curso de AI Engineering Academy incluye 4 lecciones en total.

¿Qué aprenderé en «Consumo de streams con el SDK de Python»?

Use el cliente asíncrono de OpenAI con async for para consumir completados en streaming, acumular la respuesta completa y gestionar errores durante la transmisión sin perder la salida parcial. Practicas AI Engineering Academy con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.

¿Necesito experiencia previa para empezar AI Engineering Academy?

No se requiere experiencia previa. AI Engineering Academy en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 2 de 4.

¿Cuánto tiempo toma la lección «Consumo de streams con el SDK de Python»?

La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.

¿Puedo escribir y ejecutar código en esta lección de AI Engineering Academy?

Sí. Cada lección de AI Engineering Academy incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.

Todas las lecciones de este curso

  1. Comprensión de la transmisión de tokens
  2. Consumo de streams con el SDK de Python
  3. Streaming en FastAPI con eventos enviados por el servidor
  4. Gestión de llamadas a herramientas en respuestas en streaming
← Volver a AI Engineering Academy