理解令牌流式传输
了解流式 API 如何在生成过程中发送部分补全结果、OpenAI 的 stream=True 参数如何工作，以及流式传输何时能够改善用户体验。
理解令牌流式传输 是 CoddyKit 上的免费 AI Engineering Academy 课时。 这是第 1 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Engineering Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Engineering Academy 课程共包含 4 节课。
本课时的部分内容尚未翻译，以英文显示。
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.
常见问题解答
「理解令牌流式传输」课时是免费的吗？
是的 — 「理解令牌流式传输」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Engineering Academy 课程的其余内容，请升级到 CoddyKit PRO。 AI Engineering Academy 课程共包含 4 节课。
「理解令牌流式传输」这节课中我会学到什么？
了解流式 API 如何在生成过程中发送部分补全结果、OpenAI 的 stream=True 参数如何工作，以及流式传输何时能够改善用户体验。 你通过在浏览器中直接运行的动手代码来练习 AI Engineering Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Engineering Academy 需要有经验吗？
无需任何先前经验。CoddyKit 上的 AI Engineering Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 1 节课，共 4 节。
「理解令牌流式传输」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 AI Engineering Academy 课中编写并运行代码吗？
能。每节 AI Engineering Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。