引言伴随着大语言模型LLM的爆发智能 Agent生成式代理正逐步替代传统的规则业务系统。然而大模型的推理Generation往往需要数秒甚至数十秒的时间。如果采用传统的 HTTP 一问一答的同步等待模式前端用户将面临长时间的空白等待产品体验会急剧下降。为了解决这一痛点流式传输Streaming Response成为了现代 AI 应用的核心标配。通过将大模型生成的每一个 Token字词以流的形式实时推送到前端使用户能够瞬间看到响应从而将“首字延迟TTFT”降至毫秒级。本文将带你深入剖析如何构建一个百万并发下的智能 Agent 异步流式响应系统。一、 技术选型WebSocket vs SSEServer-Sent Events在流式推水的场景下业界通常有两种主流的单向/双向长连接通信方案维度Server-Sent Events (SSE)WebSocket协议基础基于常规的 HTTP 协议轻量级。独立的 TCP 协议属于应用层全双工。数据方向服务端向前端单向推送。前后端双向全双工实时通信。断线重连HTTP 协议原生支持自动重连。需要在前端手动编写心跳与重连逻辑。穿透能力能轻松穿透多数企业级网关与防火墙。部分严苛网络下可能会被防火墙拦截。架构建议如果你的 Agent 系统只是“用户发送一段话模型持续吐出回复文本、JSON”强烈建议首选 SSE其研发和运维成本极低。只有当你的系统需要高频双向交互如实时语音对讲、多人协同画布时才考虑引入 WebSocket。二、 异步流式 Agent 系统的整体架构防线在高并发场景下直接让 Web 服务打通大模型 API 会面临严重的线程阻塞与连接数耗尽问题。因此我们需要在架构上构建一层异步缓冲防线流量拦截与限流在 API 网关层如 Nginx / Kong对非法的流式请求进行限流防止恶意刷 Token 攻击。异步队列解耦用户发起复杂的 Agent 调度任务时Web 端先将任务写入 Redis 队列或 Kafka并立即向前端返回一个Task_ID。流式消费与吐出后端专属的 Agent Worker 消费队列中的任务调用大模型并通过 SSE 管道实时向前端泵出数据。三、 实战基于 Python FastAPI OAI 的高并发流式后端Python 的 FastAPI 凭借原生对async/await的完美支持是目前开发大模型微服务的最优解。以下是一个基于 FastAPI 编写的生产级 SSE 流式输出代码模板Pythonimport asyncio from fastapi import FastAPI, HTTPException from fastapi.responses import StreamingResponse from fastapi.middleware.cors import CORSMiddleware from openai import AsyncOpenAI import json app FastAPI(titleAgent Streaming Service) # 允许跨域生产环境请收紧配置 app.add_middleware( CORSMiddleware, allow_origins[*], allow_credentialsTrue, allow_methods[*], allow_headers[*], ) # 初始化异步大模型客户端 client AsyncOpenAI(api_keyyour_actual_api_key_here) async def event_generator(prompt: str): 异步生成器用于逐步获取大模型Token并通过SSE格式输出 try: # 调用大模型开启 streamTrue response await client.chat.completions.create( modelgpt-4o-mini, # 示例模型 messages[{role: user, content: prompt}], streamTrue ) async for chunk in response: # 提取增量内容 content chunk.choices[0].delta.content if content: # 严格按照 SSE 规范格式 data: {content}\n\n payload json.dumps({text: content, status: generating}) yield fdata: {payload}\n\n # 传输流结束标志 end_payload json.dumps({text: , status: completed}) yield fdata: {end_payload}\n\n except Exception as e: # 异常捕获并通知前端 err_payload json.dumps({text: fError: {str(e)}, status: error}) yield fdata: {err_payload}\n\n app.get(/api/v1/agent/chat) async def chat_stream(prompt: str): if not prompt: raise HTTPException(status_code400, detailPrompt cannot be empty) # 返回 StreamingResponse并指定媒体类型为 text/event-stream return StreamingResponse( event_generator(prompt), media_typetext/event-stream, headers{Cache-Control: no-cache, Connection: keep-alive} ) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)四、 前端全链路打通现代前端流式解析与 Markdown 渲染在前端如 Vue3 / React我们不能使用常规的axios异步等待而必须使用fetch结合底层ReadableStream或者开源库microsoft/fetch-event-source。以下是原生 TypeScript 解析流式响应的核心算法TypeScriptasync function fetchAgentStream(prompt: string, onChunk: (text: string) void) { const url http://localhost:8000/api/v1/agent/chat?prompt${encodeURIComponent(prompt)}; try { const response await fetch(url); if (!response.body) throw new Error(ReadableStream not supported.); const reader response.body.getReader(); const decoder new TextDecoder(utf-8); let buffer ; while (true) { const { value, done } await reader.read(); if (done) break; // 解码二进制流 buffer decoder.decode(value, { stream: true }); const lines buffer.split(\n\n); // 保留最后一个可能未接收完整的行 buffer lines.pop() || ; for (const line of lines) { if (line.startsWith(data: )) { const jsonStr line.replace(data: , ).trim(); try { const data JSON.parse(jsonStr); if (data.status generating) { onChunk(data.text); // 将解析出的字词实时回调给界面渲染 } } catch (e) { console.error(Parse line error:, e); } } } } } catch (error) { console.error(Stream request failed:, error); } }五、 全链路优化核心要点与避坑指南禁用网关级 Buffering 缓存如果你使用了 Nginx必须要显式关闭代理缓存proxy_buffering off;并且在后端响应头中加入X-Accel-Buffering: no。否则Nginx 会默默把大模型吐出来的流攒到一定字节后才一次性发给前端流式响应会直接退化成同步等待。多机分布式下的 Session 状态由于 SSE 是长连接如果前端在进行流式交互的同时还伴有其他异步操作如中止生成、多轮对话切换必须引入分布式 Redis Lock 来保证在多实例环境下该用户的请求路由在同一个并发上下文中防止上下文死锁。六、 结语智能 Agent 的兴起颠覆了传统的后端设计。流式响应系统不再是简单的“快”或“慢”的问题它深刻影响着 AI 应用的用户留存率。通过前后端规范的 SSE 管道搭建以及网关的细节调优你也可以为你的大模型应用穿上一层高性能的“防线”。写在最后技术进阶资源分享搭建高并发的大模型应用和 Agent 编排系统除了流式传输还涉及到诸如VectorDB向量数据库索引调优、多 Agent 异步编排工具、以及私有化大模型的量化部署等前沿领域。我在日常跟进开源项目的过程中整理、魔改并打包了一系列高并发 Agent 部署模版、前端骨架工程组件以及大模型反向代理代理调优脚本。如果你正在转型 AI 全栈或者对微服务高并发技术感兴趣欢迎移步至我的个人主页/博客公告或者在评论区留言互动加入我们的技术大牛社群共同获取更多免费业内独家硬核技术资料与打包工具我们顶峰相见
