Understanding Token Streaming
Understand how the streaming API sends partial completions as they are generated, how the OpenAI stream=True parameter works, and when streaming improves user experience.
Understanding Token Streaming is a free AI Engineering Academy lesson on CoddyKit — lesson 1 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the AI Engineering Academy learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.
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_tokenTTFT 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 endAccumulating 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_textStreaming 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 timeoutQuick 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.
Frequently asked questions
Is the “Understanding Token Streaming” lesson free?
Yes — the full text of “Understanding Token Streaming” is free to read here on the web, and the AI Engineering Academy course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the AI Engineering Academy course, upgrade to CoddyKit PRO.
What will I learn in “Understanding Token Streaming”?
Understand how the streaming API sends partial completions as they are generated, how the OpenAI stream=True parameter works, and when streaming improves user experience. You practise AI Engineering Academy with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.
Do I need any experience to start AI Engineering Academy?
No prior experience is required. AI Engineering Academy on CoddyKit is structured for beginners through advanced learners; this is — lesson 1 of 4, so you can start here or from the beginning and move at your own pace.
How long does the “Understanding Token Streaming” lesson take?
Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.
Can I write and run code in this AI Engineering Academy lesson?
Yes. Every AI Engineering Academy lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.