1. 项目概述为什么我们需要流式输出做Web开发的朋友尤其是最近在捣鼓大语言模型LLM或者实时数据展示应用的肯定对“页面卡死转圈圈”这个场景不陌生。用户点了个按钮前端就进入漫长的等待后端可能在吭哧吭哧处理一个几十秒甚至几分钟的任务期间用户什么都做不了体验极差。这就是典型的同步阻塞请求-响应模式带来的问题。“极简智能体二Web 流式输出”这个项目要解决的就是这个核心痛点。它的目标很明确在后端处理耗时任务时让前端能够像看直播一样实时地、一段一段地接收到处理结果而不是干等着最后那个“大礼包”。这种技术我们称之为“流式输出”Streaming Output。想象一下ChatGPT那种一个字一个字“蹦”出来的效果那就是流式输出的典型应用。实现Web流式输出的技术方案有好几种比如WebSocket和Server-Sent EventsSSE。这个项目聚焦于SSE原因在于它足够“极简”。对于主要由服务器向客户端推送数据的场景比如状态更新、日志流、AI生成文本SSE比WebSocket更轻量、更简单它是基于HTTP协议的不需要额外的握手和复杂的连接管理对后端和前端的改造都更友好。结合FastAPI的异步特性和现代JavaScript的EventSource API我们可以用非常简洁的代码构建出体验流畅的实时Web应用。2. 技术选型解析SSE为何是“极简”之选在决定使用SSE之前我们得先看看赛场上的其他选手。最常被拿来比较的就是WebSocket。WebSocket是一个全双工通信协议连接建立后客户端和服务器可以随时互相发送消息功能非常强大。它适合需要高频、双向实时交互的场景比如在线游戏、协同编辑、实时聊天。但是它的“重”也是显而易见的需要单独的ws://或wss://协议有一套复杂的握手和帧协议服务器和客户端都需要专门的库来处理连接、心跳、重连等逻辑。Server-Sent Events (SSE)则走了另一条路。它基于普通的HTTP/HTTPS协议本质上是一个长连接的HTTP响应。服务器通过这个连接可以持续地向客户端发送一系列格式固定的文本消息。它的特点是单向通道仅支持服务器向客户端推送。对于智能体输出、新闻推送、股票价格更新这种“服务器说客户端听”的场景这反而是个优点概念更简单。自动重连EventSource对象内置了断线重连机制。简单的文本协议消息格式就是data: content\n\n清晰易懂。天然兼容HTTP生态无需担心防火墙拦截因为就是HTTP可以利用HTTP认证、缓存等现有设施。对于“极简智能体”这个项目核心需求是将后端AI模型或处理逻辑生成的内容实时流式地推送到网页上。这是一个典型的单向、文本为主的推送场景。选择SSE意味着后端极简FastAPI中一个返回StreamingResponse的异步生成器函数就能搞定。前端极简浏览器原生支持EventSourceAPI几行代码就能监听事件。心智负担极简你不需要管理双向连接状态只需要关心如何生成数据流并推出去。注意SSE有一个重要限制它默认不支持跨域CORS。这意味着你的前端页面和后端API必须在同一个域名下或者后端需要正确配置CORS头部如Access-Control-Allow-Origin。FastAPI的CORSMiddleware可以轻松解决这个问题。所以“极简”并非功能阉割而是在满足核心需求服务器向客户端流式推送文本的前提下选择了最直接、最轻量、开发成本最低的技术路径。SSE就是这条路径上的利器。3. 后端实战用FastAPI构建SSE流式端点接下来我们动手实现。后端我们选择FastAPI因为它对异步的原生支持让流式响应写起来非常优雅。3.1 项目环境搭建与依赖安装首先创建一个新的项目目录并安装依赖。我们只需要fastapi和uvicorn一个ASGI服务器。# 创建并进入项目目录 mkdir simple-agent-sse cd simple-agent-sse # 创建虚拟环境推荐 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate # 安装核心依赖 pip install fastapi uvicorn为了模拟一个智能体处理耗时任务我们还需要一个能“慢慢”生成内容的函数。这里我们用asyncio.sleep来模拟延迟。3.2 核心流式端点实现创建一个main.py文件写入以下代码from fastapi import FastAPI from fastapi.responses import StreamingResponse from fastapi.middleware.cors import CORSMiddleware import asyncio import json import time app FastAPI(title极简智能体流式输出API) # 添加CORS中间件允许前端跨域访问根据你的前端地址调整 app.add_middleware( CORSMiddleware, allow_origins[http://localhost:3000], # 你的前端开发服务器地址 allow_credentialsTrue, allow_methods[*], allow_headers[*], ) async def mock_agent_thinking(task: str): 模拟一个智能体思考过程分段生成回答。 这是一个异步生成器函数每次yield一段文本。 steps [ f“收到用户请求{task}。正在分析意图...\n\n” “正在检索相关知识库...\n\n” “组织语言生成初步回答...\n\n” “最终答案如下\n\n” ] answer_parts [ “流式输出Streaming是一种数据传输技术。”, “它允许服务器在准备好一部分数据后就立即发送给客户端”, “而不是等待所有数据都处理完毕再一次性发送。”, “这能极大提升用户感知到的响应速度避免长时间白屏等待。”, “SSEServer-Sent Events是实现HTTP流式输出的标准之一。”, “\n\n【模拟结束】” ] # 模拟逐步思考过程 for step in steps: yield f“data: {json.dumps({type: status, content: step}, ensure_asciiFalse)}\n\n” await asyncio.sleep(1) # 模拟每步耗时1秒 # 模拟逐步生成答案 for part in answer_parts: yield f“data: {json.dumps({type: content, content: part}, ensure_asciiFalse)}\n\n” await asyncio.sleep(0.5) # 模拟每个词块生成耗时0.5秒 app.get(“/stream”) async def stream_response(task: str “请解释什么是流式输出”): SSE流式响应端点。 客户端通过GET请求访问 /stream?task你的问题 返回的是一个持续输出的文本流。 # 必须设置的SSE相关响应头 headers { “Content-Type”: “text/event-stream” “Cache-Control”: “no-cache” “Connection”: “keep-alive” “X-Accel-Buffering”: “no”, # 禁用Nginx等代理的缓冲对流式传输至关重要 } # 使用StreamingResponse包装异步生成器 return StreamingResponse( mock_agent_thinking(task), media_type“text/event-stream” headersheaders ) app.get(“/”) async def root(): return {“message”: “极简智能体SSE后端服务已启动请访问 /stream?task你的问题”} if __name__ “__main__”: import uvicorn uvicorn.run(app, host“0.0.0.0”, port8000)代码关键点解析异步生成器 (async def ... yield)mock_agent_thinking函数是核心。它使用async和yield使其成为一个异步生成器。这意味着它可以在生成一段数据后立即挂起将数据发送给客户端然后继续执行生成下一段而不是等所有数据都生成完。SSE事件格式SSE要求每条消息以data:开头以两个换行符\n\n结尾。我们通常将内容JSON序列化以便携带更多结构信息如消息类型type。StreamingResponse这是FastAPI提供的专用于流式响应的类。它接收一个异步生成器或普通生成器并自动处理数据的流式发送。关键响应头Content-Type: text/event-stream告诉浏览器这是一个SSE流。Cache-Control: no-cache禁止缓存确保客户端总是拿到最新数据。X-Accel-Buffering: no这个非常重要。如果你将服务部署在Nginx或Apache等反向代理之后这些代理默认会缓冲后端响应直到达到一定大小或超时这会导致流式输出“失效”客户端会等到缓冲满了才一次性收到。设置此头部可以禁用代理缓冲。CORS配置因为前端通常在不同端口如localhost:3000必须配置CORS以允许跨域请求否则浏览器会阻止连接。实操心得在开发环境直接用uvicorn运行可能没问题。但在生产环境使用NginxX-Accel-Buffering: no这个头部是流式传输能正常工作的关键否则你会遇到客户端接收延迟巨大或者收不到实时数据的情况。同时确保Nginx配置中proxy_buffering设置为off。3.3 运行与测试后端服务在终端运行uvicorn main:app --reload --port 8000访问http://localhost:8000/docs可以看到自动生成的API文档。但直接点/stream接口的“Try it out”在Swagger UI里看不到流式效果因为Swagger UI不是为消费SSE设计的。我们可以用一个简单的命令行工具curl来测试curl -N http://localhost:8000/stream?task测试任务-N参数用于禁用缓冲。你应该能看到数据一段一段地打印出来每条消息都符合data: {...}的格式。这说明后端SSE流已经成功建立。4. 前端实战用JavaScript的EventSource消费流式数据后端准备好了现在需要一个网页来接收并展示这个流。我们创建一个最简单的index.html。4.1 基础HTML与EventSource连接!DOCTYPE html html lang“zh-CN” head meta charset“UTF-8” meta name“viewport” content“widthdevice-width, initial-scale1.0” title极简智能体 - 流式输出演示/title style body { font-family: sans-serif; max-width: 800px; margin: 2rem auto; padding: 1rem; } #taskInput { width: 70%; padding: 0.5rem; margin-right: 1rem; } button { padding: 0.5rem 1.5rem; cursor: pointer; } #output { border: 1px solid #ccc; min-height: 300px; margin-top: 1rem; padding: 1rem; white-space: pre-wrap; /* 保留换行符 */ font-family: monospace; background-color: #f9f9f9; overflow-y: auto; } .status { color: #666; font-style: italic; } .content { color: #333; } .error { color: #d00; } /style /head body h1极简智能体 - Web流式输出演示/h1 div input type“text” id“taskInput” placeholder“请输入任务例如解释SSE的原理” value“解释SSE的原理” / button onclick“startStream()”开始流式处理/button button onclick“closeStream()”停止接收/button /div div id“output”等待任务开始.../div script let eventSource null; const outputDiv document.getElementById(‘output’); function startStream() { // 关闭可能存在的旧连接 if (eventSource) { eventSource.close(); } const task document.getElementById(‘taskInput’).value.trim() || ‘默认任务’; outputDiv.innerHTML ‘span class“status”正在连接服务器并启动智能体.../spanbr’; // 构建带查询参数的SSE连接URL const streamUrl http://localhost:8000/stream?task${encodeURIComponent(task)}; // 创建EventSource对象建立SSE连接 eventSource new EventSource(streamUrl); // 监听默认的message事件对应服务器发送的 data: 行 eventSource.onmessage function(event) { try { const data JSON.parse(event.data); // 解析我们后端发送的JSON appendToOutput(data); } catch (e) { // 如果数据不是JSON直接显示 appendToOutput({type: ‘content‘, content: event.data}); } }; // 监听自定义事件如果需要后端可以发送 event: customEventName // eventSource.addEventListener(‘customEvent‘, function(event) { ... }); // 监听连接打开事件 eventSource.onopen function() { appendToOutput({type: ‘status‘, content: ‘SSE连接已建立开始接收流...\n‘}); }; // 监听错误事件 eventSource.onerror function(error) { console.error(‘EventSource error:‘, error); // 注意EventSource在连接断开时会自动重连这里的error可能包含网络错误或服务器错误 appendToOutput({type: ‘error‘, content: 连接出现错误或已关闭。${eventSource.readyState EventSource.CLOSED ? ‘连接已关闭。’ : ‘正在尝试重连...’}\n}); // 如果确定要停止可以关闭连接 // if (eventSource.readyState EventSource.CLOSED) { // eventSource.close(); // } }; } function appendToOutput(data) { const span document.createElement(‘span‘); span.className data.type; // ‘status‘ 或 ‘content‘ 或 ‘error‘ span.textContent data.content; outputDiv.appendChild(span); // 可选自动滚动到底部 outputDiv.scrollTop outputDiv.scrollHeight; } function closeStream() { if (eventSource) { eventSource.close(); appendToOutput({type: ‘status‘, content: ‘\n已主动停止接收流。‘}); eventSource null; } } // 页面关闭时也关闭连接 window.addEventListener(‘beforeunload‘, closeStream); /script /body /html4.2 前端代码核心机制剖析EventSource对象这是浏览器原生提供的用于连接SSE的API。你只需要传入后端流式端点的URL它就会自动管理HTTP长连接、接收消息、解析事件并触发回调。它内置了断线重连机制非常省心。事件监听onmessage这是最常用的事件处理器。当服务器发送一条data: ...消息未指定event字段时就会触发。我们从event.data中获取服务器发送的原始字符串然后解析成JSON对象。onopen当SSE连接成功建立时触发。onerror当连接发生错误时触发。这里有个关键点EventSource在遇到网络错误或服务器中断时会自动尝试重新连接。onerror事件可能会被多次触发。你可以通过eventSource.readyState属性来判断当前连接状态CONNECTING0,OPEN1,CLOSED2。数据渲染我们将接收到的数据根据type字段添加不同的CSS类status,content以不同样式如灰色斜体、正常黑色显示在页面上模拟智能体“思考状态”和“输出内容”的区别。连接管理提供了closeStream函数可以手动调用eventSource.close()来关闭连接。在页面卸载时也调用它是一个好习惯。4.3 运行与体验由于浏览器同源策略你不能直接用浏览器打开本地的file://路径来访问localhost:8000的API。你需要通过一个HTTP服务器来提供这个HTML文件。最简单的方法是使用Python内置的HTTP服务器。在index.html所在目录下打开另一个终端# Python 3 python -m http.server 3000然后用浏览器访问http://localhost:3000或你指定的端口。在输入框输入问题点击“开始流式处理”你就能看到文字像打字机一样一段一段地从服务器“流”到网页上中间有模拟的延迟。这就是Web流式输出的魅力。5. 进阶优化与生产环境考量基础功能跑通了但要用于实际项目还需要考虑更多。5.1 流式传输的稳定性保障心跳机制SSE连接可能因为代理超时、负载均衡器空闲超时等原因被断开。为了防止这种情况服务器可以定期发送一条注释行以:开头的行作为“心跳包”保持连接活跃。async def stream_with_heartbeat(): while True: # 发送业务数据 yield f“data: ...\n\n” await asyncio.sleep(1) # 每10秒发送一次心跳 yield “: heartbeat\n\n”客户端重连策略虽然EventSource有自动重连但你可能需要更精细的控制比如重连次数限制、指数退避等。这需要你在onerror事件中自己实现并创建新的EventSource实例。连接标识与状态恢复对于重要的长任务如果连接中断重连客户端可能需要告诉服务器“我从哪里断开的”。这通常需要在连接URL中携带一个唯一的session_id服务器端根据这个ID恢复上下文。5.2 与后端业务逻辑深度集成我们的示例用了模拟函数。真实场景中你需要将SSE生成器与你的核心业务逻辑比如调用LLM API、执行长时间计算、读取数据库日志结合起来。app.get(“/stream-complex”) async def stream_complex(task: str, session_id: str None): async def event_generator(): # 1. 根据session_id恢复上下文如果有 context load_context(session_id) if session_id else None # 2. 调用真实的、支持流式的AI模型API例如OpenAI, Anthropic等 # 假设有一个异步的、支持流式响应的模型调用函数 async for chunk in call_llm_streaming(task, context): # chunk可能是模型返回的原始文本或结构化数据 yield format_sse_event(‘content‘, chunk) # 3. 任务完成发送结束事件 yield format_sse_event(‘end‘, {‘session_id‘: new_session_id}) return StreamingResponse(event_generator(), ...)关键在于你的业务逻辑本身需要是“可迭代的”或“可生成”的。很多现代的AI SDK如openai库都直接提供了异步的流式响应对象。5.3 错误处理与边界情况客户端提前关闭用户关闭了网页或主动停止了流。服务器端的生成器应该能感知到并优雅地停止释放资源比如中断昂贵的模型调用。在FastAPI中当客户端断开连接时StreamingResponse会抛出一个asyncio.CancelledError或其他异常。你需要在生成器内部用try...except捕获并处理。async def event_generator(): try: while True: data await get_next_chunk() yield data except asyncio.CancelledError: print(“客户端断开连接清理资源...”) # 执行清理操作如取消模型请求 raise # 重新抛出以正常结束消息格式与编码确保发送的文本是UTF-8编码。对于包含换行符的内容SSE协议要求用单个\n消息结束用\n\n。JSON序列化时使用ensure_asciiFalse来正确支持中文。性能与并发一个SSE连接就是一个长期持有的HTTP连接。在高并发下这可能会消耗较多的服务器资源如文件描述符。确保你的服务器如uvicorn和操作系统配置了足够的资源来处理大量并发连接。对于超大规模应用可能需要考虑使用专门的消息队列如Redis Pub/Sub来解耦生产者和消费者并由专门的SSE服务节点来推送。6. 常见问题排查与调试技巧在实际开发中你肯定会遇到流不出来的情况。下面是一个快速排查清单问题现象可能原因排查步骤与解决方案前端一直显示“正在连接...”onopen不触发1. CORS问题2. 后端服务未运行3. 网络问题1. 打开浏览器开发者工具F12的“网络”(Network)标签查看对/stream的请求是否被阻塞检查Console是否有CORS错误。2. 确认后端服务uvicorn是否在运行端口是否正确。3. 用curl -N命令直接测试后端接口是否正常输出流。连接能建立但收不到任何数据一段时间后断开1. 代理服务器如Nginx缓冲2. 服务器响应头不正确3. 生成器函数卡住或报错1.最关键检查后端响应头是否包含X-Accel-Buffering: no检查Nginx配置proxy_buffering off;。2. 确认响应头Content-Type: text/event-stream。3. 在后端生成器函数中添加日志看是否正常执行到yield语句。检查是否有未处理的异常。数据接收不完整或者一次性全部收到没有流式效果1. 前端EventSource解析问题2. 服务器没有正确分块发送1. 用curl -N测试看数据是否真的是分块到达的。如果是问题在前端。2. 确保服务器是使用yield逐段发送数据而不是在函数最后返回一个完整的字符串或列表。FastAPI的StreamingResponse必须接收一个生成器。中文显示乱码字符编码问题1. 确保后端在json.dumps时使用了ensure_asciiFalse。2. 确保HTTP响应头或HTML页面指定了UTF-8编码。在Safari或某些移动浏览器上不工作浏览器兼容性或特定限制1. Safari对SSE的并发连接数有更严格的限制。2. 检查浏览器是否启用了JavaScriptSSE依赖JS。3. 考虑使用Polyfill库如eventsource库来增强兼容性但现代浏览器支持已很好。调试黄金法则永远先用curl -N测试后端。如果curl能实时看到数据流出来那么问题大概率在前端或网络配置CORS、代理。如果curl也收不到或一次性收到问题一定在后端代码或服务器配置。流式输出尤其是SSE一旦调通其稳定性和简洁性会带来巨大的开发愉悦感。它把复杂的实时通信抽象成了一个简单的“事件监听”模型让开发者能更专注于业务逻辑本身。对于构建需要向用户实时反馈进度的Web应用无论是AI对话、数据导出、日志监控还是文件处理SSE都是一个值得优先考虑的“极简”而强大的工具。
1. 项目概述为什么我们需要流式输出做Web开发的朋友尤其是最近在捣鼓大语言模型LLM或者实时数据展示应用的肯定对“页面卡死转圈圈”这个场景不陌生。用户点了个按钮前端就进入漫长的等待后端可能在吭哧吭哧处理一个几十秒甚至几分钟的任务期间用户什么都做不了体验极差。这就是典型的同步阻塞请求-响应模式带来的问题。“极简智能体二Web 流式输出”这个项目要解决的就是这个核心痛点。它的目标很明确在后端处理耗时任务时让前端能够像看直播一样实时地、一段一段地接收到处理结果而不是干等着最后那个“大礼包”。这种技术我们称之为“流式输出”Streaming Output。想象一下ChatGPT那种一个字一个字“蹦”出来的效果那就是流式输出的典型应用。实现Web流式输出的技术方案有好几种比如WebSocket和Server-Sent EventsSSE。这个项目聚焦于SSE原因在于它足够“极简”。对于主要由服务器向客户端推送数据的场景比如状态更新、日志流、AI生成文本SSE比WebSocket更轻量、更简单它是基于HTTP协议的不需要额外的握手和复杂的连接管理对后端和前端的改造都更友好。结合FastAPI的异步特性和现代JavaScript的EventSource API我们可以用非常简洁的代码构建出体验流畅的实时Web应用。2. 技术选型解析SSE为何是“极简”之选在决定使用SSE之前我们得先看看赛场上的其他选手。最常被拿来比较的就是WebSocket。WebSocket是一个全双工通信协议连接建立后客户端和服务器可以随时互相发送消息功能非常强大。它适合需要高频、双向实时交互的场景比如在线游戏、协同编辑、实时聊天。但是它的“重”也是显而易见的需要单独的ws://或wss://协议有一套复杂的握手和帧协议服务器和客户端都需要专门的库来处理连接、心跳、重连等逻辑。Server-Sent Events (SSE)则走了另一条路。它基于普通的HTTP/HTTPS协议本质上是一个长连接的HTTP响应。服务器通过这个连接可以持续地向客户端发送一系列格式固定的文本消息。它的特点是单向通道仅支持服务器向客户端推送。对于智能体输出、新闻推送、股票价格更新这种“服务器说客户端听”的场景这反而是个优点概念更简单。自动重连EventSource对象内置了断线重连机制。简单的文本协议消息格式就是data: content\n\n清晰易懂。天然兼容HTTP生态无需担心防火墙拦截因为就是HTTP可以利用HTTP认证、缓存等现有设施。对于“极简智能体”这个项目核心需求是将后端AI模型或处理逻辑生成的内容实时流式地推送到网页上。这是一个典型的单向、文本为主的推送场景。选择SSE意味着后端极简FastAPI中一个返回StreamingResponse的异步生成器函数就能搞定。前端极简浏览器原生支持EventSourceAPI几行代码就能监听事件。心智负担极简你不需要管理双向连接状态只需要关心如何生成数据流并推出去。注意SSE有一个重要限制它默认不支持跨域CORS。这意味着你的前端页面和后端API必须在同一个域名下或者后端需要正确配置CORS头部如Access-Control-Allow-Origin。FastAPI的CORSMiddleware可以轻松解决这个问题。所以“极简”并非功能阉割而是在满足核心需求服务器向客户端流式推送文本的前提下选择了最直接、最轻量、开发成本最低的技术路径。SSE就是这条路径上的利器。3. 后端实战用FastAPI构建SSE流式端点接下来我们动手实现。后端我们选择FastAPI因为它对异步的原生支持让流式响应写起来非常优雅。3.1 项目环境搭建与依赖安装首先创建一个新的项目目录并安装依赖。我们只需要fastapi和uvicorn一个ASGI服务器。# 创建并进入项目目录 mkdir simple-agent-sse cd simple-agent-sse # 创建虚拟环境推荐 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate # 安装核心依赖 pip install fastapi uvicorn为了模拟一个智能体处理耗时任务我们还需要一个能“慢慢”生成内容的函数。这里我们用asyncio.sleep来模拟延迟。3.2 核心流式端点实现创建一个main.py文件写入以下代码from fastapi import FastAPI from fastapi.responses import StreamingResponse from fastapi.middleware.cors import CORSMiddleware import asyncio import json import time app FastAPI(title极简智能体流式输出API) # 添加CORS中间件允许前端跨域访问根据你的前端地址调整 app.add_middleware( CORSMiddleware, allow_origins[http://localhost:3000], # 你的前端开发服务器地址 allow_credentialsTrue, allow_methods[*], allow_headers[*], ) async def mock_agent_thinking(task: str): 模拟一个智能体思考过程分段生成回答。 这是一个异步生成器函数每次yield一段文本。 steps [ f“收到用户请求{task}。正在分析意图...\n\n” “正在检索相关知识库...\n\n” “组织语言生成初步回答...\n\n” “最终答案如下\n\n” ] answer_parts [ “流式输出Streaming是一种数据传输技术。”, “它允许服务器在准备好一部分数据后就立即发送给客户端”, “而不是等待所有数据都处理完毕再一次性发送。”, “这能极大提升用户感知到的响应速度避免长时间白屏等待。”, “SSEServer-Sent Events是实现HTTP流式输出的标准之一。”, “\n\n【模拟结束】” ] # 模拟逐步思考过程 for step in steps: yield f“data: {json.dumps({type: status, content: step}, ensure_asciiFalse)}\n\n” await asyncio.sleep(1) # 模拟每步耗时1秒 # 模拟逐步生成答案 for part in answer_parts: yield f“data: {json.dumps({type: content, content: part}, ensure_asciiFalse)}\n\n” await asyncio.sleep(0.5) # 模拟每个词块生成耗时0.5秒 app.get(“/stream”) async def stream_response(task: str “请解释什么是流式输出”): SSE流式响应端点。 客户端通过GET请求访问 /stream?task你的问题 返回的是一个持续输出的文本流。 # 必须设置的SSE相关响应头 headers { “Content-Type”: “text/event-stream” “Cache-Control”: “no-cache” “Connection”: “keep-alive” “X-Accel-Buffering”: “no”, # 禁用Nginx等代理的缓冲对流式传输至关重要 } # 使用StreamingResponse包装异步生成器 return StreamingResponse( mock_agent_thinking(task), media_type“text/event-stream” headersheaders ) app.get(“/”) async def root(): return {“message”: “极简智能体SSE后端服务已启动请访问 /stream?task你的问题”} if __name__ “__main__”: import uvicorn uvicorn.run(app, host“0.0.0.0”, port8000)代码关键点解析异步生成器 (async def ... yield)mock_agent_thinking函数是核心。它使用async和yield使其成为一个异步生成器。这意味着它可以在生成一段数据后立即挂起将数据发送给客户端然后继续执行生成下一段而不是等所有数据都生成完。SSE事件格式SSE要求每条消息以data:开头以两个换行符\n\n结尾。我们通常将内容JSON序列化以便携带更多结构信息如消息类型type。StreamingResponse这是FastAPI提供的专用于流式响应的类。它接收一个异步生成器或普通生成器并自动处理数据的流式发送。关键响应头Content-Type: text/event-stream告诉浏览器这是一个SSE流。Cache-Control: no-cache禁止缓存确保客户端总是拿到最新数据。X-Accel-Buffering: no这个非常重要。如果你将服务部署在Nginx或Apache等反向代理之后这些代理默认会缓冲后端响应直到达到一定大小或超时这会导致流式输出“失效”客户端会等到缓冲满了才一次性收到。设置此头部可以禁用代理缓冲。CORS配置因为前端通常在不同端口如localhost:3000必须配置CORS以允许跨域请求否则浏览器会阻止连接。实操心得在开发环境直接用uvicorn运行可能没问题。但在生产环境使用NginxX-Accel-Buffering: no这个头部是流式传输能正常工作的关键否则你会遇到客户端接收延迟巨大或者收不到实时数据的情况。同时确保Nginx配置中proxy_buffering设置为off。3.3 运行与测试后端服务在终端运行uvicorn main:app --reload --port 8000访问http://localhost:8000/docs可以看到自动生成的API文档。但直接点/stream接口的“Try it out”在Swagger UI里看不到流式效果因为Swagger UI不是为消费SSE设计的。我们可以用一个简单的命令行工具curl来测试curl -N http://localhost:8000/stream?task测试任务-N参数用于禁用缓冲。你应该能看到数据一段一段地打印出来每条消息都符合data: {...}的格式。这说明后端SSE流已经成功建立。4. 前端实战用JavaScript的EventSource消费流式数据后端准备好了现在需要一个网页来接收并展示这个流。我们创建一个最简单的index.html。4.1 基础HTML与EventSource连接!DOCTYPE html html lang“zh-CN” head meta charset“UTF-8” meta name“viewport” content“widthdevice-width, initial-scale1.0” title极简智能体 - 流式输出演示/title style body { font-family: sans-serif; max-width: 800px; margin: 2rem auto; padding: 1rem; } #taskInput { width: 70%; padding: 0.5rem; margin-right: 1rem; } button { padding: 0.5rem 1.5rem; cursor: pointer; } #output { border: 1px solid #ccc; min-height: 300px; margin-top: 1rem; padding: 1rem; white-space: pre-wrap; /* 保留换行符 */ font-family: monospace; background-color: #f9f9f9; overflow-y: auto; } .status { color: #666; font-style: italic; } .content { color: #333; } .error { color: #d00; } /style /head body h1极简智能体 - Web流式输出演示/h1 div input type“text” id“taskInput” placeholder“请输入任务例如解释SSE的原理” value“解释SSE的原理” / button onclick“startStream()”开始流式处理/button button onclick“closeStream()”停止接收/button /div div id“output”等待任务开始.../div script let eventSource null; const outputDiv document.getElementById(‘output’); function startStream() { // 关闭可能存在的旧连接 if (eventSource) { eventSource.close(); } const task document.getElementById(‘taskInput’).value.trim() || ‘默认任务’; outputDiv.innerHTML ‘span class“status”正在连接服务器并启动智能体.../spanbr’; // 构建带查询参数的SSE连接URL const streamUrl http://localhost:8000/stream?task${encodeURIComponent(task)}; // 创建EventSource对象建立SSE连接 eventSource new EventSource(streamUrl); // 监听默认的message事件对应服务器发送的 data: 行 eventSource.onmessage function(event) { try { const data JSON.parse(event.data); // 解析我们后端发送的JSON appendToOutput(data); } catch (e) { // 如果数据不是JSON直接显示 appendToOutput({type: ‘content‘, content: event.data}); } }; // 监听自定义事件如果需要后端可以发送 event: customEventName // eventSource.addEventListener(‘customEvent‘, function(event) { ... }); // 监听连接打开事件 eventSource.onopen function() { appendToOutput({type: ‘status‘, content: ‘SSE连接已建立开始接收流...\n‘}); }; // 监听错误事件 eventSource.onerror function(error) { console.error(‘EventSource error:‘, error); // 注意EventSource在连接断开时会自动重连这里的error可能包含网络错误或服务器错误 appendToOutput({type: ‘error‘, content: 连接出现错误或已关闭。${eventSource.readyState EventSource.CLOSED ? ‘连接已关闭。’ : ‘正在尝试重连...’}\n}); // 如果确定要停止可以关闭连接 // if (eventSource.readyState EventSource.CLOSED) { // eventSource.close(); // } }; } function appendToOutput(data) { const span document.createElement(‘span‘); span.className data.type; // ‘status‘ 或 ‘content‘ 或 ‘error‘ span.textContent data.content; outputDiv.appendChild(span); // 可选自动滚动到底部 outputDiv.scrollTop outputDiv.scrollHeight; } function closeStream() { if (eventSource) { eventSource.close(); appendToOutput({type: ‘status‘, content: ‘\n已主动停止接收流。‘}); eventSource null; } } // 页面关闭时也关闭连接 window.addEventListener(‘beforeunload‘, closeStream); /script /body /html4.2 前端代码核心机制剖析EventSource对象这是浏览器原生提供的用于连接SSE的API。你只需要传入后端流式端点的URL它就会自动管理HTTP长连接、接收消息、解析事件并触发回调。它内置了断线重连机制非常省心。事件监听onmessage这是最常用的事件处理器。当服务器发送一条data: ...消息未指定event字段时就会触发。我们从event.data中获取服务器发送的原始字符串然后解析成JSON对象。onopen当SSE连接成功建立时触发。onerror当连接发生错误时触发。这里有个关键点EventSource在遇到网络错误或服务器中断时会自动尝试重新连接。onerror事件可能会被多次触发。你可以通过eventSource.readyState属性来判断当前连接状态CONNECTING0,OPEN1,CLOSED2。数据渲染我们将接收到的数据根据type字段添加不同的CSS类status,content以不同样式如灰色斜体、正常黑色显示在页面上模拟智能体“思考状态”和“输出内容”的区别。连接管理提供了closeStream函数可以手动调用eventSource.close()来关闭连接。在页面卸载时也调用它是一个好习惯。4.3 运行与体验由于浏览器同源策略你不能直接用浏览器打开本地的file://路径来访问localhost:8000的API。你需要通过一个HTTP服务器来提供这个HTML文件。最简单的方法是使用Python内置的HTTP服务器。在index.html所在目录下打开另一个终端# Python 3 python -m http.server 3000然后用浏览器访问http://localhost:3000或你指定的端口。在输入框输入问题点击“开始流式处理”你就能看到文字像打字机一样一段一段地从服务器“流”到网页上中间有模拟的延迟。这就是Web流式输出的魅力。5. 进阶优化与生产环境考量基础功能跑通了但要用于实际项目还需要考虑更多。5.1 流式传输的稳定性保障心跳机制SSE连接可能因为代理超时、负载均衡器空闲超时等原因被断开。为了防止这种情况服务器可以定期发送一条注释行以:开头的行作为“心跳包”保持连接活跃。async def stream_with_heartbeat(): while True: # 发送业务数据 yield f“data: ...\n\n” await asyncio.sleep(1) # 每10秒发送一次心跳 yield “: heartbeat\n\n”客户端重连策略虽然EventSource有自动重连但你可能需要更精细的控制比如重连次数限制、指数退避等。这需要你在onerror事件中自己实现并创建新的EventSource实例。连接标识与状态恢复对于重要的长任务如果连接中断重连客户端可能需要告诉服务器“我从哪里断开的”。这通常需要在连接URL中携带一个唯一的session_id服务器端根据这个ID恢复上下文。5.2 与后端业务逻辑深度集成我们的示例用了模拟函数。真实场景中你需要将SSE生成器与你的核心业务逻辑比如调用LLM API、执行长时间计算、读取数据库日志结合起来。app.get(“/stream-complex”) async def stream_complex(task: str, session_id: str None): async def event_generator(): # 1. 根据session_id恢复上下文如果有 context load_context(session_id) if session_id else None # 2. 调用真实的、支持流式的AI模型API例如OpenAI, Anthropic等 # 假设有一个异步的、支持流式响应的模型调用函数 async for chunk in call_llm_streaming(task, context): # chunk可能是模型返回的原始文本或结构化数据 yield format_sse_event(‘content‘, chunk) # 3. 任务完成发送结束事件 yield format_sse_event(‘end‘, {‘session_id‘: new_session_id}) return StreamingResponse(event_generator(), ...)关键在于你的业务逻辑本身需要是“可迭代的”或“可生成”的。很多现代的AI SDK如openai库都直接提供了异步的流式响应对象。5.3 错误处理与边界情况客户端提前关闭用户关闭了网页或主动停止了流。服务器端的生成器应该能感知到并优雅地停止释放资源比如中断昂贵的模型调用。在FastAPI中当客户端断开连接时StreamingResponse会抛出一个asyncio.CancelledError或其他异常。你需要在生成器内部用try...except捕获并处理。async def event_generator(): try: while True: data await get_next_chunk() yield data except asyncio.CancelledError: print(“客户端断开连接清理资源...”) # 执行清理操作如取消模型请求 raise # 重新抛出以正常结束消息格式与编码确保发送的文本是UTF-8编码。对于包含换行符的内容SSE协议要求用单个\n消息结束用\n\n。JSON序列化时使用ensure_asciiFalse来正确支持中文。性能与并发一个SSE连接就是一个长期持有的HTTP连接。在高并发下这可能会消耗较多的服务器资源如文件描述符。确保你的服务器如uvicorn和操作系统配置了足够的资源来处理大量并发连接。对于超大规模应用可能需要考虑使用专门的消息队列如Redis Pub/Sub来解耦生产者和消费者并由专门的SSE服务节点来推送。6. 常见问题排查与调试技巧在实际开发中你肯定会遇到流不出来的情况。下面是一个快速排查清单问题现象可能原因排查步骤与解决方案前端一直显示“正在连接...”onopen不触发1. CORS问题2. 后端服务未运行3. 网络问题1. 打开浏览器开发者工具F12的“网络”(Network)标签查看对/stream的请求是否被阻塞检查Console是否有CORS错误。2. 确认后端服务uvicorn是否在运行端口是否正确。3. 用curl -N命令直接测试后端接口是否正常输出流。连接能建立但收不到任何数据一段时间后断开1. 代理服务器如Nginx缓冲2. 服务器响应头不正确3. 生成器函数卡住或报错1.最关键检查后端响应头是否包含X-Accel-Buffering: no检查Nginx配置proxy_buffering off;。2. 确认响应头Content-Type: text/event-stream。3. 在后端生成器函数中添加日志看是否正常执行到yield语句。检查是否有未处理的异常。数据接收不完整或者一次性全部收到没有流式效果1. 前端EventSource解析问题2. 服务器没有正确分块发送1. 用curl -N测试看数据是否真的是分块到达的。如果是问题在前端。2. 确保服务器是使用yield逐段发送数据而不是在函数最后返回一个完整的字符串或列表。FastAPI的StreamingResponse必须接收一个生成器。中文显示乱码字符编码问题1. 确保后端在json.dumps时使用了ensure_asciiFalse。2. 确保HTTP响应头或HTML页面指定了UTF-8编码。在Safari或某些移动浏览器上不工作浏览器兼容性或特定限制1. Safari对SSE的并发连接数有更严格的限制。2. 检查浏览器是否启用了JavaScriptSSE依赖JS。3. 考虑使用Polyfill库如eventsource库来增强兼容性但现代浏览器支持已很好。调试黄金法则永远先用curl -N测试后端。如果curl能实时看到数据流出来那么问题大概率在前端或网络配置CORS、代理。如果curl也收不到或一次性收到问题一定在后端代码或服务器配置。流式输出尤其是SSE一旦调通其稳定性和简洁性会带来巨大的开发愉悦感。它把复杂的实时通信抽象成了一个简单的“事件监听”模型让开发者能更专注于业务逻辑本身。对于构建需要向用户实时反馈进度的Web应用无论是AI对话、数据导出、日志监控还是文件处理SSE都是一个值得优先考虑的“极简”而强大的工具。
)
