AI Engineering Academy · Ders

Belirteç Akışını Anlama

Belirteç Akışını Anlama, CoddyKit'te ücretsiz bir AI Engineering Academy dersidir. Bu, 4 dersinin 1. 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.

Why Streaming Matters for User Experience

Without streaming, your application must wait for the LLM to generate the complete response before displaying anything — often 5-30 seconds for long answers. With streaming, the first token appears within 200-500ms of sending the request, and subsequent tokens stream in as they are generated. This transforms the perceived user experience from waiting to an engaging live generation effect, dramatically improving perceived responsiveness even though the total generation time is identical.

How LLMs Generate Tokens

LLMs are autoregressive: they generate text one token at a time, where each new token is conditioned on all previous tokens. When the API receives a request, the GPU starts sampling the first token immediately after the prompt is processed. Each subsequent token takes roughly the same time. Streaming sends each token to the client as soon as it is sampled, rather than buffering all tokens and sending the complete string at the end.

# Conceptual model of autoregressive generation
prompt = 'The capital of France is'

# Step 1: process full prompt, predict next token
# token_1 = sample(logits) → ' Paris'

# Step 2: append token_1 to context, predict next
# token_2 = sample(logits) → '.'

# Step 3: append token_2 to context, predict next
# token_3 = sample(logits) → '<|end|>'

# Total time: time_to_process_prompt + n_tokens * time_per_token
# With streaming: first token arrives after time_to_process_prompt (TTFT)
# Without streaming: everything arrives after TTFT + n_tokens * time_per_token

TTFT and TPOT: Two Latency Metrics

Streaming introduces two distinct latency concepts. TTFT (Time to First Token) is the delay from sending the request to receiving the first token — dominated by prompt processing time. TPOT (Time Per Output Token) is the time between consecutive tokens — determined by model size and hardware. TTFT affects how quickly the UI responds; TPOT affects how smoothly text streams. Both should be tracked separately in your observability stack.

import time
from openai import OpenAI

client = OpenAI()

def measure_streaming_latency(prompt: str):
    t_start = time.perf_counter()
    t_first_token = None
    token_times = []

    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
        stream=True,
    )
    for chunk in stream:
        if chunk.choices[0].delta.content:
            t_now = time.perf_counter()
            if t_first_token is None:
                t_first_token = t_now
                print(f'TTFT: {(t_first_token - t_start) * 1000:.0f}ms')
            else:
                token_times.append(t_now - token_times[-1] if token_times else t_now - t_first_token)
            token_times.append(t_now)
    print(f'TPOT avg: {1000 * (token_times[-1] - t_first_token) / max(len(token_times)-1, 1):.1f}ms')

The stream=True Parameter

Enabling streaming in the OpenAI SDK requires setting stream=True in the chat.completions.create call. The response type changes from a ChatCompletion object to a Stream[ChatCompletionChunk] iterator. Each chunk contains a delta with either a content string fragment or None when the token is a tool call or the stream is ending.

from openai import OpenAI

client = OpenAI()

# Non-streaming: wait for complete response
response = client.chat.completions.create(
    model='gpt-4o-mini',
    messages=[{'role': 'user', 'content': 'Explain RAG in one paragraph.'}],
)
full_text = response.choices[0].message.content

# Streaming: receive tokens incrementally
stream = client.chat.completions.create(
    model='gpt-4o-mini',
    messages=[{'role': 'user', 'content': 'Explain RAG in one paragraph.'}],
    stream=True,
)
for chunk in stream:
    delta = chunk.choices[0].delta.content
    if delta:  # delta can be None for non-content chunks
        print(delta, end='', flush=True)
print()  # newline at end

Accumulating the Full Response

In many application flows you need both to stream tokens to the UI for responsiveness and to accumulate the full response text for downstream processing such as logging, caching, or further pipeline steps. The pattern is simple: iterate over the stream, print or yield each chunk to the client, and simultaneously concatenate the content into a full string.

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

    full_text = ''
    finish_reason = None

    for chunk in stream:
        choice = chunk.choices[0]
        delta = choice.delta.content
        if delta:
            print(delta, end='', flush=True)  # real-time display
            full_text += delta               # accumulate
        if choice.finish_reason:
            finish_reason = choice.finish_reason

    print()  # newline
    print(f'Finished: {finish_reason}, total chars: {len(full_text)}')
    return full_text

Streaming with Usage Statistics

By default, the streaming response does not include token usage statistics (prompt tokens, completion tokens). To include them, pass stream_options={'include_usage': True}. The usage data arrives in a final chunk after the content stream ends. This is important for cost tracking and rate limit monitoring in production applications.

stream = client.chat.completions.create(
    model='gpt-4o-mini',
    messages=[{'role': 'user', 'content': 'What is a vector database?'}],
    stream=True,
    stream_options={'include_usage': True},  # include token counts
)

full_text = ''
usage = None

for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:
        full_text += chunk.choices[0].delta.content
    if chunk.usage:  # arrives in the final chunk
        usage = chunk.usage

