AI 大模型 Stream 流式输出实战
介绍 AI 大模型 Stream 流式输出的核心原理及实战落地。通过 ChatOpenAI 直接调用和 LangChain LCEL 表达式两种方案,实现故事生成与科普助手的流式响应。分析了流式输出在用户体验、内存占用及中断支持方面的优势,同时指出其无法减少总耗时、增加代码复杂度等限制。提供了 API 密钥安全、流式中断处理及前端展示等避坑指南,帮助开发者在聊天机器人或长文本场景下优化交互体验。

介绍 AI 大模型 Stream 流式输出的核心原理及实战落地。通过 ChatOpenAI 直接调用和 LangChain LCEL 表达式两种方案,实现故事生成与科普助手的流式响应。分析了流式输出在用户体验、内存占用及中断支持方面的优势,同时指出其无法减少总耗时、增加代码复杂度等限制。提供了 API 密钥安全、流式中断处理及前端展示等避坑指南,帮助开发者在聊天机器人或长文本场景下优化交互体验。

在之前的 LangChain 实战中,我们调用大模型时都是「一次性获取完整结果」——比如生成一篇文案、回答一个问题,需要等模型把所有内容生成完才返回。但在实际场景中(如 ChatGPT 聊天界面、长文本生成),这种方式会让用户面对'空白加载页'等待几秒甚至更久,体验大打折扣。
本文将聚焦 LLM 的Stream 流式输出,从核心原理讲起,通过'故事小助手''科普助手'两个实战案例,带你掌握从基础调用到 LCEL 表达式的流式落地,最后分析流式输出的优劣势与实战注意事项。

在学习流式输出前,我们先明确:流式输出不是'让模型生成更快',而是'让用户感知更快'。
先看'一次性输出'的典型问题:
而流式输出的解决思路很简单:模型生成一个字/一个短句,就立刻返回一个'片段(Chunk)',前端实时拼接展示——就像与人对话时'边说边听',而非'等对方说完一长段再回应'。
LLM 的流式输出本质是基于HTTP 流式传输(或 WebSocket)实现的'增量返回'机制,核心逻辑可拆解为 3 步:
类比生活场景:就像外卖小哥送 10 份餐,'一次性输出'是等 10 份餐全做好再一起送;'流式输出'是做好 1 份送 1 份,你收到后可以先吃,不用等全部到齐。
首先从最基础的「直接调用大模型流式接口」入手,实现一个'输入关键词,流式生成故事'的小工具。我们依然用DeepSeek 作为示例模型,你也可以替换为 GPT-4、Llama 3 等支持流式的模型。
确保已安装最新版 langchain-openai(流式功能依赖较新的 API 封装):
pip install --upgrade langchain-openai
# 1. 导入依赖:ChatOpenAI 是大模型客户端,支持流式调用
from langchain_openai import ChatOpenAI
# 2. 初始化大模型:关键是确保模型支持流式(qwen-plus、gpt-4 等均支持)
model = ChatOpenAI(
model_name='deepseek-r1:7b', # 本地模型名称,根据实际情况填写
base_url="http://127.0.0.1:11434/v1", # 本地模型 API 地址
api_key="none", # 本地模型通常不需要真实 API 密钥
temperature=0.7, # 可根据需要调整温度参数
streaming=True # 开启流式模式(核心参数,必须设为 True)
)
# 3. 流式调用:用 for 循环迭代接收'片段(Chunk)'
print("=== 故事小助手(流式输出)===")
print("正在生成《翠花的山村故事》...\n")
# model.stream() 返回的是'片段迭代器',每次循环获取一个生成片段
for chunk in model.stream("讲一个关于山村女孩翠花的短故事,500 字以内,温暖治愈风格。"):
# chunk.content 是当前片段的文本内容(如'翠花''住在''太行山脚下')
#:取消默认的'换行',让片段连续拼接
# flush=True:强制实时刷新输出(避免片段在缓冲区堆积,导致'一次性打印')
print(chunk.content, end="", flush=True)
代码执行后,你会看到控制台逐字/逐词输出故事,而非等待几秒后一次性显示:
=== 故事小助手(流式输出)=== 正在生成《翠花的山村故事》... ## 《奶奶的病床》我端着水碗站在厨房里,看着奶奶平日里气色红润的脸庞此刻变得苍白。她坐在角落的藤椅上,手里紧紧攥着那张泛黄的药方。'小翠花来。'奶奶颤巍巍的声音像是从很远的地方传来。我小心翼翼地走进房间,脚步声轻得像一片落叶。将药方放在桌上时,我不由得放慢了手速。红肿的眼睛里,我能看到奶奶年轻时在镜子里的倒影。'这是医生说的药方。'我轻声说,'我先给你倒水吧。'茶几上的杯子空空如也,奶奶指节分明的手指轻轻叩击着杯沿,似乎这样就能让药效更慢一些。我知道她又要犯那个习惯——等我看书看累了才来。'小翠花,你多大了?'她的声音里带着不容置疑的关心。我打量着手里的水杯,又看着这间熟悉的小屋。竹制的窗外已经有些开裂,但奶奶还是坚持每天午休时坐在那里晒太阳。那时的她总是笑着的,像是有说不尽的话题。'十岁。'我回答得很快,'写作业的时候写累了就来吧。''不用了不用了...'奶奶摇摇头,水声中带着淡淡的苦涩,'你先去写作业吧,别站着太久了。'

