When you build AI applications (chatbots, copilots, data assistants), users expect a real-time, interactive experience. Without streaming, the entire response must be generated on the server before a single character reaches the user. This leads to:
- Long waiting times and blank screens.
- Poor perceived performance, even if the backend is fast.
- Inability to show “thinking” progress.
Streaming solves this by sending tokens as soon as they are generated, dramatically improving user experience. It’s a fundamental skill for any LLM engineer.
What is “Streaming” in AI/LLM?
In the context of Large Language Models (LLMs), streaming refers to the technique of returning generated tokens one by one (or in small chunks) as soon as they are produced by the model, rather than waiting for the entire response to be completed. The underlying transport is typically Server-Sent Events (SSE) or chunked HTTP responses, where the server pushes incremental updates to the client.
1. Set Client and message
# ============================================================
# 第一部分：环境设置 / Environment Setup
# ============================================================
from openai import OpenAI
import os
# ---------- 初始化客户端 ----------
# 这里以 DeepSeek 为例（兼容 OpenAI SDK）。
# 如果使用 OpenAI 官方，只需设置 api_key，base_url 可省略。
# 如果使用 Azure OpenAI，需设置 api_version 和 azure_endpoint。
client = OpenAI(
api_key=os.getenv("DEEPSEEK_API_KEY"), # 从环境变量读取密钥，安全实践
base_url="https://api.deepseek.com" # DeepSeek 的端点地址
)
# ---------- 准备对话消息 ----------
# 这里的 messages 是标准的 Chat Completion 格式。
# system prompt 用于设定角色，user message 是用户输入。
messages = [
{"role": "system", "content": "You are a helpful data engineering assistant."},
{"role": "user", "content": "Explain what a data lake is in 3 sentences."}
]2. Standard (Non-Streaming) Call
# ============================================================
# 第二部分：标准调用（非流式）/ Standard (Non-Streaming) Call
# ============================================================
response = client.chat.completions.create(
model="deepseek-chat", # 模型名称
messages=messages,
stream=False # 默认为 False，一次性返回完整结果
)
# 直接打印整个回答
print(response.choices[0].message.content)
# 这个调用会等待全部 token 生成完毕才返回，前端会出现空白等待。3. Streaming Call
# ============================================================
# 第三部分：流式调用 / Streaming Call
# ============================================================
# ---------- 发起流式请求 ----------
# 设置 stream=True，API 会返回一个可迭代的 chunk 流。
stream = client.chat.completions.create(
model="deepseek-chat",
messages=messages,
stream=True, # 🔑 关键参数：开启流式输出
temperature=0.7
)
# ---------- 处理 chunk 流 ----------
# 每个 chunk 代表一个增量更新，包含一个 token 或几个 token。
# 我们需要从 chunk 中提取 choices[0].delta.content 并打印。
# delta 是本次增量的意思，与标准响应中的 message 不同。
print("Assistant: ", end="", flush=True) # 先打印前缀，flush 立即输出
for chunk in stream:
# 从 chunk 中取出增量 delta
delta = chunk.choices[0].delta # 流式响应中，每个 chunk 的文本内容存放在哪里？
# Where is the text content inside a streaming chunk?
# delta.content 可能为 None（如只包含角色信息或结束标记）
if delta.content is not None:
# 逐 token 打印，end="" 防止换行，flush 立即刷新到控制台
print(delta.content, end="", flush=True)
print() # 最后换行4. Tool Call Streaming
当模型决定调用工具时，流式响应会包含 tool_calls 的增量信息。我们需要手动拼装完整的 function name 和 arguments（JSON 字符串）。
# ============================================================
# 第四部分：工具调用流式处理 / Tool Call Streaming
# ============================================================
# 定义一个简单的工具 schema（例如查询天气）
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get current weather for a city",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "City name"}
},
"required": ["city"]
}
}
}
]
messages = [
{"role": "system", "content": "You are a helpful assistant with access to tools."},
{"role": "user", "content": "What's the weather in San Francisco?"}
]
stream = client.chat.completions.create(
model="deepseek-chat",
messages=messages,
tools=tools,
stream=True
)
# ---------- 准备变量，用于增量拼装工具调用 ----------
# 因为工具调用数据是分块到达的，我们需要累积它们。
tool_call_accumulator = {} # 格式: {index: {"id": ..., "function_name": ..., "arguments": ""}}
print("Assistant thinking...", flush=True)
for chunk in stream:
delta = chunk.choices[0].delta
# 检查是否存在 tool_calls 增量
if delta.tool_calls:
for tool_call_delta in delta.tool_calls:
# tool_call_delta 包含: index (索引), id (可能只在第一个 chunk 出现), function (含 name 和 arguments 片段)
idx = tool_call_delta.index
# 如果该索引的工具调用尚未记录，初始化它
if idx not in tool_call_accumulator:
tool_call_accumulator[idx] = {
"id": tool_call_delta.id or "", # 工具调用的唯一ID
"function_name": "",
"arguments": ""
}
# 累积 id（通常只在第一个 chunk 出现）
if tool_call_delta.id:
tool_call_accumulator[idx]["id"] = tool_call_delta.id
# 累积 function name（通常只在第一个 function chunk 出现）
if tool_call_delta.function and tool_call_delta.function.name:
tool_call_accumulator[idx]["function_name"] = tool_call_delta.function.name
# 累积 arguments 片段（JSON 字符串的逐片追加）
if tool_call_delta.function and tool_call_delta.function.arguments:
tool_call_accumulator[idx]["arguments"] += tool_call_delta.function.arguments
# 打印拼装完成的工具调用
import json
print("\n--- Completed Tool Calls ---")
for idx, tc in tool_call_accumulator.items():
print(f"Call {idx}:")
print(f" ID: {tc['id']}")
print(f" Function: {tc['function_name']}")
print(f" Arguments: {tc['arguments']}")
# 可以进一步将 arguments 解析为 Python dict
try:
args_dict = json.loads(tc['arguments'])
print(f" Parsed Args: {args_dict}")
except json.JSONDecodeError:
print(" Arguments not yet complete or invalid JSON.")