if usage:
    print(f'Prompt tokens: {usage.prompt_tokens}')
    print(f'Completion tokens: {usage.completion_tokens}')
    print(f'Total tokens: {usage.total_tokens}')

When Not to Stream

Streaming is not always the right choice. Avoid streaming when: (1) you need the complete response before doing anything with it, such as JSON parsing or tool call detection; (2) the response is very short (under 30 tokens) where streaming overhead adds more delay than it saves; or (3) you are batch processing many requests where throughput matters more than individual response latency. In these cases, standard non-streaming calls are simpler and equally fast.

Streaming with Anthropic and Gemini APIs

Streaming is available on all major LLM provider APIs, not just OpenAI. The pattern is similar but the SDK interfaces differ slightly. Anthropic's Python SDK uses client.messages.stream() as a context manager, while Gemini uses generate_content(stream=True). When building provider-agnostic applications, abstract the streaming interface behind a common generator function.

import anthropic

ant_client = anthropic.Anthropic(api_key='YOUR_KEY')

# Anthropic streaming
with ant_client.messages.stream(
    model='claude-sonnet-4-5',
    max_tokens=1024,
    messages=[{'role': 'user', 'content': 'Explain hybrid search briefly.'}],
) as stream:
    for text in stream.text_stream:
        print(text, end='', flush=True)

# Final message with usage stats
final_msg = stream.get_final_message()
print(f'\nInput tokens: {final_msg.usage.input_tokens}')
print(f'Output tokens: {final_msg.usage.output_tokens}')

Generator-Based Streaming Interface

A clean architecture pattern wraps streaming in a Python generator function that yields token strings. This decouples the streaming logic from the consumption logic — callers can iterate over the generator, write to a file, forward to a WebSocket, or accumulate to a string without the streaming code knowing how its output is used. This is the foundation of most production streaming APIs.

from typing import Generator

def stream_completion(
    messages: list[dict],
    model: str = 'gpt-4o-mini',
    **kwargs,
) -> Generator[str, None, None]:
    stream = client.chat.completions.create(
        model=model,
        messages=messages,
        stream=True,
        **kwargs,
    )
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            yield delta

# Usage: pipe to stdout
for token in stream_completion([{'role': 'user', 'content': 'Hello!'}]):
    print(token, end='', flush=True)

# Usage: accumulate
full = ''.join(stream_completion([{'role': 'user', 'content': 'Hello!'}]))

Streaming in Terminal and CLI Applications

In terminal applications, streamed output looks identical to typing — each character appears immediately as it is generated. The key requirement is using flush=True in every print call. Without flushing, Python buffers output until a newline, which defeats the purpose of streaming. You can also use sys.stdout.write(token) followed by sys.stdout.flush() for more control over output formatting.

import sys

def stream_to_terminal(messages: list[dict]):
    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=messages,
        stream=True,
    )
    token_count = 0
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            sys.stdout.write(delta)  # no newline added
            sys.stdout.flush()       # MUST flush or output buffers
            token_count += 1
    print()  # final newline
    print(f'({token_count} tokens generated)')

Streaming and Error Recovery

Streaming complicates error handling because a failure may occur mid-stream after you have already sent some tokens to the client. The recommended pattern is to wrap the stream iteration in a try/except block and on error either send an error sentinel to the client or close the stream cleanly. Always implement a timeout on the overall stream to handle cases where the server starts streaming but then stops mid-generation.

import signal

def stream_with_timeout(messages, timeout_seconds=30):
    def timeout_handler(signum, frame):
        raise TimeoutError('LLM stream timed out')

    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(timeout_seconds)

    try:
        stream = client.chat.completions.create(
            model='gpt-4o-mini',
            messages=messages,
            stream=True,
        )
        for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                yield delta
    except TimeoutError:
        yield '\n[Response timed out]'
    except Exception as e:
        yield f'\n[Error: {str(e)}]'
    finally:
        signal.alarm(0)  # cancel timeout

Quick Check

Test your understanding of LLM token streaming from this lesson.

Lesson Recap

In this lesson you learned: streaming sends each generated token to the client as soon as it is sampled, dramatically improving perceived responsiveness, TTFT and TPOT are the two key latency metrics to track separately, and stream=True changes the OpenAI SDK response to a chunk iterator that you consume with a for loop. Wrap streams in generator functions for a clean, reusable interface. Next up we implement async streaming with the Python SDK.

Sıkça Sorulan Sorular

“Belirteç Akışını Anlama” dersi ücretsiz mi?

Evet — “Belirteç Akışını Anlama” 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.

“Belirteç Akışını Anlama” dersinde ne öğreneceğim?

Akış API'sinin üretildikçe kısmi tamamlamaları nasıl gönderdiğini, OpenAI stream=True parametresinin nasıl çalıştığını ve akışın kullanıcı deneyimini ne zaman iyileştirdiğini anlayın. 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 1. dersidir.

“Belirteç Akışını Anlama” 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

Belirteç Akışını Anlama
Python SDK ile Akışları Tüketme
Server-Sent Events ile FastAPI'da Akış
Akış Halindeki Yanıtlarda Araç Çağrılarını İşleme

← AI Engineering Academy Sayfasına Dön