关键注意点:
streaming=True是开启流式的核心参数,若未设置,model.stream()会报错;flush=True必须加,否则 Python 会将片段缓存起来,导致'看似流式实则一次性输出'。
在 LangChain 0.3+ 中,推荐用LCEL 表达式(| 管道符)串联组件——流式输出也不例外。相比直接调用 model.stream(),LCEL 能更灵活地结合'提示模板、输出解析器',适配复杂场景(如带格式的科普内容生成)。
实现功能:接收用户输入的'科普主题',按'【核心概念】→【生活案例】→【一句话总结】'的格式,流式生成科普内容。
# 1. 导入依赖:LCEL+ 流式所需组件
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate # 提示模板(支持多轮格式)
from langchain_core.output_parsers import StrOutputParser # 流式输出解析器
from langchain_core.runnables import RunnablePassthrough # 辅助传递变量
# 2. 初始化核心组件
# 2.1 大模型(开启流式)
model = ChatOpenAI(
model_name='deepseek-r1:7b', # 本地模型名称,根据实际情况填写
base_url="http://127.0.0.1:11434/v1", # 本地模型 API 地址
api_key="none", # 本地模型通常不需要真实 API 密钥
temperature=0.7, # 可根据需要调整温度参数
streaming=True # 开启流式模式(核心参数,必须设为 True)
)
# 2.2 提示模板(定义科普内容的格式)
prompt = ChatPromptTemplate.from_messages([
("system","你是专业科普助手,按以下格式生成内容:\n1. 【核心概念】:用 30 字内解释主题;\n2. 【生活案例】:举 1 个贴近生活的例子;\n3. 【一句话总结】:提炼核心价值。"),
("user","科普主题:{topic}") # 动态变量:用户输入的科普主题
])
# 2.3 输出解析器(将流式片段转为纯文本,避免多余格式)
parser = StrOutputParser()
# 3. 用 LCEL 管道符串联组件:prompt → 变量传递 → model → parser
# RunnablePassthrough():确保'topic'变量能传递到 prompt 中
lcel_stream_chain = ({"topic": RunnablePassthrough()} # 传递用户输入的'topic'
| prompt # 填充提示模板
| model # 流式调用模型
| parser # 解析流式片段)
# 4. 流式调用科普助手
print("=== 科普助手(LCEL 流式输出)===")
user_topic = input("请输入科普主题:")
()
chunk lcel_stream_chain.stream(user_topic):
(chunk, end=, flush=)
=== 科普助手(LCEL 流式输出)=== 请输入科普主题:人工智能 正在生成科普内容... 1. 【核心概念】:人工智能(AI)是模仿人类智能的系统技术。2. 【生活案例】:自动驾驶汽车通过 AI 识别道路信号、规划路线,帮助司机做出复杂决策。3. 【一句话总结】:人工智能的核心价值在于模拟人类智能,解决复杂问题并推动科技发展。

相比直接调用 model.stream(),LCEL 的核心优势在于组件解耦与复用:
prompt,无需动模型和解析器;model 实例,流式逻辑不变;输入验证 → prompt → model → parser → 数据库存储),流式特性不受影响。流式输出虽能提升体验,但并非适用于所有场景。我们需要客观看待其优劣势,避免盲目使用。
| 优势点 | 具体场景说明 |
|---|---|
| 提升用户感知体验 | 聊天机器人、实时问答场景中,用户看到'逐字输出',会觉得'响应很快',减少等待焦虑; |
| 降低内存占用 | 生成万字报告、小说章节时,无需一次性加载完整文本(可能占几十 MB 内存),而是逐段处理; |
| 支持中途中断 | 若用户看到部分内容后不想继续(如生成的故事不符合预期),可随时停止流式接收,节省资源; |
| 适配长文本生成 | 一次性输出长文本可能触发'超时错误'(如 API 限制单次返回时长),流式分段返回可规避此问题; |
| 限制点 | 实际影响与应对方案 |
|---|---|
| 完整结果总耗时未减少 | 流式只是'分段返回',模型生成完整内容的总时间基本不变(甚至略增,因多了分段传输开销); → 应对:仅在'用户需要实时感知'的场景用(如聊天),后台批量生成(如报表)仍用一次性输出; |
| 代码复杂度提升 | 需处理'片段拼接''中断逻辑''错误重试'(如某片段传输失败,需重新请求后续内容); → 应对:用 LangChain LCEL 封装,减少重复代码; |
| 部分模型/API 不支持 | 并非所有 LLM 都支持流式输出(如部分轻量开源模型、旧版 API); → 应对:调用前查看模型文档(如通义千问、GPT 系列、Llama 3 均明确支持流式); |
| 输出解析难度增加 | 一次性输出可直接用 JSON 解析器提取结构化数据(如'{'核心概念': 'xxx', '案例': 'xxx'}'),流式需逐段判断'是否解析完整'; → 应对:用 LangChain 的 StreamingOutputParser(专为流式设计的解析器); |
api_key 不要硬编码到项目中,建议用环境变量(如 os.getenv("QWEN_API_KEY"))或配置文件存储,避免泄露;stream.close() 停止流式传输,避免模型继续生成浪费资源;stream() 返回的是'包含 content 字段的 Chunk 对象';transformers 库的 pipeline("text-generation", streaming=True) 实现流式,接口略有不同;EventSource 接收并实时更新 DOM;ChatOpenAI.stream(),复杂场景用 LCEL 串联'模板 + 模型 + 解析器',减少代码复杂度;
微信公众号「极客日志」,在微信中扫描左侧二维码关注。展示文案:极客日志 zeeklog
使用加密算法(如AES、TripleDES、Rabbit或RC4)加密和解密文本明文。 在线工具,加密/解密文本在线工具,online
生成新的随机RSA私钥和公钥pem证书。 在线工具,RSA密钥对生成器在线工具,online
基于 Mermaid.js 实时预览流程图、时序图等图表,支持源码编辑与即时渲染。 在线工具,Mermaid 预览与可视化编辑在线工具,online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online