Day 12：流式输出与打字机效果
欢迎来到第十二天！今天我们将学习如何实现流式输出（Streaming），让模型的回复像打字机一样逐字显示，而不是等待全部生成后一次性返回。流式输出不仅能显著提升用户体验（减少等待焦虑），还能在 Agent 交互中提供实时的反馈，让用户感知到模型正在“思考”或“行动”。
一、今日学习目标
二、详细实现步骤
步骤 1：理解流式输出的原理
默认情况下，当我们调用 client.chat.completions.create 时，API 会等待模型生成完整的回复，然后一次性返回。这种方式对于长回复会导致用户长时间等待，体验较差。
流式输出则是通过设置 stream=True，API 在模型生成过程中逐步返回一个个小的数据块（chunk）。每个 chunk 包含一小段文本（通常是几个 Token），客户端可以立即处理并显示，而无需等待全部完成。
流式输出的优势：
- 即时反馈：用户看到文字不断出现，感觉响应更快。
- 长文本友好：对于长文章生成，可以边生成边阅读。
- Agent 可视化：在多步推理或工具调用时，可以逐步显示模型的思考过程。
步骤 2：基础流式调用示例
新建 streaming_demo.py，编写一个简单的流式调用，逐字打印模型回复。
import os
from dotenv import load_dotenv
from openai import OpenAI
load_dotenv()
client = OpenAI(
api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com"
)
def stream_chat(prompt: str):
"""流式聊天，逐字打印回复"""
print("AI: ", end="", flush=True)
response = client.chat.completions.create(
model="deepseek-chat",
messages=[{"role": "user", "content": prompt}],
stream=True, # 开启流式
temperature=0.7,
max_tokens=200
)
for chunk in response:
# 每个 chunk 是一个 ChatCompletionChunk 对象
# 提取 delta 中的 content（可能为 None 或空字符串）
delta_content = chunk.choices[0].delta.content if chunk.choices else ""
if delta_content:
print(delta_content, end="", flush=True)
print() # 换行
# 测试
user_input = "请用一句话介绍人工智能。"
stream_chat(user_input)
运行脚本：
python streaming_demo.py
你会看到文字逐字出现，模拟了 ChatGPT 的回复效果。
步骤 3：深入理解 chunk 结构
流式响应返回的是一个迭代器，每个元素是一个 ChatCompletionChunk 对象。它的结构与普通响应类似，但 choices[0].delta 包含增量内容，而不是完整的 message。delta 可能包含：
- content：文本增量。
- role：第一个 chunk 中通常是 "assistant"。
- function_call 或 tool_calls（如果使用了工具，可能包含部分参数）。
- finish_reason：在最后一个 chunk 中可能为 "stop" 或 "length"。
我们可以打印每个 chunk 的原始内容来观察：
def stream_chat_debug(prompt: str):
response = client.chat.completions.create(
model="deepseek-chat",
messages=[{"role": "user", "content": prompt}],
stream=True,
max_tokens=50
)
for i, chunk in enumerate(response):
print(f"Chunk {i}: {chunk}")
if i > 5:
break
运行这个调试函数，你会看到类似如下的输出（简化）：
Chunk 0: ChatCompletionChunk(id='…', choices=[Choice(delta=ChoiceDelta(content='', function_call=None, role='assistant', tool_calls=None), finish_reason=None, index=0)])
Chunk 1: ChatCompletionChunk(id='…', choices=[Choice(delta=ChoiceDelta(content='人工智能', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0)])
Chunk 2: ChatCompletionChunk(id='…', choices=[Choice(delta=ChoiceDelta(content='是', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0)])
…
最后一个 chunk 的 finish_reason 为 'stop'
注意：
- 第一个 chunk 通常不包含 content，而是设置了 role。
- 后续 chunk 的 role 为 None，content 包含实际文本片段。
- finish_reason 在最后一个 chunk 中才有意义。
步骤 4：实现完整的打字机效果（带光标控制）
上面的简单实现已经能逐字打印，但为了更优雅，我们可以使用 sys.stdout.write 替代 print，并确保没有额外的换行。此外，可以模拟打字时的停顿（time.sleep）来增加效果（但实际项目中不应故意延迟，这里仅演示）。
import sys
import time
def typewriter_effect(text: str, delay: float = 0.02):
"""逐字打印文本，模拟打字机"""
for char in text:
sys.stdout.write(char)
sys.stdout.flush()
time.sleep(delay)
print() # 最后换行
def stream_chat_typewriter(prompt: str, delay: float = 0.01):
print("AI: ", end="", flush=True)
response = client.chat.completions.create(
model="deepseek-chat",
messages=[{"role": "user", "content": prompt}],
stream=True,
temperature=0.7,
max_tokens=200
)
for chunk in response:
delta_content = chunk.choices[0].delta.content if chunk.choices else ""
if delta_content:
typewriter_effect(delta_content, delay) # 对每个增量进一步逐字
print()
# 测试
stream_chat_typewriter("讲一个简短的笑话")
注意：在实际使用中，我们通常不需要在 typewriter_effect 中再加 time.sleep，因为网络传输本身就有间隔。但作为演示可以加上。
步骤 5：流式输出与 Function Calling（选做）
当模型使用 Function Calling 时，如果开启了流式，tool_calls 也会以流式方式返回。这意味着 delta 中可能包含 tool_calls 的片段，而不是完整的参数。处理起来更复杂，因为我们需要累积这些片段才能得到完整的函数名和参数。
一个简单的做法是：在流式过程中检测 delta.tool_calls，将其存储到缓冲区，结束后解析。但这超出了今天的重点，我们只需了解流式与工具调用可以结合，但在实践中，很多框架（如 LangChain）已经处理了这些细节。
为了简单，我们只在纯文本对话中使用流式输出。后续在 Agent 开发中，我们可能会使用 LangChain 的流式回调来显示 Agent 的推理过程。
三、常见问题与调试
Q1：流式输出时，为什么有时候最后一个字符丢失？
→ 确保在循环结束后打印一个换行符，或者检查 finish_reason 是否为 "length"（表示因达到 max_tokens 而截断）。如果输出被截断，可以增加 max_tokens。
Q2：如何处理流式输出中的异常？
→ 在循环中捕获异常，例如网络中断，可以尝试重试或提示用户。通常使用 try-except 包裹整个流式请求。
Q3：如何在 Flask 或 Web 应用中实现流式输出？
→ 可以使用 Server-Sent Events (SSE) 或 WebSocket 将流式 chunk 推送到前端。这超出了今天的范围，但原理相同：将每个 chunk 发送给客户端。
Q4：流式输出会增加 API 调用成本吗？
→ 不会，成本按 Token 计算，与是否流式无关。但流式响应可能增加网络开销和客户端处理复杂度。
Q5：为什么第一个 chunk 的 content 是空字符串？
→ 第一个 chunk 主要用于传递 role 信息（如 "assistant"），这是 API 的设计。你在处理时应忽略空字符串。
四、今日总结与作业
今天你完成了：
- ✅ 理解了流式输出的原理和优势。
- ✅ 使用 stream=True 实现了基本的逐字打印。
- ✅ 探索了 chunk 的结构，并编写了带打字机效果的完整脚本。
- ✅ 了解了流式输出与 Function Calling 结合的复杂性。
今日作业（必做）：
明日预告： 我们将学习系统提示词设计，深入掌握如何编写高质量的 System Prompt，为后续复杂 Agent 的角色设定和任务约束打下坚实基础。
有任何问题欢迎随时提问！
