AI Engineering Academy · Lección

Comprensión de la transmisión de tokens

Comprensión de la transmisión de tokens es una lección gratuita de AI Engineering Academy en CoddyKit. Esta es la lección 1 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.

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.

Preguntas frecuentes

¿La lección «Comprensión de la transmisión de tokens» es gratis?

Sí — el texto completo de «Comprensión de la transmisión de tokens» 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 «Comprensión de la transmisión de tokens»?

Comprenda cómo la API de streaming envía completados parciales a medida que se generan, cómo funciona el parámetro stream=True de OpenAI y cuándo la transmisión mejora la experiencia de usuario. 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 1 de 4.

¿Cuánto tiempo toma la lección «Comprensión de la transmisión de tokens»?

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

Comprensión de la transmisión de tokens
Consumo de streams con el SDK de Python
Streaming en FastAPI con eventos enviados por el servidor
Gestión de llamadas a herramientas en respuestas en streaming

← Volver a AI Engineering Academy