在构建现代 AI 应用时，用户对响应速度和交互体验的要求日益提高。传统的“提交-等待-返回”模式已无法满足需求，流式输出（Streaming）成为提升用户体验的关键技术。LangChain 作为主流的 LLM 应用开发框架，提供了多层次的流式支持，允许开发者将大语言模型的生成过程以实时、渐进的方式呈现给用户。
流式输出的基础：LLM 配置与底层原理
实现流式输出的前提是底层 LLM 服务本身支持流式接口。LangChain 通过封装各服务商的 API，统一了流式调用的入口。以 OpenAI 为例，其 ChatCompletion API 提供了 stream=True 参数，服务器会以 SSE（Server-Sent Events）协议逐个返回 token。
在 LangChain 中，开发者需要在初始化 LLM 对象时显式启用流式模式：
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="gpt-4-turbo",
streaming=True, # 关键配置
temperature=0.7
)
若未设置 streaming=True，后续调用 .stream() 方法将抛出异常。这并非 LangChain 的限制，而是为了确保开发者明确知晓当前使用的 LLM 是否具备流式能力。
LangChain 的流式机制基于 Python 的生成器（Generator）。当调用 .stream() 时，它会返回一个可迭代对象，每次迭代产生一个 Chunk（通常是 AIMessageChunk）。这种设计使得流式处理可以无缝集成到任何支持迭代的上下文中，如 Web 框架的响应流或命令行界面。
基于 LCEL 的链式流式处理
LangChain Expression Language（LCEL）是官方推荐的声明式编程范式，它极大地简化了流式 Chain 的构建。一个典型的 LCEL 链由 PromptTemplate、LLM 和 OutputParser 组成。要使其支持流式，只需在 LLM 层启用 streaming=True，并在链末尾使用 StrOutputParser 将 AIMessage 转换为纯字符串。
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
prompt = ChatPromptTemplate.from_template("请详细解释 {concept} 的工作原理")
chain = prompt | llm | StrOutputParser()
for chunk in chain.stream({"concept": "Transformer"}):
print(chunk, end="", flush=True)
此代码会逐字打印出模型的回答。对于更复杂的场景，如 RAG（检索增强生成），LangChain 提供了 create_retrieval_chain 工具。该链返回一个包含 context 和 answer 的字典，流式处理时需检查 answer 字段的存在性，以避免打印中间检索结果。
Agent 与高级流式事件处理
Agent 的流式处理比普通 Chain 更为复杂，因为它涉及多步推理循环：思考（Thought）、行动（Action）、观察（Observation）。LangChain 提供了两种粒度的流式接口。
基础的 .stream() 方法仅返回最终答案的流，隐藏了内部的工具调用过程。这对于追求简洁用户体验的场景足够，但牺牲了透明度。为了展示完整的推理链，LangChain 引入了 astream_events（异步）和 stream_events（同步）方法。这些方法会发射一系列结构化事件，包括 on_chat_model_stream（模型输出流）、on_tool_start（工具调用开始）和 on_tool_end（工具调用结束）。
通过监听这些事件，开发者可以构建类似 ReAct（Reasoning + Action）框架的可视化界面，在前端实时展示 AI 的“思考过程”。例如，当 Agent 调用搜索引擎时，界面可以显示“正在查询…”，随后逐步展示搜索结果和最终结论。这种透明化的设计能显著增强用户对 AI 决策的信任感。
Web 应用集成与性能考量
将 LangChain 流式输出集成到 Web 应用是常见需求。以 FastAPI 为例，可以利用其 StreamingResponse 将生成器直接转换为 HTTP 流式响应：
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
app = FastAPI()
@app.get("/chat")
async def chat_stream(query: str):
async def event_generator():
for chunk in chain.stream({"input": query}):
yield chunk
return StreamingResponse(event_generator(), media_type="text/plain")
前端可通过 JavaScript 的 fetch().body.getReader() 或 EventSource 接收并渲染流式内容。
在生产环境中，还需考虑性能和成本。流式输出虽然改善了用户体验，但并未减少总的 Token 消耗。对于摘要型 Memory 或需要多次 LLM 调用的复杂 Agent，应仔细监控首 Token 延迟（TTFT）和总生成时间。此外，流式连接会占用服务器资源更长时间，需合理配置连接超时和并发限制。
