尧图网站建设 
1. 项目缘起为什么是 FastAPI Vue SSE最近在折腾一个AI问答项目后台用大模型生成答案前端需要实时、流畅地展示生成过程。最开始图省事直接让前端轮询接口或者等后端生成完整个答案再一次性返回。结果要么是前端疯狂刷请求服务器压力山大要么是用户盯着空白页面干等好几秒体验极差。这让我意识到在这种需要“边想边说”的交互场景里流式传输Streaming不是锦上添花而是雪中送炭。在众多技术方案里我最终锁定了FastAPI Vue SSE这个组合。为什么不是 WebSocket对于问答这种典型的“一问一答”、且主要由服务器向客户端推送数据的场景SSEServer-Sent Events协议更轻量、更简单。它基于 HTTP不需要额外的握手和复杂的连接管理浏览器原生支持后端实现也直观。FastAPI 对异步和流式响应的支持堪称优雅几行代码就能搭起一个高效的流式端点。Vue 作为前端主流框架其响应式系统能完美地将后端推送来的数据碎片实时渲染到页面上整个过程行云流水。这个组合的目标很明确搭建一个最小可行原型跑通从用户提问、后端流式生成、到前端逐字展示的完整链路。这不仅是技术上的验证更是为后续集成更复杂的 RAG检索增强生成、Agent 等能力打下坚实的地基。下面我就把从零开始搭建这个骨架的每一步包括我踩过的坑和总结的技巧毫无保留地分享出来。2. 后端基石用 FastAPI 构建 SSE 流式端点后端是整个流式问答的心脏负责接收问题、调用模型或模拟生成、并以流的形式吐出答案。FastAPI 的异步特性让这一切变得非常高效。2.1 环境准备与依赖安装首先创建一个干净的 Python 虚拟环境是专业开发的好习惯能避免包版本冲突。# 创建项目目录并进入 mkdir stream-qa-backend cd stream-qa-backend # 创建虚拟环境这里使用 venv你也可以用 conda python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate激活虚拟环境后安装核心依赖。除了 FastAPI我们还需要uvicorn作为 ASGI 服务器来运行应用sse-starlette这个库能帮我们更规范地处理 SSE 响应。pip install fastapi uvicorn sse-starlette这里有个小坑sse-starlette的版本。我一开始用了最新版结果和 FastAPI 的某些中间件有兼容性问题事件流无法正常关闭。后来锁定到sse-starlette1.6.5这个版本就非常稳定。所以建议在requirements.txt里直接指定fastapi0.104.1 uvicorn[standard]0.24.0 sse-starlette1.6.52.2 核心应用结构与 SSE 路由实现项目结构保持清晰很重要。我习惯这样组织stream-qa-backend/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用实例和主路由 │ └── api/ │ ├── __init__.py │ └── endpoints/ │ ├── __init__.py │ └── sse_stream.py # 专门的 SSE 流式端点 ├── requirements.txt └── .env # 环境变量如果需要现在我们来编写最核心的 SSE 端点。在app/api/endpoints/sse_stream.py中import asyncio import json from typing import AsyncGenerator from fastapi import APIRouter, Request from sse_starlette.sse import EventSourceResponse router APIRouter() # 模拟一个异步的、流式的文本生成器 async def mock_llm_stream_generator(prompt: str) - AsyncGenerator[str, None]: 模拟大语言模型的流式生成。 在实际项目中这里会替换为调用 OpenAI API、本地 Llama 模型等。 simulated_response f这是针对问题“{prompt}”生成的流式回答。 words list(simulated_response) for word in words: # 模拟每个词或token的生成延迟 await asyncio.sleep(0.05) # 以 SSE 格式要求的 data: 字段返回 yield json.dumps({token: word}, ensure_asciiFalse) # 可选发送一个结束事件 yield json.dumps({event: end, data: 生成结束}, ensure_asciiFalse) router.get(/stream-answer) async def stream_answer(request: Request, question: str): SSE流式问答端点。 客户端通过 GET 请求访问并传递 question 参数。 例如GET /api/stream-answer?question什么是人工智能 async def event_generator(): # 这里可以添加身份验证、请求日志等逻辑 try: async for chunk in mock_llm_stream_generator(question): # 关键SSE 格式要求每行数据以 data: 开头并以两个换行符结束 yield fdata: {chunk}\n\n # 检查客户端是否还连接着防止客户端断开后服务器还在发送数据 if await request.is_disconnected(): print(客户端断开连接) break except asyncio.CancelledError: # 处理任务被取消的情况例如客户端刷新页面 print(流式生成任务被取消) raise # 使用 EventSourceResponse 包装生成器并设置正确的媒体类型 return EventSourceResponse( event_generator(), headers{ Cache-Control: no-cache, Connection: keep-alive, Content-Type: text/event-stream, } )关键点解析与避坑EventSourceResponse的使用sse-starlette提供的这个响应类帮我们处理了 SSE 协议的大部分细节比如自动发送retry事件、正确处理连接关闭等。比自己手动构造响应省心太多。生成器函数async for我们使用async def定义了一个异步生成器函数mock_llm_stream_generator。这是 FastAPI 处理流式响应的核心模式。在实际项目中这个函数内部就是调用真实的 LLM API如 OpenAI、通义千问等并yield每一个返回的 token 或 chunk。请求断开检测if await request.is_disconnected():这行代码至关重要。没有它即使浏览器页面关闭了后端的生成器可能还在运行浪费服务器资源。这个检查确保了连接断开时及时退出循环。SSE 数据格式SSE 协议规定每条消息由若干行组成以两个换行符\n\n结尾。最常见的是data:行。我们yield fdata: {chunk}\n\n就是在构造这个格式。消息体chunk通常是一个 JSON 字符串方便前端解析。2.3 整合应用并配置 CORS接下来在app/main.py中创建 FastAPI 应用实例并挂载我们的路由。由于前端和后端通常是分开部署的跨域必须配置 CORS跨源资源共享。from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.api.endpoints import sse_stream # 创建 FastAPI 应用实例 app FastAPI(title流式问答后端 API, version1.0.0) # 配置 CORS 中间件 # 在生产环境中应严格指定 origins而不是用 * app.add_middleware( CORSMiddleware, allow_origins[*], # 允许所有前端源仅用于开发测试 allow_credentialsTrue, allow_methods[*], # 允许所有方法 allow_headers[*], # 允许所有头 ) # 挂载路由 app.include_router(sse_stream.router, prefix/api, tags[流式问答]) app.get(/) async def root(): return {message: Streaming QA Backend is running.}CORS 配置的注意事项在开发阶段为了方便我们设置了allow_origins[*]。但这在生产环境是极其危险的它会允许任何网站向你的后端发起请求。正确的做法是明确列出你前端应用部署的域名例如[https://your-vue-app.com, http://localhost:5173]。2.4 运行与测试后端服务在项目根目录下使用 Uvicorn 启动服务uvicorn app.main:app --reload --host 0.0.0.0 --port 8000--reload: 代码修改后自动重启仅用于开发。--host 0.0.0.0: 监听所有网络接口方便同一局域网内的设备如手机访问。--port 8000: 指定端口。启动后访问http://localhost:8000/docs就能看到自动生成的交互式 API 文档Swagger UI。你可以直接在那里测试/api/stream-answer接口。但更直观的测试方法是使用curl命令它能直接展示原始的 SSE 流curl -N http://localhost:8000/api/stream-answer?question你好世界如果看到一行行以data:开头的 JSON 数据流式地输出到终端恭喜你后端 SSE 服务已经成功跑起来了3. 前端呈现Vue 3 连接与消费 SSE 流后端流已经就绪现在需要前端来连接这个流并实时地将数据渲染到页面上。Vue 3 的组合式 API 让这个过程变得非常清晰。3.1 创建 Vue 项目与安装依赖使用官方的 Vite 工具创建 Vue 3 项目选择 TypeScript 和更现代的架构。npm create vuelatest stream-qa-frontend # 按照提示选择Vue Router可选 Pinia可选 ESLint等根据喜好。 cd stream-qa-frontend npm install为了更好的 UI 体验我们可以安装一个组件库。这里以 Element Plus 为例npm install element-plus element-plus/icons-vue然后在main.ts中全局引入import { createApp } from vue import App from ./App.vue import ElementPlus from element-plus import element-plus/dist/index.css const app createApp(App) app.use(ElementPlus) app.mount(#app)3.2 构建 SSE 连接管理模块我们不建议把 SSE 连接逻辑直接写在组件里而是封装成一个可复用的、易于管理状态和生命周期的模块。在src目录下创建utils/sseClient.ts// src/utils/sseClient.ts import { ref, onUnmounted } from vue interface SSEMessage { token?: string event?: string data?: any } export function useSSEStream(url: string) { const message ref() // 累积的完整消息 const isConnected ref(false) const isLoading ref(false) const error refError | null(null) let eventSource: EventSource | null null const connect () { if (eventSource) { console.warn(SSE 连接已存在正在关闭旧连接) close() } isLoading.value true error.value null message.value // 开始新的会话清空历史消息 try { // 创建 EventSource 实例浏览器原生 API eventSource new EventSource(url) isConnected.value true // 监听 message 事件标准数据事件 eventSource.onmessage (event) { isLoading.value false try { const data: SSEMessage JSON.parse(event.data) // 处理 token 类型的消息 if (data.token ! undefined) { message.value data.token // 逐 token 累加 } // 处理自定义事件例如结束事件 if (data.event end) { console.log(流式传输结束:, data.data) close() // 收到结束事件后可以主动关闭连接 } } catch (e) { console.error(解析 SSE 消息失败:, e, event.data) } } // 监听 error 事件 eventSource.onerror (err) { console.error(SSE 连接错误:, err) error.value new Error(SSE 连接异常或已断开) isLoading.value false isConnected.value false close() // 发生错误时清理资源 } // 监听 open 事件 eventSource.onopen () { console.log(SSE 连接已建立) isLoading.value false } } catch (err) { error.value err as Error isLoading.value false isConnected.value false } } const close () { if (eventSource) { eventSource.close() eventSource null isConnected.value false console.log(SSE 连接已关闭) } } // 组件卸载时自动关闭连接防止内存泄漏 onUnmounted(() { close() }) // 返回响应式状态和方法 return { message, isConnected, isLoading, error, connect, close } }这个模块的设计精妙之处响应式状态使用 Vue 的ref管理连接状态、加载状态、错误信息和累积的消息。这些状态变化会自动触发视图更新。生命周期管理通过onUnmounted钩子确保在组件销毁时自动关闭 EventSource 连接这是避免内存泄漏的关键。错误处理完善地处理了网络错误、解析错误并将错误信息暴露给组件便于 UI 展示。可复用性在任何 Vue 组件中只需调用useSSEStream(url)即可获得一套完整的 SSE 连接能力。3.3 实现问答页面组件现在我们来创建一个使用上述 Hook 的页面组件。在src/views或src/components下创建StreamQA.vuetemplate div classstream-qa-container el-card classbox-card template #header div classcard-header spanSSE 流式问答演示/span /div /template !-- 输入区域 -- div classinput-area el-input v-modelquestion placeholder请输入您的问题... :disabledisLoading keyup.enterhandleAsk template #append el-button :iconSearch clickhandleAsk :loadingisLoading :disabled!question.trim() 发送 /el-button /template /el-input /div !-- 状态与操作区域 -- div classstatus-area el-tag :typeisConnected ? success : info {{ isConnected ? 已连接 : 未连接 }} /el-tag el-button v-ifisConnected sizesmall clickhandleClose :disabledisLoading 断开连接 /el-button el-button v-else sizesmall clickhandleReconnect :disabledisLoading || !question.trim() 重新连接 /el-button el-button sizesmall clickhandleClear清空回答/el-button /div !-- 错误提示 -- el-alert v-iferror :titleerror.message typeerror :closabletrue closeerror null classerror-alert / !-- 回答展示区域 -- div classanswer-area el-card shadownever template #header div模型回答/div /template div v-ifisLoading message classloading-placeholder el-icon classis-loadingLoading //el-icon 正在思考... /div !-- 关键这里直接绑定 message它会随着流式数据自动更新 -- div classanswer-content{{ message }}/div div v-ifisLoading message classtyping-indicator span classcursor▌/span /div /el-card /div /el-card /div /template script setup langts import { ref, computed } from vue import { Search, Loading } from element-plus/icons-vue import { useSSEStream } from /utils/sseClient // 导入我们封装的 Hook const question ref() // 使用 SSE Hook const { message, isConnected, isLoading, error, connect, close } useSSEStream() // 初始 URL 为空将在提问时动态构造 // 处理提问 const handleAsk () { if (!question.value.trim()) { return } // 构造带参数的 SSE 请求 URL const apiUrl http://localhost:8000/api/stream-answer?question${encodeURIComponent(question.value)} // 先关闭可能存在的旧连接 close() // 使用新的 URL 建立连接 connect() } // 处理手动关闭连接 const handleClose () { close() } // 处理重新连接使用当前问题 const handleReconnect () { if (question.value.trim()) { handleAsk() } } // 清空回答 const handleClear () { message.value } /script style scoped .stream-qa-container { max-width: 800px; margin: 20px auto; padding: 20px; } .input-area { margin-bottom: 20px; } .status-area { margin-bottom: 15px; display: flex; gap: 10px; align-items: center; } .error-alert { margin-bottom: 15px; } .answer-area { margin-top: 20px; } .answer-content { min-height: 200px; line-height: 1.6; white-space: pre-wrap; /* 保留空格和换行 */ word-break: break-word; } .loading-placeholder { color: #909399; display: flex; align-items: center; gap: 8px; } .typing-indicator { display: inline-block; } .cursor { animation: blink 1s infinite; color: #409eff; } keyframes blink { 0%, 100% { opacity: 1; } 50% { opacity: 0; } } /style前端实现的核心细节动态 URL 构造我们没有在useSSEStream初始化时传入固定 URL而是在点击“发送”时将用户输入的问题encodeURIComponent后拼接到 URL 上。这样每次提问都是一个新的、独立的 SSE 连接符合问答场景。连接状态管理UI 上清晰展示了连接状态isConnected并提供了“断开连接”和“重新连接”的按钮让用户有控制感也便于调试。加载与打字机效果通过isLoading和message的组合实现了“正在思考...”的初始加载提示以及在流式输出过程中在末尾添加一个闪烁的光标动画模拟打字机效果极大提升用户体验。错误友好提示使用 Element Plus 的el-alert组件展示错误信息用户可手动关闭。自动滚动优化进阶上面的基础代码在回答很长时新内容会出现在可视区域下方。一个常见的优化是在message更新后自动将回答区域滚动到底部。这可以通过一个watch和操作 DOM 来实现但更优雅的方式是使用一个专门的指令或第三方库如vueuc的useScrollTo。3.4 配置开发服务器代理前端运行在localhost:5173后端在localhost:8000直接请求会触发浏览器的跨域限制。虽然我们后端配置了 CORS但在开发阶段更推荐使用 Vite 的代理功能。这样前端代码里可以直接写相对路径/apiVite 开发服务器会帮你转发到后端避免跨域问题。修改vite.config.tsimport { defineConfig } from vite import vue from vitejs/plugin-vue // https://vitejs.dev/config/ export default defineConfig({ plugins: [vue()], server: { proxy: { // 将 /api 开头的请求代理到后端服务器 /api: { target: http://localhost:8000, changeOrigin: true, // 如果后端接口路径本身就有 /api通常不需要重写 // rewrite: (path) path.replace(/^\/api/, ) } } } })然后前端代码中的请求 URL 就可以简化为const apiUrl /api/stream-answer?question${encodeURIComponent(question.value)}现在分别启动后端和前端# 终端1启动后端 cd stream-qa-backend uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 # 终端2启动前端 cd stream-qa-frontend npm run dev打开浏览器访问http://localhost:5173或 Vite 提示的地址输入问题点击发送你应该就能看到答案像打字一样一个字一个字地流式显示出来了4. 从模拟到真实集成 LLM 与 RAG 的进阶思考跑通骨架只是第一步。我们的mock_llm_stream_generator函数目前只是模拟。在实际项目中你需要在这里集成真正的大语言模型。4.1 集成 OpenAI 或国产大模型 API以 OpenAI 的流式接口为例其他如通义千问、DeepSeek 等类似# 首先安装 openai 库: pip install openai import openai from openai import AsyncOpenAI # 初始化客户端请将 API KEY 放在环境变量中 client AsyncOpenAI(api_keyos.getenv(OPENAI_API_KEY)) async def real_llm_stream_generator(prompt: str) - AsyncGenerator[str, None]: 调用真实的 OpenAI 流式接口。 try: stream await client.chat.completions.create( modelgpt-3.5-turbo, messages[{role: user, content: prompt}], streamTrue, # 关键参数开启流式 max_tokens500, ) async for chunk in stream: # chunk 是一个 ChatCompletionChunk 对象 if chunk.choices and chunk.choices[0].delta.content is not None: token chunk.choices[0].delta.content yield json.dumps({token: token}, ensure_asciiFalse) # 流结束 yield json.dumps({event: end, data: 生成结束}, ensure_asciiFalse) except Exception as e: # 处理 API 调用异常 yield json.dumps({event: error, data: f模型调用失败: {str(e)}}, ensure_asciiFalse)关键点异步客户端使用AsyncOpenAI以配合 FastAPI 的异步框架。streamTrue这是开启流式响应的关键。错误处理一定要用try...except包裹 API 调用并将错误信息通过 SSE 流推送给前端让用户知道发生了什么。4.2 引入 RAG 架构的考量当你的问答需要基于特定知识库如公司文档、产品手册时就需要引入 RAG。这会让后端流程变得复杂检索Retrieval用户提问后首先将问题转换为向量在向量数据库中检索出最相关的文档片段chunks。增强Augmentation将检索到的片段和原始问题一起构造成一个更丰富的提示Prompt交给 LLM。生成GenerationLLM 基于增强后的提示生成答案并以流式形式返回。这个过程中检索和增强步骤通常是同步的、非流式的只有最后的生成步骤是流式的。因此一个常见的优化模式是阶段一快速响应后端收到问题后立即返回一个 SSE 连接确认并开始执行检索和 Prompt 构建。此时前端可以显示“正在检索相关知识...”。阶段二流式生成一旦 Prompt 准备好开始调用 LLM 流式接口并持续向前端推送 token。同时可以将检索到的文档片段来源也一并推送给前端用于展示“引用”或“参考依据”。这要求我们对 SSE 事件进行更精细的设计例如定义不同的事件类型# 在流式生成器中 # 1. 发送检索开始事件 yield json.dumps({event: retrieval_start}, ensure_asciiFalse) # ... 执行检索 ... # 2. 发送检索到的文档片段可选 for doc in retrieved_docs: yield json.dumps({event: reference, data: doc.metadata}, ensure_asciiFalse) # 3. 发送生成开始事件 yield json.dumps({event: generation_start}, ensure_asciiFalse) # 4. 流式生成 token async for token in llm_stream: yield json.dumps({token: token}, ensure_asciiFalse) # 5. 发送结束事件 yield json.dumps({event: end}, ensure_asciiFalse)前端则需要相应地扩展useSSEStreamHook除了累积message还要能处理reference等事件更新不同的 UI 状态。4.3 性能、稳定性与生产化部署当流量增大时这个简单的骨架会面临挑战连接数限制每个 SSE 连接都是一个长期的 HTTP 连接。像 Nginx 这样的反向代理默认对单个工作进程的并发连接数有限制。需要调整worker_connections和keepalive_timeout等参数。超时与重连网络不稳定时SSE 连接可能中断。前端需要实现自动重连机制通常是在EventSource的onerror回调中设置一个延迟后重新调用connect()。后端也需要设置合理的超时时间避免僵尸连接。身份验证与授权上面的例子没有认证。在生产环境中你不能让任何人随意连接你的 SSE 端点。常见的做法是Token 验证前端在连接时将身份验证 Token 放在 URL 的查询参数中如?tokenxxx后端在建立连接前先验证 Token。Session/Cookie如果前端和后端同域可以使用 Cookie 携带会话信息。注意避免将敏感信息放在 URL 中如果必须确保使用 HTTPS。部署分离前端使用 Nginx 或云服务托管静态文件。后端使用 Uvicorn/Gunicorn 搭配 Nginx 反向代理。Nginx 配置中需要特别注意对/api/stream-answer路径的代理设置确保其支持长连接和流式响应通常需要禁用代理缓冲location /api/ { proxy_pass http://backend:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_buffering off; # 关键关闭代理缓冲让数据流直接通过 proxy_cache off; proxy_http_version 1.1; proxy_set_header Connection ; chunked_transfer_encoding off; proxy_read_timeout 300s; # 设置较长的读取超时 }监控与日志记录 SSE 连接的建立、关闭、错误以及每个问题的处理耗时、token 消耗等对于后期性能分析和成本控制至关重要。5. 常见问题排查与调试技巧在开发过程中你肯定会遇到各种问题。这里总结几个我踩过的坑和解决方法问题一前端收不到任何数据或者连接立即关闭。检查后端 CORS 配置确保allow_origins包含了你的前端地址如http://localhost:5173。在浏览器开发者工具的“网络”选项卡中查看 SSE 请求的响应头是否包含Content-Type: text/event-stream以及正确的 CORS 头。检查代理配置如果你用了 Vite 代理确保vite.config.ts中的target地址和端口正确。可以尝试直接访问后端完整 URLhttp://localhost:8000/api/stream-answer?questiontest看是否有数据流。检查后端生成器在mock_llm_stream_generator函数里加print语句或者用日志确认生成器确实在yield数据。确保没有未处理的异常导致生成器提前退出。问题二前端能连接但数据是一次性收到而不是流式的。罪魁祸首通常是代理或网关的缓冲。Nginx、Apache 或者一些云平台的负载均衡器默认会缓冲上游服务器的响应等收齐了再发给客户端。这就是为什么必须设置proxy_buffering off;和proxy_cache off;。检查 FastAPI 响应确保返回的是EventSourceResponse或StreamingResponse而不是普通的JSONResponse。问题三连接一段时间后自动断开。可能是防火墙、负载均衡器或代理的超时设置太短。SSE 是长连接需要调整相关超时参数。例如 Nginx 的proxy_read_timeout可以设大一点如 300 秒。后端没有发送“心跳”事件。一些网络设备会关闭长时间没有数据流动的连接。一个好的实践是在后端生成器中定期发送一个注释行以:开头作为心跳保活例如每 15 秒yield “:ping\n\n”。EventSource会自动忽略注释行但能保持连接活跃。问题四前端页面切换或刷新后旧的连接没有关闭。务必在 Vue 组件的onUnmounted生命周期钩子中调用close()方法。这是我们封装useSSEStream时已经做好的。如果组件被销毁而连接未关闭会导致内存泄漏和服务器资源浪费。调试利器浏览器开发者工具。网络Network标签找到类型为eventsource的请求点击它在“响应Response”或“事件流EventStream”标签页Chrome 有可以实时看到服务器推送过来的原始事件流数据。这是调试 SSE 问题最直接的方式。控制台Console查看EventSource的onerror和onopen事件打印的日志。骨架已经搭好流已打通。但这仅仅是万里长征的第一步。接下来你可以根据实际需求在这个骨架上填充血肉接入真实的 LLM API设计更复杂的 Prompt集成向量数据库实现 RAG加入对话历史管理优化前端 UI 交互考虑分布式部署下的连接管理等等。每一个环节都有新的挑战和最佳实践等待探索。希望这个扎实的起点能让你在构建流式 AI 应用的道路上走得更稳、更快。 FastAPI+Vue实现SSE流式AI问答：从原理到实战部署
1. 项目缘起为什么是 FastAPI Vue SSE最近在折腾一个AI问答项目后台用大模型生成答案前端需要实时、流畅地展示生成过程。最开始图省事直接让前端轮询接口或者等后端生成完整个答案再一次性返回。结果要么是前端疯狂刷请求服务器压力山大要么是用户盯着空白页面干等好几秒体验极差。这让我意识到在这种需要“边想边说”的交互场景里流式传输Streaming不是锦上添花而是雪中送炭。在众多技术方案里我最终锁定了FastAPI Vue SSE这个组合。为什么不是 WebSocket对于问答这种典型的“一问一答”、且主要由服务器向客户端推送数据的场景SSEServer-Sent Events协议更轻量、更简单。它基于 HTTP不需要额外的握手和复杂的连接管理浏览器原生支持后端实现也直观。FastAPI 对异步和流式响应的支持堪称优雅几行代码就能搭起一个高效的流式端点。Vue 作为前端主流框架其响应式系统能完美地将后端推送来的数据碎片实时渲染到页面上整个过程行云流水。这个组合的目标很明确搭建一个最小可行原型跑通从用户提问、后端流式生成、到前端逐字展示的完整链路。这不仅是技术上的验证更是为后续集成更复杂的 RAG检索增强生成、Agent 等能力打下坚实的地基。下面我就把从零开始搭建这个骨架的每一步包括我踩过的坑和总结的技巧毫无保留地分享出来。2. 后端基石用 FastAPI 构建 SSE 流式端点后端是整个流式问答的心脏负责接收问题、调用模型或模拟生成、并以流的形式吐出答案。FastAPI 的异步特性让这一切变得非常高效。2.1 环境准备与依赖安装首先创建一个干净的 Python 虚拟环境是专业开发的好习惯能避免包版本冲突。# 创建项目目录并进入 mkdir stream-qa-backend cd stream-qa-backend # 创建虚拟环境这里使用 venv你也可以用 conda python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate激活虚拟环境后安装核心依赖。除了 FastAPI我们还需要uvicorn作为 ASGI 服务器来运行应用sse-starlette这个库能帮我们更规范地处理 SSE 响应。pip install fastapi uvicorn sse-starlette这里有个小坑sse-starlette的版本。我一开始用了最新版结果和 FastAPI 的某些中间件有兼容性问题事件流无法正常关闭。后来锁定到sse-starlette1.6.5这个版本就非常稳定。所以建议在requirements.txt里直接指定fastapi0.104.1 uvicorn[standard]0.24.0 sse-starlette1.6.52.2 核心应用结构与 SSE 路由实现项目结构保持清晰很重要。我习惯这样组织stream-qa-backend/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用实例和主路由 │ └── api/ │ ├── __init__.py │ └── endpoints/ │ ├── __init__.py │ └── sse_stream.py # 专门的 SSE 流式端点 ├── requirements.txt └── .env # 环境变量如果需要现在我们来编写最核心的 SSE 端点。在app/api/endpoints/sse_stream.py中import asyncio import json from typing import AsyncGenerator from fastapi import APIRouter, Request from sse_starlette.sse import EventSourceResponse router APIRouter() # 模拟一个异步的、流式的文本生成器 async def mock_llm_stream_generator(prompt: str) - AsyncGenerator[str, None]: 模拟大语言模型的流式生成。 在实际项目中这里会替换为调用 OpenAI API、本地 Llama 模型等。 simulated_response f这是针对问题“{prompt}”生成的流式回答。 words list(simulated_response) for word in words: # 模拟每个词或token的生成延迟 await asyncio.sleep(0.05) # 以 SSE 格式要求的 data: 字段返回 yield json.dumps({token: word}, ensure_asciiFalse) # 可选发送一个结束事件 yield json.dumps({event: end, data: 生成结束}, ensure_asciiFalse) router.get(/stream-answer) async def stream_answer(request: Request, question: str): SSE流式问答端点。 客户端通过 GET 请求访问并传递 question 参数。 例如GET /api/stream-answer?question什么是人工智能 async def event_generator(): # 这里可以添加身份验证、请求日志等逻辑 try: async for chunk in mock_llm_stream_generator(question): # 关键SSE 格式要求每行数据以 data: 开头并以两个换行符结束 yield fdata: {chunk}\n\n # 检查客户端是否还连接着防止客户端断开后服务器还在发送数据 if await request.is_disconnected(): print(客户端断开连接) break except asyncio.CancelledError: # 处理任务被取消的情况例如客户端刷新页面 print(流式生成任务被取消) raise # 使用 EventSourceResponse 包装生成器并设置正确的媒体类型 return EventSourceResponse( event_generator(), headers{ Cache-Control: no-cache, Connection: keep-alive, Content-Type: text/event-stream, } )关键点解析与避坑EventSourceResponse的使用sse-starlette提供的这个响应类帮我们处理了 SSE 协议的大部分细节比如自动发送retry事件、正确处理连接关闭等。比自己手动构造响应省心太多。生成器函数async for我们使用async def定义了一个异步生成器函数mock_llm_stream_generator。这是 FastAPI 处理流式响应的核心模式。在实际项目中这个函数内部就是调用真实的 LLM API如 OpenAI、通义千问等并yield每一个返回的 token 或 chunk。请求断开检测if await request.is_disconnected():这行代码至关重要。没有它即使浏览器页面关闭了后端的生成器可能还在运行浪费服务器资源。这个检查确保了连接断开时及时退出循环。SSE 数据格式SSE 协议规定每条消息由若干行组成以两个换行符\n\n结尾。最常见的是data:行。我们yield fdata: {chunk}\n\n就是在构造这个格式。消息体chunk通常是一个 JSON 字符串方便前端解析。2.3 整合应用并配置 CORS接下来在app/main.py中创建 FastAPI 应用实例并挂载我们的路由。由于前端和后端通常是分开部署的跨域必须配置 CORS跨源资源共享。from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.api.endpoints import sse_stream # 创建 FastAPI 应用实例 app FastAPI(title流式问答后端 API, version1.0.0) # 配置 CORS 中间件 # 在生产环境中应严格指定 origins而不是用 * app.add_middleware( CORSMiddleware, allow_origins[*], # 允许所有前端源仅用于开发测试 allow_credentialsTrue, allow_methods[*], # 允许所有方法 allow_headers[*], # 允许所有头 ) # 挂载路由 app.include_router(sse_stream.router, prefix/api, tags[流式问答]) app.get(/) async def root(): return {message: Streaming QA Backend is running.}CORS 配置的注意事项在开发阶段为了方便我们设置了allow_origins[*]。但这在生产环境是极其危险的它会允许任何网站向你的后端发起请求。正确的做法是明确列出你前端应用部署的域名例如[https://your-vue-app.com, http://localhost:5173]。2.4 运行与测试后端服务在项目根目录下使用 Uvicorn 启动服务uvicorn app.main:app --reload --host 0.0.0.0 --port 8000--reload: 代码修改后自动重启仅用于开发。--host 0.0.0.0: 监听所有网络接口方便同一局域网内的设备如手机访问。--port 8000: 指定端口。启动后访问http://localhost:8000/docs就能看到自动生成的交互式 API 文档Swagger UI。你可以直接在那里测试/api/stream-answer接口。但更直观的测试方法是使用curl命令它能直接展示原始的 SSE 流curl -N http://localhost:8000/api/stream-answer?question你好世界如果看到一行行以data:开头的 JSON 数据流式地输出到终端恭喜你后端 SSE 服务已经成功跑起来了3. 前端呈现Vue 3 连接与消费 SSE 流后端流已经就绪现在需要前端来连接这个流并实时地将数据渲染到页面上。Vue 3 的组合式 API 让这个过程变得非常清晰。3.1 创建 Vue 项目与安装依赖使用官方的 Vite 工具创建 Vue 3 项目选择 TypeScript 和更现代的架构。npm create vuelatest stream-qa-frontend # 按照提示选择Vue Router可选 Pinia可选 ESLint等根据喜好。 cd stream-qa-frontend npm install为了更好的 UI 体验我们可以安装一个组件库。这里以 Element Plus 为例npm install element-plus element-plus/icons-vue然后在main.ts中全局引入import { createApp } from vue import App from ./App.vue import ElementPlus from element-plus import element-plus/dist/index.css const app createApp(App) app.use(ElementPlus) app.mount(#app)3.2 构建 SSE 连接管理模块我们不建议把 SSE 连接逻辑直接写在组件里而是封装成一个可复用的、易于管理状态和生命周期的模块。在src目录下创建utils/sseClient.ts// src/utils/sseClient.ts import { ref, onUnmounted } from vue interface SSEMessage { token?: string event?: string data?: any } export function useSSEStream(url: string) { const message ref() // 累积的完整消息 const isConnected ref(false) const isLoading ref(false) const error refError | null(null) let eventSource: EventSource | null null const connect () { if (eventSource) { console.warn(SSE 连接已存在正在关闭旧连接) close() } isLoading.value true error.value null message.value // 开始新的会话清空历史消息 try { // 创建 EventSource 实例浏览器原生 API eventSource new EventSource(url) isConnected.value true // 监听 message 事件标准数据事件 eventSource.onmessage (event) { isLoading.value false try { const data: SSEMessage JSON.parse(event.data) // 处理 token 类型的消息 if (data.token ! undefined) { message.value data.token // 逐 token 累加 } // 处理自定义事件例如结束事件 if (data.event end) { console.log(流式传输结束:, data.data) close() // 收到结束事件后可以主动关闭连接 } } catch (e) { console.error(解析 SSE 消息失败:, e, event.data) } } // 监听 error 事件 eventSource.onerror (err) { console.error(SSE 连接错误:, err) error.value new Error(SSE 连接异常或已断开) isLoading.value false isConnected.value false close() // 发生错误时清理资源 } // 监听 open 事件 eventSource.onopen () { console.log(SSE 连接已建立) isLoading.value false } } catch (err) { error.value err as Error isLoading.value false isConnected.value false } } const close () { if (eventSource) { eventSource.close() eventSource null isConnected.value false console.log(SSE 连接已关闭) } } // 组件卸载时自动关闭连接防止内存泄漏 onUnmounted(() { close() }) // 返回响应式状态和方法 return { message, isConnected, isLoading, error, connect, close } }这个模块的设计精妙之处响应式状态使用 Vue 的ref管理连接状态、加载状态、错误信息和累积的消息。这些状态变化会自动触发视图更新。生命周期管理通过onUnmounted钩子确保在组件销毁时自动关闭 EventSource 连接这是避免内存泄漏的关键。错误处理完善地处理了网络错误、解析错误并将错误信息暴露给组件便于 UI 展示。可复用性在任何 Vue 组件中只需调用useSSEStream(url)即可获得一套完整的 SSE 连接能力。3.3 实现问答页面组件现在我们来创建一个使用上述 Hook 的页面组件。在src/views或src/components下创建StreamQA.vuetemplate div classstream-qa-container el-card classbox-card template #header div classcard-header spanSSE 流式问答演示/span /div /template !-- 输入区域 -- div classinput-area el-input v-modelquestion placeholder请输入您的问题... :disabledisLoading keyup.enterhandleAsk template #append el-button :iconSearch clickhandleAsk :loadingisLoading :disabled!question.trim() 发送 /el-button /template /el-input /div !-- 状态与操作区域 -- div classstatus-area el-tag :typeisConnected ? success : info {{ isConnected ? 已连接 : 未连接 }} /el-tag el-button v-ifisConnected sizesmall clickhandleClose :disabledisLoading 断开连接 /el-button el-button v-else sizesmall clickhandleReconnect :disabledisLoading || !question.trim() 重新连接 /el-button el-button sizesmall clickhandleClear清空回答/el-button /div !-- 错误提示 -- el-alert v-iferror :titleerror.message typeerror :closabletrue closeerror null classerror-alert / !-- 回答展示区域 -- div classanswer-area el-card shadownever template #header div模型回答/div /template div v-ifisLoading message classloading-placeholder el-icon classis-loadingLoading //el-icon 正在思考... /div !-- 关键这里直接绑定 message它会随着流式数据自动更新 -- div classanswer-content{{ message }}/div div v-ifisLoading message classtyping-indicator span classcursor▌/span /div /el-card /div /el-card /div /template script setup langts import { ref, computed } from vue import { Search, Loading } from element-plus/icons-vue import { useSSEStream } from /utils/sseClient // 导入我们封装的 Hook const question ref() // 使用 SSE Hook const { message, isConnected, isLoading, error, connect, close } useSSEStream() // 初始 URL 为空将在提问时动态构造 // 处理提问 const handleAsk () { if (!question.value.trim()) { return } // 构造带参数的 SSE 请求 URL const apiUrl http://localhost:8000/api/stream-answer?question${encodeURIComponent(question.value)} // 先关闭可能存在的旧连接 close() // 使用新的 URL 建立连接 connect() } // 处理手动关闭连接 const handleClose () { close() } // 处理重新连接使用当前问题 const handleReconnect () { if (question.value.trim()) { handleAsk() } } // 清空回答 const handleClear () { message.value } /script style scoped .stream-qa-container { max-width: 800px; margin: 20px auto; padding: 20px; } .input-area { margin-bottom: 20px; } .status-area { margin-bottom: 15px; display: flex; gap: 10px; align-items: center; } .error-alert { margin-bottom: 15px; } .answer-area { margin-top: 20px; } .answer-content { min-height: 200px; line-height: 1.6; white-space: pre-wrap; /* 保留空格和换行 */ word-break: break-word; } .loading-placeholder { color: #909399; display: flex; align-items: center; gap: 8px; } .typing-indicator { display: inline-block; } .cursor { animation: blink 1s infinite; color: #409eff; } keyframes blink { 0%, 100% { opacity: 1; } 50% { opacity: 0; } } /style前端实现的核心细节动态 URL 构造我们没有在useSSEStream初始化时传入固定 URL而是在点击“发送”时将用户输入的问题encodeURIComponent后拼接到 URL 上。这样每次提问都是一个新的、独立的 SSE 连接符合问答场景。连接状态管理UI 上清晰展示了连接状态isConnected并提供了“断开连接”和“重新连接”的按钮让用户有控制感也便于调试。加载与打字机效果通过isLoading和message的组合实现了“正在思考...”的初始加载提示以及在流式输出过程中在末尾添加一个闪烁的光标动画模拟打字机效果极大提升用户体验。错误友好提示使用 Element Plus 的el-alert组件展示错误信息用户可手动关闭。自动滚动优化进阶上面的基础代码在回答很长时新内容会出现在可视区域下方。一个常见的优化是在message更新后自动将回答区域滚动到底部。这可以通过一个watch和操作 DOM 来实现但更优雅的方式是使用一个专门的指令或第三方库如vueuc的useScrollTo。3.4 配置开发服务器代理前端运行在localhost:5173后端在localhost:8000直接请求会触发浏览器的跨域限制。虽然我们后端配置了 CORS但在开发阶段更推荐使用 Vite 的代理功能。这样前端代码里可以直接写相对路径/apiVite 开发服务器会帮你转发到后端避免跨域问题。修改vite.config.tsimport { defineConfig } from vite import vue from vitejs/plugin-vue // https://vitejs.dev/config/ export default defineConfig({ plugins: [vue()], server: { proxy: { // 将 /api 开头的请求代理到后端服务器 /api: { target: http://localhost:8000, changeOrigin: true, // 如果后端接口路径本身就有 /api通常不需要重写 // rewrite: (path) path.replace(/^\/api/, ) } } } })然后前端代码中的请求 URL 就可以简化为const apiUrl /api/stream-answer?question${encodeURIComponent(question.value)}现在分别启动后端和前端# 终端1启动后端 cd stream-qa-backend uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 # 终端2启动前端 cd stream-qa-frontend npm run dev打开浏览器访问http://localhost:5173或 Vite 提示的地址输入问题点击发送你应该就能看到答案像打字一样一个字一个字地流式显示出来了4. 从模拟到真实集成 LLM 与 RAG 的进阶思考跑通骨架只是第一步。我们的mock_llm_stream_generator函数目前只是模拟。在实际项目中你需要在这里集成真正的大语言模型。4.1 集成 OpenAI 或国产大模型 API以 OpenAI 的流式接口为例其他如通义千问、DeepSeek 等类似# 首先安装 openai 库: pip install openai import openai from openai import AsyncOpenAI # 初始化客户端请将 API KEY 放在环境变量中 client AsyncOpenAI(api_keyos.getenv(OPENAI_API_KEY)) async def real_llm_stream_generator(prompt: str) - AsyncGenerator[str, None]: 调用真实的 OpenAI 流式接口。 try: stream await client.chat.completions.create( modelgpt-3.5-turbo, messages[{role: user, content: prompt}], streamTrue, # 关键参数开启流式 max_tokens500, ) async for chunk in stream: # chunk 是一个 ChatCompletionChunk 对象 if chunk.choices and chunk.choices[0].delta.content is not None: token chunk.choices[0].delta.content yield json.dumps({token: token}, ensure_asciiFalse) # 流结束 yield json.dumps({event: end, data: 生成结束}, ensure_asciiFalse) except Exception as e: # 处理 API 调用异常 yield json.dumps({event: error, data: f模型调用失败: {str(e)}}, ensure_asciiFalse)关键点异步客户端使用AsyncOpenAI以配合 FastAPI 的异步框架。streamTrue这是开启流式响应的关键。错误处理一定要用try...except包裹 API 调用并将错误信息通过 SSE 流推送给前端让用户知道发生了什么。4.2 引入 RAG 架构的考量当你的问答需要基于特定知识库如公司文档、产品手册时就需要引入 RAG。这会让后端流程变得复杂检索Retrieval用户提问后首先将问题转换为向量在向量数据库中检索出最相关的文档片段chunks。增强Augmentation将检索到的片段和原始问题一起构造成一个更丰富的提示Prompt交给 LLM。生成GenerationLLM 基于增强后的提示生成答案并以流式形式返回。这个过程中检索和增强步骤通常是同步的、非流式的只有最后的生成步骤是流式的。因此一个常见的优化模式是阶段一快速响应后端收到问题后立即返回一个 SSE 连接确认并开始执行检索和 Prompt 构建。此时前端可以显示“正在检索相关知识...”。阶段二流式生成一旦 Prompt 准备好开始调用 LLM 流式接口并持续向前端推送 token。同时可以将检索到的文档片段来源也一并推送给前端用于展示“引用”或“参考依据”。这要求我们对 SSE 事件进行更精细的设计例如定义不同的事件类型# 在流式生成器中 # 1. 发送检索开始事件 yield json.dumps({event: retrieval_start}, ensure_asciiFalse) # ... 执行检索 ... # 2. 发送检索到的文档片段可选 for doc in retrieved_docs: yield json.dumps({event: reference, data: doc.metadata}, ensure_asciiFalse) # 3. 发送生成开始事件 yield json.dumps({event: generation_start}, ensure_asciiFalse) # 4. 流式生成 token async for token in llm_stream: yield json.dumps({token: token}, ensure_asciiFalse) # 5. 发送结束事件 yield json.dumps({event: end}, ensure_asciiFalse)前端则需要相应地扩展useSSEStreamHook除了累积message还要能处理reference等事件更新不同的 UI 状态。4.3 性能、稳定性与生产化部署当流量增大时这个简单的骨架会面临挑战连接数限制每个 SSE 连接都是一个长期的 HTTP 连接。像 Nginx 这样的反向代理默认对单个工作进程的并发连接数有限制。需要调整worker_connections和keepalive_timeout等参数。超时与重连网络不稳定时SSE 连接可能中断。前端需要实现自动重连机制通常是在EventSource的onerror回调中设置一个延迟后重新调用connect()。后端也需要设置合理的超时时间避免僵尸连接。身份验证与授权上面的例子没有认证。在生产环境中你不能让任何人随意连接你的 SSE 端点。常见的做法是Token 验证前端在连接时将身份验证 Token 放在 URL 的查询参数中如?tokenxxx后端在建立连接前先验证 Token。Session/Cookie如果前端和后端同域可以使用 Cookie 携带会话信息。注意避免将敏感信息放在 URL 中如果必须确保使用 HTTPS。部署分离前端使用 Nginx 或云服务托管静态文件。后端使用 Uvicorn/Gunicorn 搭配 Nginx 反向代理。Nginx 配置中需要特别注意对/api/stream-answer路径的代理设置确保其支持长连接和流式响应通常需要禁用代理缓冲location /api/ { proxy_pass http://backend:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_buffering off; # 关键关闭代理缓冲让数据流直接通过 proxy_cache off; proxy_http_version 1.1; proxy_set_header Connection ; chunked_transfer_encoding off; proxy_read_timeout 300s; # 设置较长的读取超时 }监控与日志记录 SSE 连接的建立、关闭、错误以及每个问题的处理耗时、token 消耗等对于后期性能分析和成本控制至关重要。5. 常见问题排查与调试技巧在开发过程中你肯定会遇到各种问题。这里总结几个我踩过的坑和解决方法问题一前端收不到任何数据或者连接立即关闭。检查后端 CORS 配置确保allow_origins包含了你的前端地址如http://localhost:5173。在浏览器开发者工具的“网络”选项卡中查看 SSE 请求的响应头是否包含Content-Type: text/event-stream以及正确的 CORS 头。检查代理配置如果你用了 Vite 代理确保vite.config.ts中的target地址和端口正确。可以尝试直接访问后端完整 URLhttp://localhost:8000/api/stream-answer?questiontest看是否有数据流。检查后端生成器在mock_llm_stream_generator函数里加print语句或者用日志确认生成器确实在yield数据。确保没有未处理的异常导致生成器提前退出。问题二前端能连接但数据是一次性收到而不是流式的。罪魁祸首通常是代理或网关的缓冲。Nginx、Apache 或者一些云平台的负载均衡器默认会缓冲上游服务器的响应等收齐了再发给客户端。这就是为什么必须设置proxy_buffering off;和proxy_cache off;。检查 FastAPI 响应确保返回的是EventSourceResponse或StreamingResponse而不是普通的JSONResponse。问题三连接一段时间后自动断开。可能是防火墙、负载均衡器或代理的超时设置太短。SSE 是长连接需要调整相关超时参数。例如 Nginx 的proxy_read_timeout可以设大一点如 300 秒。后端没有发送“心跳”事件。一些网络设备会关闭长时间没有数据流动的连接。一个好的实践是在后端生成器中定期发送一个注释行以:开头作为心跳保活例如每 15 秒yield “:ping\n\n”。EventSource会自动忽略注释行但能保持连接活跃。问题四前端页面切换或刷新后旧的连接没有关闭。务必在 Vue 组件的onUnmounted生命周期钩子中调用close()方法。这是我们封装useSSEStream时已经做好的。如果组件被销毁而连接未关闭会导致内存泄漏和服务器资源浪费。调试利器浏览器开发者工具。网络Network标签找到类型为eventsource的请求点击它在“响应Response”或“事件流EventStream”标签页Chrome 有可以实时看到服务器推送过来的原始事件流数据。这是调试 SSE 问题最直接的方式。控制台Console查看EventSource的onerror和onopen事件打印的日志。骨架已经搭好流已打通。但这仅仅是万里长征的第一步。接下来你可以根据实际需求在这个骨架上填充血肉接入真实的 LLM API设计更复杂的 Prompt集成向量数据库实现 RAG加入对话历史管理优化前端 UI 交互考虑分布式部署下的连接管理等等。每一个环节都有新的挑战和最佳实践等待探索。希望这个扎实的起点能让你在构建流式 AI 应用的道路上走得更稳、更快。