1. 从“卡顿等待”到“丝滑对话”为什么流式输出是LLM应用的刚需如果你用过早期的聊天机器人或者尝试过一些本地部署的大语言模型一定对那种体验记忆犹新你问了一个问题然后屏幕就卡在那里光标闪烁你只能干等着不知道它是在思考还是已经卡死。过了十几秒甚至更久一大段文字“唰”地一下全部蹦出来。这种体验在技术术语里叫做“阻塞式响应”Blocking Response它极大地损害了交互的实时性和自然感。而流式输出Streaming Output彻底改变了这一点。它的核心思想很简单模型不是一次性生成全部内容再返回而是像打开水龙头一样每生成一个词或一个片段就立刻“流”给前端。对于用户来说看到文字一个接一个地“打”出来就像有人在实时打字一样交互感、沉浸感和响应速度都得到了质的提升。这不仅仅是体验优化对于处理长文本、复杂推理或需要中途调整的场景流式输出几乎是必备能力。在 LangGraph 的语境下流式输出又有了更深一层的意义。LangGraph 本身是一个用于构建有状态、多步骤智能体Agent工作流的框架。一个工作流可能包含调用工具、条件判断、循环等复杂逻辑。如果这样一个工作流的最终结果也要等全部执行完毕才返回那等待时间将是不可接受的。因此LangGraph 的流式输出不仅要流式返回最终的文本答案更要能流式返回工作流执行的“中间状态”——即每一步发生了什么哪个节点在执行返回了什么结果。这让开发者能够实时监控智能体的“思考过程”进行调试也让最终用户能理解AI的决策路径。从网络热词中频繁出现的stream disconnected before completion等错误可以看出在实际对接流式接口时网络稳定性、前后端协议、错误处理都是需要精心设计的坑点。而像yudao-cloud项目中flux流式输出与spring security的权限控制问题解析这样的讨论则揭示了在成熟的企业级框架中集成流式能力时会与认证授权等基础设施产生怎样的碰撞。本文将聚焦于 LangGraph深入探讨其stream方法的两种核心模式并结合实战拆解如何构建一个稳定、可观测的流式智能体应用。2. LangGraph 流式输出的两种核心模式与底层机制LangGraph 的CompiledStateGraph对象提供了.stream()方法来启动工作流的流式执行。这个方法返回的是一个异步迭代器Async Iterator这是实现流式的技术基础。理解其两种模式是正确使用它的前提。2.1 模式一流式更新Streaming Updates这是最常用、信息最丰富的模式。它流式返回的是工作流执行过程中每个节点Node执行完成后产生的更新Update。这里的“更新”指的是对 LangGraph 核心概念——State的修改。什么是 State你可以把它理解为一个共享的、贯穿整个工作流的字典Dict。每个节点都可以读取和修改这个字典里的内容。比如一个智能体工作流的 State 里可能包含这些键input用户输入、messages对话历史、tool_calls工具调用结果、response最终回复等。当你调用graph.stream(input, stream_modeupdates)时迭代器会产生一系列值。每个值都是一个元组(node_name, update_dict)。node_name: 当前执行完毕的节点名称。update_dict: 该节点对 State 所做的修改。注意这不是完整的 State而是本次操作的增量更新。让我们看一个极其简单的例子假设我们有一个包含两个节点的工作流from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated import operator # 1. 定义状态结构 class State(TypedDict): value: Annotated[int, operator.add] # 使用注解声明此字段为可累加 history: Annotated[list, operator.add] # 历史记录列表也是累加 # 2. 定义节点函数 def node_a(state: State): # 修改state new_value state.get(value, 0) 10 new_history fNode A added 10. return {value: new_value, history: [new_history]} def node_b(state: State): current_value state.get(value, 0) new_value current_value * 2 new_history fNode B doubled value to {new_value}. return {value: new_value, history: [new_history]} # 3. 构建图 builder StateGraph(State) builder.add_node(node_a, node_a) builder.add_node(node_b, node_b) builder.set_entry_point(node_a) builder.add_edge(node_a, node_b) builder.add_edge(node_b, END) graph builder.compile() # 4. 流式执行Updates模式 inputs {value: 5, history: []} print(Streaming Updates:) async for node_name, update in graph.astream(inputs, stream_modeupdates): print(f Node: {node_name}, Update: {update})可能的输出会是Streaming Updates: Node: node_a, Update: {value: 15, history: [Node A added 10.]} Node: node_b, Update: {value: 30, history: [Node B doubled value to 30.]}关键解读我们输入{value: 5}。node_a执行它读取到value5加上10得到15。它返回{value: 15, ...}。这个返回的字典就是update。LangGraph 会根据 State 定义如operator.add将这个更新合并到全局 State 中。流式接口立刻抛出了这个更新(node_a, {value: 15, ...})。node_b执行此时它读取到的value已经是更新后的15。它将其翻倍得到30并返回更新。流式接口再次抛出(node_b, {value: 30, ...})。这种模式的巨大优势在于可观测性。前端不仅可以实时显示最终答案的生成过程如果response字段是逐步追加的还可以展示智能体的“思考链”比如“调用了搜索工具 - 获得了结果 - 开始总结”。这对于调试复杂工作流和提升用户信任度至关重要。2.2 模式二流式数值Streaming Values这种模式相对直接。它流式返回的是每个节点执行完成后当前整个 State 的快照Snapshot。调用方式为graph.stream(input, stream_modevalues)。沿用上面的例子print(\nStreaming Values:) async for snapshot in graph.astream(inputs, stream_modevalues): print(f Full State Snapshot: {snapshot})可能的输出Streaming Values: Full State Snapshot: {value: 5, history: []} # 初始状态注意这里 Full State Snapshot: {value: 15, history: [Node A added 10.]} Full State Snapshot: {value: 30, history: [Node A added 10., Node B doubled value to 30.]}这里有一个非常重要的细节你可能注意到第一个snapshot是初始输入{value: 5, ...}。在values模式下迭代器会首先 yield 出初始的输入状态然后每次节点更新后再 yield 出合并更新后的完整状态。两种模式如何选择选updates当你需要精确知道“每一步发生了什么变化”用于前端动画、调试日志、构建复杂的交互式UI如高亮显示刚生成的部分。信息粒度更细。选values当你只关心“当前最新的完整状态是什么”用于简单的前端显示或者下游系统需要基于最新完整状态做决策。获取的数据更完整但可能包含冗余信息每次都是全量。底层机制浅析LangGraph 的流式能力建立在异步编程之上。工作流引擎将每个节点的执行封装为异步任务当一个节点修改 State 后这个修改会通过一个异步队列或通道立即发送给外部的迭代器而不是等到所有节点执行完。这要求你的整个调用链服务器、网络、客户端都支持异步和非阻塞处理这也是为什么在热词中会出现大量网络断开stream disconnected相关错误的原因——任何一环的同步阻塞或超时都可能导致连接中断。3. 实战构建一个支持流式输出的问答智能体理论说得再多不如动手实践。我们来构建一个稍微真实一点的智能体它接收用户问题决定是否需要调用网络搜索然后生成回答并且整个过程以流式方式呈现。我们将使用LangGraphTavily搜索工具 Ollama本地运行Llama 3.2模型来搭建。请确保已安装相关包pip install langgraph langchain-community tavily-python ollama。3.1 定义状态与工具首先定义智能体的状态。对于问答场景我们需要记录对话历史、问题、搜索结果和最终答案。from typing import TypedDict, Annotated, List, Union from langgraph.graph import StateGraph, END from langchain_community.tools.tavily_search import TavilySearchResults from langchain_community.chat_models import ChatOllama from langchain_core.messages import HumanMessage, AIMessage, ToolMessage import operator import asyncio # 定义状态结构 class AgentState(TypedDict): # 消息列表记录完整对话 messages: Annotated[List[Union[HumanMessage, AIMessage, ToolMessage]], operator.add] # 用户当前问题也可从messages里取这里单独存便于使用 question: str # 是否需要搜索的标志 should_search: bool # 搜索得到的结果 search_results: List[dict] # 模型生成的最终回答流式生成的关键 final_answer: str # 初始化工具和模型 search_tool TavilySearchResults(max_results2) # 限制2条结果避免太长 llm ChatOllama(modelllama3.2, temperature0) # 使用本地Ollama的Llama模型3.2 构建图节点路由、搜索、回答一个典型的智能体工作流包含路由决策、执行工具、处理结果、生成回答等节点。# 节点1路由决策 - 判断是否需要搜索 def route_question(state: AgentState): 根据问题决定是否需要调用搜索工具。 last_message state[messages][-1] question last_message.content if hasattr(last_message, content) else state[question] # 一个简单的基于关键词的路由逻辑。实践中可以用LLM判断。 need_search_keywords [最近, 新闻, 2024, 谁, 哪里, 如何] should_search any(keyword in question for keyword in need_search_keywords) # 更新状态 return { question: question, should_search: should_search, messages: [AIMessage(contentf系统判断是否需要搜索: {should_search})] } # 节点2执行搜索 def call_search_tool(state: AgentState): 调用搜索工具获取信息。 if not state[should_search]: return {search_results: [], messages: [AIMessage(content无需搜索。)]} try: results search_tool.invoke({query: state[question]}) # 格式化结果便于后续使用 formatted_results [{title: r[title], content: r[content][:200]} for r in results] tool_message ToolMessage(contentstr(formatted_results), tool_call_idsearch_01) return { search_results: formatted_results, messages: [tool_message] } except Exception as e: error_msg f搜索失败: {e} return {search_results: [], messages: [AIMessage(contenterror_msg)]} # 节点3生成最终回答流式核心 async def generate_streaming_answer(state: AgentState): 基于对话历史和搜索结果流式生成最终答案。 # 准备对话上下文 context_messages state[messages][-5:] # 取最近5条消息作为上下文 # 如果有搜索结果将其作为系统提示的一部分 system_prompt 你是一个有帮助的AI助手。请根据以下信息回答用户问题信息不足时可说明。 if state[search_results]: search_info \n.join([f- {r[title]}: {r[content]} for r in state[search_results]]) system_prompt f{system_prompt}\n\n【搜索到的信息】\n{search_info} full_prompt [{role: system, content: system_prompt}] context_messages # 关键调用模型的流式接口 final_answer_chunks [] streamed_text # 这里模拟流式生成。实际应调用 llm.astream() # 例如async for chunk in llm.astream(full_prompt): # 为演示我们模拟一个延迟生成的过程 simulated_answer 基于现有信息这是一个流式生成的示例回答。它会被逐词返回。 for i, char in enumerate(simulated_answer): await asyncio.sleep(0.05) # 模拟生成延迟 streamed_text char final_answer_chunks.append(char) # 关键点在流式生成过程中我们可以持续更新state并yield出去 # 但LangGraph的节点函数本身不支持yield更新是在节点执行完后统一发生的。 # 因此对于真正的逐词流式需要将生成器放在图的外部循环中或者使用支持中间yield的自定义节点。 # 本例中我们先将所有内容收集再一次性更新。 # 节点执行完毕更新最终答案 return { final_answer: streamed_text, messages: [AIMessage(contentstreamed_text)] }3.3 组装工作流并测试流式输出现在我们将节点组装成图并设置条件边Conditional Edge来实现动态路由。# 构建图 builder StateGraph(AgentState) builder.add_node(route, route_question) builder.add_node(search, call_search_tool) builder.add_node(answer, generate_streaming_answer) builder.set_entry_point(route) # 设置条件边根据 should_search 决定下一步 def decide_next_step(state: AgentState): if state[should_search]: return search else: return answer builder.add_conditional_edges( route, decide_next_step, {search: search, answer: answer} ) builder.add_edge(search, answer) builder.add_edge(answer, END) agent_graph builder.compile() # 让我们测试一下流式更新Updates模式 async def run_streaming_demo(): print( 启动智能体流式演示 ) initial_input { messages: [HumanMessage(contentLangGraph是什么)], question: , should_search: False, search_results: [], final_answer: } print(执行流式 (Updates 模式):) async for node_name, update in agent_graph.astream(initial_input, stream_modeupdates): print(f\n[节点完成] {node_name}) # 打印有意义的更新内容 for key, val in update.items(): if key messages and val: last_msg val[-1] print(f 消息: {type(last_msg).__name__}: {last_msg.content[:80]}...) elif key final_answer: print(f 最终答案更新: {val[:50]}...) elif val: # 忽略空值 print(f {key}: {val}) print(\n 演示结束 ) # 运行演示 asyncio.run(run_streaming_demo())这个演示会展示工作流如何一步步执行route节点判断是否需要搜索然后执行search节点如果需要最后answer节点生成回复。在updates模式下你能清晰地看到每个节点对AgentState所做的修改。4. 陷阱、排错与性能优化让流式稳定可靠从热词中大量的stream disconnected错误可以看出流式接口的稳定性是实战中的最大挑战。以下是我在多个项目中总结的关键陷阱和解决方案。4.1 连接断开网络、超时与心跳错误信息stream disconnected before completion: error sending request for url是流式应用的“头号杀手”。其根源通常在于网络层不稳定代理、防火墙、负载均衡器可能中断长连接。超时设置不当服务器、反向代理如Nginx、客户端或HTTP库的读写超时时间设置过短。流式响应可能持续数十秒任何一方的超时设置小于这个时间都会导致连接被强行关闭。缺乏心跳机制在生成间隙长时间没有数据发送一些激进的网络设备或云服务商可能会切断空闲连接。解决方案配置超时将相关超时设置调大。例如在 Nginx 中location /api/stream { proxy_read_timeout 300s; # 读超时300秒 proxy_send_timeout 300s; # 写超时300秒 proxy_buffering off; # 关键关闭代理缓冲否则Nginx会尝试缓存完整响应 proxy_cache off; chunked_transfer_encoding on; # 启用分块传输编码 proxy_set_header Connection ; proxy_http_version 1.1; }实现应用层心跳在流式生成过程中定期发送一个心跳包如一个空格字符或特定的注释行data: \n\n来保持连接活跃。这需要在前端和后端协议中约定。使用 WebSocket 或 SSE对于复杂的双向流式交互WebSocket 是更自然的选择。对于简单的服务器推送Server-Sent Events (SSE) 比长轮询更高效且天然支持断线重连。LangGraph 的astream返回的异步迭代器可以很方便地适配到 SSE 端点。4.2 前端对接处理分块数据与错误前端对接流式接口绝不是简单的fetch().then()。你需要使用能够处理流式数据的API。使用 Fetch API 处理流async function streamLangGraphAnswer(question) { const response await fetch(/api/agent/stream, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ input: question }), }); if (!response.ok || !response.body) { throw new Error(Network response was not ok); } const reader response.body.getReader(); const decoder new TextDecoder(utf-8); let buffer ; try { while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); // 处理可能的行分隔数据如SSE的data:格式 const lines buffer.split(\n); buffer lines.pop(); // 最后一行可能不完整放回缓冲区 for (const line of lines) { if (line.startsWith(data: )) { const data line.slice(6); if (data [DONE]) break; // 结束标记 try { const parsed JSON.parse(data); // 更新UIparsed.update 或 parsed.value updateUI(parsed); } catch (e) { console.warn(Failed to parse stream data:, data); } } } } } catch (error) { console.error(Stream reading failed:, error); // 在这里处理连接断开错误可以提示用户或尝试重连 showError(连接中断请重试); } finally { reader.releaseLock(); } }关键点错误处理流式读取必须在try...catch块内进行并妥善处理reader.read()可能抛出的异常如网络断开。数据缓冲与解析TCP数据是分块到达的一个JSON对象可能被拆成多个chunk。前端需要实现简单的缓冲和拼接逻辑确保能解析出完整的数据包。使用SSE格式data: {...}\n\n可以简化这一过程。连接状态管理提供UI反馈如“正在输入...”并在连接断开时提供重试按钮。4.3 权限与安全Spring Security的启示热词中提到了yudao-cloud项目中flux流式输出与spring security的权限控制问题。这指向了一个深层问题传统的基于Filter链的权限拦截可能会破坏流式响应。在Spring MVC中Filter和Interceptor通常会在响应提交后即response.flushBuffer()或response.getWriter().close()才执行后续逻辑。但流式响应需要多次写入并刷新缓冲区。如果安全框架在第一次刷新后就认为请求已完成并尝试清理资源或进行后续拦截可能会导致流被意外关闭或报错。解决方案思路为流式端点配置特殊的安全规则在Spring Security配置中将流式API的路径排除在可能干扰响应流的Filter之外或为其配置更宽松的规则。使用响应式安全对于使用Spring WebFlux基于Reactor的流式端点配套使用Spring Security Reactive它专为非阻塞、流式场景设计。手动管理响应在控制器中直接获取HttpServletResponse对象手动控制写入和刷新并确保不在流结束前触发任何可能导致连接关闭的框架行为。4.4 性能与资源管理避免内存泄漏流式处理是长生命周期的操作必须谨慎管理资源。及时清理确保在流式迭代器完成或客户端断开连接时及时关闭数据库连接、文件句柄、外部API会话等资源。使用try...finally块或异步上下文管理器。背压Backpressure处理如果生成速度远快于客户端消费速度如下游网络慢会导致数据在服务器内存中堆积。在设计流式接口时应考虑背压机制。简单的做法是使用有界队列或者利用像asyncio.Queue这样的结构在队列满时暂停生成。监控与熔断对活跃的流式连接数进行监控。如果某个工作流节点异常缓慢或阻塞可能会拖垮整个服务器。实现超时中断和熔断机制防止单个请求占用资源过久。5. 超越基础高级流式模式与自定义控制当你掌握了基础的流式输出后可以探索更高级的模式来满足复杂需求。5.1 子图Subgraph的流式在复杂应用中你可能将某些功能模块封装为子图。主图流式执行时子图内部的执行细节默认是不会暴露出来的。如果你需要监控子图内部的步骤有两种方法将子图也编译为可流式对象在主图的流式循环中遇到需要调用子图的节点时手动创建子图的流式迭代器并嵌套循环将子图的更新也合并到主数据流中。这会增加代码复杂度。使用“调试”或“追踪”模式LangGraph 本身提供了更强大的Tracer或类似机制来记录详细的执行轨迹。你可以自定义一个Tracer在节点开始、结束时记录信息并将这些信息通过另一个通道如WebSocket实时推送给前端实现比标准updates更细粒度的观测。5.2 暂停、继续与取消流式执行热词中提到了langgraph compiledstategraph.stream()如何终止。这是一个高级需求例如用户中途不想等了点击了“停止生成”。.stream()方法返回的异步迭代器其生命周期与生成它的任务绑定。要取消执行本质上是取消这个异步任务。import asyncio async def stream_with_cancel(graph, inputs, cancel_event: asyncio.Event): 一个支持外部取消的流式包装函数。 try: async for item in graph.astream(inputs, stream_modeupdates): if cancel_event.is_set(): print(检测到取消信号正在终止流...) # 这里可以尝试一些清理工作 break # 跳出循环迭代器会被异步清理 yield item except asyncio.CancelledError: print(流式任务被取消。) raise finally: print(流式任务结束清理。) # 使用示例 async def main(): cancel_event asyncio.Event() stream_task asyncio.create_task(consume_stream(cancel_event)) # 模拟3秒后用户取消 await asyncio.sleep(3) cancel_event.set() await stream_task async def consume_stream(cancel_event): async for update in stream_with_cancel(agent_graph, initial_input, cancel_event): print(update) # 处理更新...更优雅的方式是利用asyncio的Task.cancel()。你可以将graph.astream()放在一个独立的asyncio.Task中运行当需要取消时调用task.cancel()。但需要注意在节点函数中处理asyncio.CancelledError异常进行资源清理。5.3 将流式输出适配到不同协议LangGraph 的异步迭代器是协议无关的。你可以轻松地将其桥接到不同的网络协议适配 Server-Sent Events (SSE):from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse import json app FastAPI() app.post(/stream) async def stream_endpoint(request: Request): data await request.json() inputs data.get(input, {}) async def event_generator(): async for node_name, update in agent_graph.astream(inputs, stream_modeupdates): # 将数据格式化为SSE格式 yield fdata: {json.dumps({node: node_name, update: update})}\n\n yield data: [DONE]\n\n # 结束标记 return StreamingResponse(event_generator(), media_typetext/event-stream)适配 WebSocket:from fastapi import WebSocket app.websocket(/ws) async def websocket_endpoint(websocket: WebSocket): await websocket.accept() try: while True: data await websocket.receive_json() inputs data.get(input, {}) async for node_name, update in agent_graph.astream(inputs, stream_modeupdates): await websocket.send_json({type: update, node: node_name, data: update}) await websocket.send_json({type: done}) except Exception as e: print(fWebSocket error: {e}) finally: await websocket.close()选择哪种协议取决于你的具体需求SSE更简单单向服务器推送WebSocket功能更全支持双向通信适合需要客户端实时发送指令如暂停、修改查询的交互式智能体应用。流式输出绝非仅仅是让文字一个个蹦出来的“特效”它是构建现代、高效、可交互AI应用的核心技术组件。通过深入理解 LangGraph 的updates和values模式掌握其与前后端整合的实战技巧并规避网络、安全、资源上的各种陷阱你才能真正释放智能体工作流的潜力打造出用户体验卓越的AI产品。