0Pricing
AI Engineering Academy · レッスン

Python SDKでストリームを利用する

OpenAIの非同期クライアントとasync forを使ってストリーミングされる完成結果を受け取り、完全なレスポンスを蓄積しながら、途中で発生したエラーにも部分出力を失わず対応します。

「Python SDKでストリームを利用する」はCoddyKit上の無料AI Engineering Academyレッスンです。 これはレッスン2/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Engineering Academy学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Engineering Academyコースには全4レッスンが含まれています。

このレッスンの一部はまだ翻訳されておらず、英語で表示されています。

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.

よくある質問

「Python SDKでストリームを利用する」レッスンは無料ですか？

はい。「Python SDKでストリームを利用する」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Engineering Academyコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Engineering Academyコースには全4レッスンが含まれています。

「Python SDKでストリームを利用する」で何を学びますか？

OpenAIの非同期クライアントとasync forを使ってストリーミングされる完成結果を受け取り、完全なレスポンスを蓄積しながら、途中で発生したエラーにも部分出力を失わず対応します。 ブラウザで直接実行するハンズオンコードでAI Engineering Academyを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

AI Engineering Academyを始めるのに経験は必要ですか？

事前経験は必要ありません。CoddyKitのAI Engineering Academyは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン2/4です。

「Python SDKでストリームを利用する」レッスンにはどのくらい時間がかかりますか？

ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。

このAI Engineering Academyレッスンでコードを書いて実行できますか？

はい。すべてのAI Engineering Academyレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。

このコースのすべてのレッスン

  1. トークンストリーミングを理解する
  2. Python SDKでストリームを利用する
  3. Server-Sent EventsによるFastAPIストリーミング
  4. ストリーミングレスポンスのツール呼び出しを処理する
← AI Engineering Academyに戻る