逐字回复怎么实现?大模型 Stream 流式输出在 LangChain 中的实践
调用大模型生成内容时,最常见的做法是等模型把完整答案算出来,再一次返回——少则几百毫秒,多则几十秒。这种'憋大招'的方式在聊天场景里很致命:用户盯着空白页面,心里早就犯嘀咕了。
流式输出(Streaming)就是来解决这个痛点的。它让模型边想边往外蹦字,前端再把这些字粘起来展示,像真人对话一样有'正在输入'的感觉。在 LangChain 里实现流式并不复杂,但有几个坑值得留意。下面我会带你把原理、代码和实际考虑走一遍。

为什么需要流式?不只是图快
'一次返回'的问题很直观:比如生成 1000 字的文章要 5 秒,这 5 秒里用户什么也看不见,容易误以为程序死了。如果要生成上万字的报告,前端可能因为一次性加载大量文本而卡顿,中途用户想停还没法停。
流式输出改变的不是总生成时间,而是'感知时间'。模型每生成一个 token(大致对应一两个字或词),就立刻推送给调用方,调用方再实时显示。这种'所见即所得'的方式对交互式应用尤其重要。
流式的底层:分段生成,实时推送
LLM 生成文本本身就是 token 级别的递进过程。模型在内部状态机的驱动下,一次生成一小段(比如'翠花是'),然后基于前文继续生成下一个 token('山村里的')。流式输出就是把这个内部过程暴露出来:
- 模型每算出一个 token 片段,就通过 HTTP 流式响应或 WebSocket 推送给客户端。
- 客户端拿到片段后立即处理——打印到控制台、更新 UI,或者存起来。
- 客户端最后把所有片段拼接成完整回答。
打个比方:外卖配送不是等 10 份餐全做好了再一块送,而是做一份送一份,你收到就能吃。虽然全部餐送到手的总时间可能差不多,但你饿肚子的焦虑感低多了。
快速上手:用 ChatOpenAI 做故事小助手
我们先用最直接的方式——model.stream()——实现一个能流式讲故事的脚本。示例用的模型是本地部署的 DeepSeek,换成其他支持流式的模型(如 GPT-4、Llama 3)也一样。
环境准备
pip install --upgrade langchain-openai
代码
from langchain_openai import ChatOpenAI
# 初始化模型,streaming=True 是必须的
model = ChatOpenAI(
model_name='deepseek-r1:7b',
base_url="http://127.0.0.1:11434/v1",
api_key="none",
temperature=0.7,
streaming=True
)
print("开始生成故事:")
for chunk in model.stream("讲一个关于山村女孩翠花的短故事,500 字以内,温暖治愈风格。"):
print(chunk.content, end="", flush=True)




