0Pricing
AI Engineering Academy · Ders

Python SDK ile Akışları Tüketme

Akış halinde gelen tamamlamaları tüketmek için OpenAI eşzamansız istemcisini async for ile kullanın, tam yanıtı biriktirin ve kısmi çıktıyı kaybetmeden akış ortasındaki hataları işleyin.

Python SDK ile Akışları Tüketme, CoddyKit'te ücretsiz bir AI Engineering Academy dersidir. Bu, 4 dersinin 2. dersidir. Aşağıdan dersin tamamını ücretsiz okuyabilir, sonra tarayıcıda yerleşik kod editörü ve 7/24 yapay zeka koçu ile uygulamalı olarak pratik yapabilirsin. Bu, AI Engineering Academy öğrenme yolunun bir parçasıdır ve ilerlemeniz web ve CoddyKit uygulaması arasında senkronize olur. AI Engineering Academy kursu toplamda 4 dersten oluşur.

Bu dersin bazı bölümleri henüz çevrilmemiş olup İngilizce olarak gösterilmektedir.

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.

Sıkça Sorulan Sorular

“Python SDK ile Akışları Tüketme” dersi ücretsiz mi?

Evet — “Python SDK ile Akışları Tüketme” dersin tüm metni burada web'de ücretsiz olarak okunabilir. Etkileşimli olarak pratik yapmak (yerleşik kod editörü ve 7/24 yapay zeka koçu) ve AI Engineering Academy kursunun geri kalanını açmak için CoddyKit PRO'ya yükselt. AI Engineering Academy kursu toplamda 4 dersten oluşur.

“Python SDK ile Akışları Tüketme” dersinde ne öğreneceğim?

Akış halinde gelen tamamlamaları tüketmek için OpenAI eşzamansız istemcisini async for ile kullanın, tam yanıtı biriktirin ve kısmi çıktıyı kaybetmeden akış ortasındaki hataları işleyin. AI Engineering Academy ile uygulamalı kodu tarayıcıda doğrudan çalıştırarak pratik yaparsın ve 7/24 yapay zeka koçu dersi çalışırken sorularını yanıtlar.

AI Engineering Academy öğrenmeye başlamak için deneyim gerekli mi?

Önceden deneyim gerekmez. CoddyKit'te AI Engineering Academy, başlangıçtan ileri seviyeye kadar yapılandırıldığı için buradan başlayabilir veya başından başlayıp kendi hızında ilerleme yapabilirsin. Bu, 4 dersinin 2. dersidir.

“Python SDK ile Akışları Tüketme” dersi ne kadar sürer?

Çoğu CoddyKit dersi yaklaşık 5–10 dakika sürer. Her biri kısa ve etkileşimli olduğu için sabit ilerleme yaparsın ve web ile uygulama arasında tam olarak bıraktığın yerden devam edebilirsin.

Bu AI Engineering Academy dersinde kod yazıp çalıştırabilir miyim?

Evet. Her AI Engineering Academy dersi yerleşik bir kod editörü içerir, bu sayede tarayıcıda gerçek kod yazıp çalıştırabilir ve anlık yapay zeka geri bildirimi alırsın — yerel kurulum gerekli değildir.

Bu kursun tüm dersleri

  1. Belirteç Akışını Anlama
  2. Python SDK ile Akışları Tüketme
  3. Server-Sent Events ile FastAPI'da Akış
  4. Akış Halindeki Yanıtlarda Araç Çağrılarını İşleme
← AI Engineering Academy Sayfasına Dön