0Pricing
AI Engineering Academy · Leçon

Comprendre la diffusion des jetons

Comprenez comment l’API de diffusion envoie les complétions partielles au fur et à mesure de leur génération, comment fonctionne le paramètre stream=True d’OpenAI et dans quels cas la diffusion améliore l’expérience utilisateur.

Comprendre la diffusion des jetons est une leçon AI Engineering Academy gratuite sur CoddyKit. Ceci est la leçon 1 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage AI Engineering Academy, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours AI Engineering Academy comprend 4 leçons au total.

Certaines parties de cette leçon n'ont pas encore été traduites et s'affichent en anglais.

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.

Questions Fréquemment Posées

La leçon « Comprendre la diffusion des jetons » est-elle gratuite ?

Oui — le texte complet de « Comprendre la diffusion des jetons » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours AI Engineering Academy, passe à CoddyKit PRO. Le cours AI Engineering Academy comprend 4 leçons au total.

Qu'est-ce que j'apprendrai dans « Comprendre la diffusion des jetons » ?

Comprenez comment l’API de diffusion envoie les complétions partielles au fur et à mesure de leur génération, comment fonctionne le paramètre stream=True d’OpenAI et dans quels cas la diffusion améli… Tu pratiques AI Engineering Academy avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.

Dois-je avoir de l'expérience pour commencer AI Engineering Academy ?

Aucune expérience préalable n'est requise. AI Engineering Academy sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 1 sur 4.

Combien de temps prend la leçon « Comprendre la diffusion des jetons » ?

La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.

Peux-tu écrire et exécuter du code dans cette leçon AI Engineering Academy ?

Oui. Chaque leçon AI Engineering Academy inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.

Toutes les leçons de ce cours

  1. Comprendre la diffusion des jetons
  2. Consommer des flux avec le SDK Python
  3. Diffusion dans FastAPI avec les événements envoyés par le serveur
  4. Gérer les appels d’outils dans les réponses diffusées
← Retour à AI Engineering Academy