01 · Streaming#
默认情况下，chat() 要等模型生成完所有内容才返回。
对于长回答，这可能需要几十秒——用户体验很差。
Streaming 让模型边生成边推送 token，实现「打字机」效果， 首个 token 的延迟从几十秒降到不到 1 秒。
本章覆盖：基础流式输出、进度统计、中途取消、实用封装。
import time
import os
import litellm
from dotenv import load_dotenv
from utils.llm_client import stream_chat
load_dotenv()
True
1. 基础：非流式 vs 流式#
感受一下等待时间的差异。
from utils.llm_client import chat
prompt = "用200字介绍人工智能的发展历史。"
# 非流式：等全部生成完才返回
print("[非流式] 等待中...", end="", flush=True)
t0 = time.time()
response = chat(prompt)
t1 = time.time()
print(f" 完成（{t1-t0:.1f}s）")
print(response[:100], "...")
[非流式] 等待中... 完成（13.5s）
人工智能始于20世纪50年代，图灵提出机器思考设想，1956年达特茅斯会议奠基。60—70年代以符号主义与专家系统为主，80年代遭遇AI寒冬。90年代统计学习兴起，2000年代大数据与GPU推动深度学 ...
# 流式：token 边生成边显示
print("[流式] 实时输出：\n")
t0 = time.time()
first_token_time = None
for chunk in stream_chat(prompt):
if first_token_time is None:
first_token_time = time.time() - t0
print(chunk, end="", flush=True)
total_time = time.time() - t0
if first_token_time is not None:
print(f"\n\n首个 token 延迟: {first_token_time:.2f}s")
else:
print("\n\n首个 token 延迟: N/A (模型未返回流式内容块)")
print(f"总耗时: {total_time:.1f}s")
[流式] 实时输出：
人工智能起源于上世纪五十年代，1956年达特茅斯会议奠基。早期以符号主义和专家系统为主，六七十年代与八十年代出现热潮与衰退。九十年代统计学习和机器学习崛起，二十一世纪深度学习特别是2006年后复兴，2012年卷积神经网络在图像识别突破带动大规模应用。如今AI广泛进入工业、医疗、交通等领域，同时围绕数据、算法和伦理展开新一轮挑战与治理。包括隐私、安全、公平和可解释性问题，并要求跨学科合作和制度创新。
首个 token 延迟: 24.77s
总耗时: 26.5s
2. 统计 Token 速率#
流式输出时实时统计生成速度（tokens/秒）。
import tiktoken
enc = tiktoken.get_encoding("cl100k_base")
def stream_with_stats(prompt: str) -> dict:
"""流式输出并统计速率。"""
chunks = []
t_start = time.time()
t_first = None
print("输出：")
for chunk in stream_chat(prompt):
if t_first is None:
t_first = time.time()
chunks.append(chunk)
print(chunk, end="", flush=True)
t_end = time.time()
full_text = "".join(chunks)
token_count = len(enc.encode(full_text))
duration = t_end - t_first if t_first else 0
stats = {
"首 token 延迟": f"{t_first - t_start:.2f}s" if t_first else "N/A",
"生成 token 数": token_count,
"生成耗时": f"{duration:.1f}s",
"生成速率": f"{token_count / duration:.0f} tokens/s" if duration > 0 else "N/A",
}
return stats
stats = stream_with_stats("列出5种常见的设计模式，每种一句话说明。")
print("\n\n统计：")
for k, v in stats.items():
print(f" {k}: {v}")
输出：
- 单例模式：保证一个类只有一个实例并提供全局访问点。
- 工厂方法模式：定义创建对象的接口，让子类决定实例化哪一个类以将创建与使用解耦。
- 观察者模式：在对象间建立一对多依赖，当被观察者状态改变时自动通知并更新所有观察者。
- 策略模式：将可互换的算法封装为独立策略类，运行时选择不同策略以改变行为。
- 装饰器模式：通过组合/包装在不修改原对象的情况下动态为对象添加职责或功能。
统计：
首 token 延迟: 5.45s
生成 token 数: 201
生成耗时: 1.8s
生成速率: 112 tokens/s
3. 中途取消#
流式输出的一个优势：可以在生成过程中中途停止，节省 token 开销。
def stream_until(prompt: str, stop_keyword: str) -> str:
"""
流式输出，遇到 stop_keyword 时中止。
适用于：找到关键信息后不需要继续生成。
"""
collected = []
for chunk in stream_chat(prompt):
collected.append(chunk)
print(chunk, end="", flush=True)
full = "".join(collected)
if stop_keyword in full:
print(f"\n\n[遇到 '{stop_keyword}'，提前停止]", flush=True)
break
return "".join(collected)
result = stream_until(
"列举10种编程语言，每种一行，格式：序号. 语言名 - 主要用途",
stop_keyword="5.", # 看到第5个就停
)
1. Python - 通用编程、数据科学、机器学习与Web开发
2. Java - 企业级后端与Android应用开发
3. JavaScript - 浏览器端交互与前后端（Node.js）全栈开发
4. C - 系统编程与嵌入式开发
5.
[遇到 '5.'，提前停止]
4. 收集完整响应#
有时需要流式展示，但最后也要拿到完整字符串。
def stream_and_collect(prompt: str, **kwargs) -> str:
"""流式显示的同时，返回完整响应字符串。"""
chunks = []
for chunk in stream_chat(prompt, **kwargs):
print(chunk, end="", flush=True)
chunks.append(chunk)
print() # 换行
return "".join(chunks)
full_response = stream_and_collect("用三句话总结深度学习的核心思想。")
print(f"\n完整响应长度: {len(full_response)} 字符")
深度学习通过多层神经网络自动从原始数据中学习逐级抽象的特征表示，将复杂任务分解为一系列非线性变换。网络参数通过反向传播结合梯度下降等优化方法在大量数据上端到端训练，使得深模型具备强大的函数逼近和模式识别能力。正是这种表征学习能力推动了视觉、语音和自然语言处理等领域的突破，但也依赖海量数据与计算并面临泛化、可解释性等挑战。
完整响应长度: 161 字符
5. 流式 + 结构化输出#
流式输出与 JSON 结合：边输出边解析，当 JSON 完整时再处理。
def stream_json(prompt: str) -> dict:
"""
流式输出 JSON，完整接收后解析。
显示实时输出让用户感知进度，同时保证结构化数据完整性。
"""
response = litellm.completion(
model=os.getenv("LLM_MODEL"),
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
stream=True,
)
chunks = []
print("实时输出：")
for chunk in response:
delta = chunk.choices[0].delta.content
if delta:
print(delta, end="", flush=True)
chunks.append(delta)
print("\n")
import json
return json.loads("".join(chunks))
data = stream_json("""
分析「Python」这门编程语言，输出 JSON，包含：
name, created_year, creator, paradigms(列表), use_cases(列表), pros(列表), cons(列表)
""")
print("解析后的结构化数据：")
import json
print(json.dumps(data, ensure_ascii=False, indent=2))
实时输出：
{
"name": "Python",
"created_year": 1991,
"creator": "Guido van Rossum",
"paradigms": [
"面向对象",
"命令式/过程式",
"函数式",
"脚本式/解释型",
"反射式"
],
"use_cases": [
"Web 开发（后端）",
"数据科学与机器学习",
"科学计算与数值分析",
"自动化脚本与系统运维",
"教育与入门编程",
"快速原型与开发",
"测试与持续集成",
"网络编程与爬虫",
"桌面应用开发",
"嵌入式/物联网（如 Raspberry Pi）"
],
"pros": [
"语法简洁、可读性高，易于学习",
"丰富且成熟的标准库与第三方生态（如 NumPy、Pandas、TensorFlow、Django）",
"跨平台，开发与部署门槛低",
"适合快速原型和迭代开发",
"强大的社区支持与大量文档、教程",
"良好的扩展性，可与 C/C++ 等语言集成以提升性能"
],
"cons": [
"解释型语言，单线程性能和原始执行速度通常不及编译型语言",
"全局解释器锁（GIL）限制了多线程在 CPU 密集型任务上的并行效率",
"动态类型在大型项目中可能导致运行时错误，需要更多测试和类型检查",
"内存占用相对较高，不适合对资源高度受限的场景",
"移动端原生开发支持较弱，生态不如原生平台成熟",
"包管理与依赖环境在不同项目间可能出现冲突（需要使用虚拟环境）"
]
}
解析后的结构化数据：
{
"name": "Python",
"created_year": 1991,
"creator": "Guido van Rossum",
"paradigms": [
"面向对象",
"命令式/过程式",
"函数式",
"脚本式/解释型",
"反射式"
],
"use_cases": [
"Web 开发（后端）",
"数据科学与机器学习",
"科学计算与数值分析",
"自动化脚本与系统运维",
"教育与入门编程",
"快速原型与开发",
"测试与持续集成",
"网络编程与爬虫",
"桌面应用开发",
"嵌入式/物联网（如 Raspberry Pi）"
],
"pros": [
"语法简洁、可读性高，易于学习",
"丰富且成熟的标准库与第三方生态（如 NumPy、Pandas、TensorFlow、Django）",
"跨平台，开发与部署门槛低",
"适合快速原型和迭代开发",
"强大的社区支持与大量文档、教程",
"良好的扩展性，可与 C/C++ 等语言集成以提升性能"
],
"cons": [
"解释型语言，单线程性能和原始执行速度通常不及编译型语言",
"全局解释器锁（GIL）限制了多线程在 CPU 密集型任务上的并行效率",
"动态类型在大型项目中可能导致运行时错误，需要更多测试和类型检查",
"内存占用相对较高，不适合对资源高度受限的场景",
"移动端原生开发支持较弱，生态不如原生平台成熟",
"包管理与依赖环境在不同项目间可能出现冲突（需要使用虚拟环境）"
]
}
小结#
场景 |
方案 |
|---|---|
用户界面实时显示 |
|
需要完整响应字符串 |
|
找到关键信息后停止 |
|
流式 + 结构化 |
流式接收 → 完整后解析 JSON |
核心规律： 首 token 延迟 < 1s，用户体验远好于等待整个响应。 在生产应用中，所有面向用户的 LLM 调用都应该用 Streaming。
下一章 → Function Calling：给模型装上「手」，让它调用真实工具