引言

伴随着大语言模型（LLM）的爆发，智能 Agent（生成式代理）正逐步替代传统的规则业务系统。然而，大模型的推理（Generation）往往需要数秒甚至数十秒的时间。如果采用传统的 HTTP 一问一答的同步等待模式，前端用户将面临长时间的空白等待，产品体验会急剧下降。

为了解决这一痛点，流式传输（Streaming Response） 成为了现代 AI 应用的核心标配。通过将大模型生成的每一个 Token（字词）以流的形式实时推送到前端，使用户能够瞬间看到响应，从而将“首字延迟（TTFT）”降至毫秒级。本文将带你深入剖析如何构建一个百万并发下的智能 Agent 异步流式响应系统。

一、 技术选型：WebSocket vs SSE（Server-Sent Events）

在流式推水的场景下，业界通常有两种主流的单向/双向长连接通信方案：

架构建议：

如果你的 Agent 系统只是“用户发送一段话，模型持续吐出回复（文本、JSON）”，强烈建议首选 SSE，其研发和运维成本极低。只有当你的系统需要高频双向交互（如实时语音对讲、多人协同画布）时，才考虑引入 WebSocket。

二、 异步流式 Agent 系统的整体架构防线

在高并发场景下，直接让 Web 服务打通大模型 API 会面临严重的线程阻塞与连接数耗尽问题。因此，我们需要在架构上构建一层异步缓冲防线：

1. 流量拦截与限流： 在 API 网关层（如 Nginx / Kong）对非法的流式请求进行限流，防止恶意刷 Token 攻击。

2. 异步队列解耦： 用户发起复杂的 Agent 调度任务时，Web 端先将任务写入 Redis 队列或 Kafka，并立即向前端返回一个 Task_ID。

3. 流式消费与吐出： 后端专属的 Agent Worker 消费队列中的任务，调用大模型，并通过 SSE 管道实时向前端泵出数据。

三、 实战：基于 Python FastAPI + OAI 的高并发流式后端

Python 的 FastAPI 凭借原生对 async/await 的完美支持，是目前开发大模型微服务的最优解。以下是一个基于 FastAPI 编写的生产级 SSE 流式输出代码模板：

Python

import 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(title="Agent Streaming Service")

# 允许跨域（生产环境请收紧配置）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 初始化异步大模型客户端
client = AsyncOpenAI(api_key="your_actual_api_key_here")

async def event_generator(prompt: str):
    """异步生成器，用于逐步获取大模型Token并通过SSE格式输出"""
    try:
        # 调用大模型，开启 stream=True
        response = await client.chat.completions.create(
            model="gpt-4o-mini",  # 示例模型
            messages=[{"role": "user", "content": prompt}],
            stream=True
        )
        
        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 f"data: {payload}\n\n"
                
        # 传输流结束标志
        end_payload = json.dumps({"text": "", "status": "completed"})
        yield f"data: {end_payload}\n\n"

    except Exception as e:
        # 异常捕获并通知前端
        err_payload = json.dumps({"text": f"Error: {str(e)}", "status": "error"})
        yield f"data: {err_payload}\n\n"

@app.get("/api/v1/agent/chat")
async def chat_stream(prompt: str):
    if not prompt:
        raise HTTPException(status_code=400, detail="Prompt cannot be empty")
        
    # 返回 StreamingResponse，并指定媒体类型为 text/event-stream
    return StreamingResponse(
        event_generator(prompt), 
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "Connection": "keep-alive"}
    )

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

四、 前端全链路打通：现代前端流式解析与 Markdown 渲染

在前端（如 Vue3 / React），我们不能使用常规的 axios 异步等待，而必须使用 fetch 结合底层 ReadableStream 或者开源库 @microsoft/fetch-event-source。

以下是原生 TypeScript 解析流式响应的核心算法：

TypeScript

async 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);
  }
}

五、 全链路优化核心要点与避坑指南

1. 禁用网关级 Buffering 缓存：

   如果你使用了 Nginx，必须要显式关闭代理缓存（proxy_buffering off;），并且在后端响应头中加入 X-Accel-Buffering: no。否则，Nginx 会默默把大模型吐出来的流攒到一定字节后才一次性发给前端，流式响应会直接退化成同步等待。

2. 多机分布式下的 Session 状态：

   由于 SSE 是长连接，如果前端在进行流式交互的同时还伴有其他异步操作（如中止生成、多轮对话切换），必须引入分布式 Redis Lock 来保证在多实例环境下，该用户的请求路由在同一个并发上下文中，防止上下文死锁。

六、 结语

智能 Agent 的兴起颠覆了传统的后端设计。流式响应系统不再是简单的“快”或“慢”的问题，它深刻影响着 AI 应用的用户留存率。通过前后端规范的 SSE 管道搭建以及网关的细节调优，你也可以为你的大模型应用穿上一层高性能的“防线”。

写在最后（技术进阶资源分享）：

搭建高并发的大模型应用和 Agent 编排系统，除了流式传输，还涉及到诸如 VectorDB（向量数据库）索引调优、多 Agent 异步编排工具、以及私有化大模型的量化部署 等前沿领域。

我在日常跟进开源项目的过程中，整理、魔改并打包了一系列高并发 Agent 部署模版、前端骨架工程组件以及大模型反向代理代理调优脚本。如果你正在转型 AI 全栈，或者对微服务高并发技术感兴趣，欢迎移步至我的个人主页/博客公告，或者在评论区留言互动，加入我们的技术大牛社群，共同获取更多免费业内独家硬核技术资料与打包工具，我们顶峰相见！