用户的耐心比金子还贵。当你盯着屏幕上那个旋转的加载圈，等待大模型吐出第一个字的时候，每一秒的延迟dou在消耗用户的信任。这就是为什么“流式传输”不再是一个可选项，而是现代 AI 应用的标配。但说实话，要真正把这件事Zuo好，从用户的点击到屏幕上文字的跳动，这中间的链路远比你想象的要复杂。

今天我们不想只谈肤浅的“打字机效果”。我们要深入到底层，去拆解一套完整的、生产级的 AI Streaming 架构。这不仅仅是加一个 `stream: true` 参数那么简单，这是一场从协议选型、服务端管道、客户端渲染到基础设施优化的全方位战役。

在动手写第一行代码之前，我们得先选好武器。大部分时候，我们不需要杀鸡用牛刀。

对于典型的单轮对话场景，fetch API 配合 ReadableStream Yi经足够强大了。基于 HTTP 的 Server-Sent Events 协议简单、单向，天然契合浏览器环境，而且还Neng穿透大多数代理和防火墙。这就像是开着一辆自动挡的轿车去上班，平稳且省心。

但是当你的场景变得复杂——比如你需要多个 Agent 同时在后台思考，或者你需要客户端和服务端进行频繁的双向通信时SSE 可Neng就显得力不从心了。这时候，WebSocket 才是你的重型坦克。它虽然建立连接的开销稍大，但一旦连接建立，双向通信的低延迟特性就Neng带来极致的体验。

所以别一上来就上 WebSocket。先问问自己：我真的需要全双工通信吗？Ru果不需要，老老实实用 SSE，你的架构会geng简洁，维护成本也geng低。

hen多人误以为服务端只是个“传声筒”，把 AI 厂商的 API 结果转发给前端就完事了。大错特错。你的服务端必须是一个精密的“流式处理工厂”。

想象一下数据流像水流一样经过你的服务器：AI Provider → → → → → 客户端。每一个方括号dou是一个中间件环节，它们必须对流式数据进行逐块处理，而不是等全部接收完再处理。

我们来kan一段基于 FastAPI 的实现示例。这里我们不仅Zuo了流式转发，还加入了内容过滤和性Neng指标收集：

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from openai import OpenAI
import json
import time
import os
app = FastAPI
# 初始化客户端，这里以 DeepSeek 为例
client = OpenAI(
    base_url="https://api.deepseek.com",
    api_key=os.getenv,
)
# 中间件：内容安全过滤
def content_filter -> str | None:
    # 实际业务中这里可Neng接geng复杂的审核模型
    sensitive_words = 
    for word in sensitive_words:
        if word in content:
            return None  # 发现敏感词，拦截
    return content
# 中间件：流式指标收集器
class StreamMetrics:
    def __init__:
        self.first_token_time = None
        self.total_tokens = 0
        self.start_time = time.time
    def on_token:
        if self.first_token_time is None:
            # 记录首字延迟
            self.first_token_time = time.time - self.start_time
        self.total_tokens += 1
    def summary -> dict:
        return {
            "first_token_latency_ms": round * 1000),
            "total_tokens": self.total_tokens,
            "total_time_ms": round - self.start_time) * 1000),
        }
@app.post
async def chat_stream:
    body = await request.json
    messages = body.get
    session_id = body.get
    metrics = StreamMetrics
    def generate:
        # 调用上游大模型
        stream = client.chat.completions.create(
            model="deepseek-chat",
            messages=messages,
            stream=True,
        )
        try:
            for chunk in stream:
                content = chunk.choices.delta.content
                if not content:
                    continue
                metrics.on_token
                # 1. 过滤层
                filtered = content_filter
                if filtered is None:
                    filtered = "***" # 替换敏感内容
                # 2. 格式化层：构造 SSE 数据包
                data = json.dumps({
                    "choices": ,
                    "metrics": {
                        "tokens": metrics.total_tokens,
                    },
                })
                yield f"data: {data}

"
        except GeneratorExit:
            # 客户端断开，及时关闭上游连接
            stream.close
        finally:
            # 流结束后的清理与上报
            summary = metrics.summary
            # 这里Ke以异步发送到监控系统
            # async_report_metrics 
            yield f"data: {json.dumps})}

"
            yield "data: 

"
    return StreamingResponse(
        generate,
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",  # 关键：告诉 Nginx 别搞小动作
        },
    )

kan到了吗？每一个 Token 的产生，dou经过了过滤和统计。这才是合格的流式架构。特别是那个 X-Accel-Buffering: no 响应头，Ru果你前面挂着 Nginx，忘了这行代码，你的流式体验可Neng会瞬间变成“等待下载”。

数据到了客户端，战斗才刚刚开始。Ru果 AI 回复的内容hen长，比如生成一篇几千字的技术文档，前端直接暴力渲染会导致页面卡顿，甚至浏览器假死。

当消息列表变得hen长时DOM 节点的数量会爆炸。这时候，虚拟滚动就是救命稻草。它的核心思想hen简单：只渲染用户当前Nengkan到的那几条消息，其他的统统“虚拟化”。

利用像 @tanstack/react-virtual 这样的库，我们Ke以轻松实现这一点：

import { useVirtualizer } from '@tanstack/react-virtual';
function VirtualMessageList {
  const parentRef = useRef;
  // 定义虚拟化逻辑
  const virtualizer = useVirtualizer({
    count: messages.length,
    getScrollElement:  => parentRef.current,
    estimateSize:  => estimateMessageHeight,
    overscan: 5, // 预渲染上下各5条，保证滚动流畅
  });
  return (
    

      

        {virtualizer.getVirtualItems.map => (
          

            
          
        ))}
      
    
  );
}

有时候，数据来得太快，浏览器渲染跟不上。Ru果每个 Token dou触发一次重排，CPU 占用率会飙升。我们Ke以利用 requestAnimationFrame 来Zuo一个天然的节流器，把渲染频率锁定在屏幕刷新率。

class ThrottledRenderer {
  private buffer = '';
  private rendered = '';
  private frameId: number | null = null;
  private container: HTMLElement;
  constructor {
    this.container = container;
  }
  append {
    this.buffer += text;
    this.scheduleRender;
  }
  private scheduleRender {
    if  return; // Yi经在队列里了别重复加
    this.frameId = requestAnimationFrame => {
      this.frameId = null;
      if  return; // 没变化就不渲染
      // 真正的 DOM 操作
      this.rendered = this.buffer;
      this.container.innerText = this.rendered;
      this.scrollToBottom;
    });
  }
  finish {
    if  {
      cancelAnimationFrame;
      this.scheduleRender; // 强制渲染Zui后一帧
    }
  }
}

基础流式只是入门。真正的挑战在于复杂场景下的稳定性。

当你的系统里有多个 Agent 同时在工作，Ru果用 HTTP 轮询或者多个 SSE 连接，开销会hen大。这时候，WebSocket 的多路复用特性就派上用场了。

我们Ke以给每个 Agent 分配一个 channelId，在同一个 WebSocket 连接里传输不同频道的消息：

// 客户端简单的频道分发逻辑
const channels = new Map void>;
ws.onmessage =  => {
  const data = JSON.parse;
  if  {
    // 根据 channelId 找到对应的处理函数
    const handler = channels.get;
    if  handler;
  }
  if  {
    channels.delete;
  }
};
// 注册不同 Agent 的处理器
channels.set => updateSearchAgentUI);
channels.set => updateCodeAgentUI);

移动端网络环境复杂，连接随时可Neng断开。Ru果用户Yi经生成了 500 个字，突然断网，重连后从头开始生成，那体验简直是灾难。

我们需要在服务端维护一个临时的缓存，记录Yi经生成的内容。客户端重连时带上 resume_from 参数，服务端先发缓存里剩下的，再继续生成新的。

// 服务端伪代码：支持断点的流生成器
async function* streamWithResume(
  requestId: string,
  messages: Message,
  resumeFrom: number = 0, 
) {
  let fullContent = streamCache.get || '';
  // 1. 补发丢失的数据
  if  {
    const missed = fullContent.slice;
    yield { type: 'catch-up', content: missed };
  }
  // 2. Ru果还没生成完，继续接上
  if ) {
    // ... 这里接续调用上游 API 的逻辑
  }
  // 3. 正常的流式输出循环
  const stream = await client.chat.completions.create({
    model: 'deepseek-chat',
    messages,
    stream: true,
  });
  for await  {
    const content = chunk.choices?.delta?.content;
    if  continue;
    fullContent += content;
    streamCache.set; // 实时geng新缓存
    yield { type: 'delta', content };
  }
  // 标记完成，设置过期时间防止内存泄漏
  streamCache.set;
  setTimeout => {
    streamCache.delete;
  }, 300_000); 
}

客户端这边也需要配合，实现指数退避的重连机制：

class ResumableStream {
  // ... 初始化代码
  private async connect {
    try {
      const response = await fetch('/api/chat/stream', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          messages,
          request_id: this.requestId,
          resume_from: this.receivedLength, // 告诉服务端我从哪断的
        }),
      });
      // ... 读取流逻辑
    } catch  {
      if  {
        this.retryCount++;
        const delay = Math.min; // 指数退避
        console.warn;
        await sleep;
        await this.connect; // 递归重连
      }
    }
  }
}

Zui后千万别忘了你的反向代理。Nginx 默认会缓冲响应，这对于流式传输来说是致命的。它可Neng会等数据攒够了再一次性发给客户端，导致你的流式效果完全失效。

这是必须要加的 Nginx 配置：

location /api/chat/stream {
    proxy_pass http://backend;
    # 1. 关闭代理缓冲——这是灵魂配置
    proxy_buffering off;
    # 2. 关闭 gzip 压缩
    gzip off;
    # 3. 超时设置要足够长，大模型生成慢
    proxy_read_timeout 300s;
    proxy_send_timeout 300s;
    # 4. SSE 必须的 HTTP 头设置
    proxy_set_header Connection '';
    proxy_http_version 1.1;
    # 5. 禁用分块传输编码的某些优化，保持原样传输
    chunked_transfer_encoding off;
}

从单 Chat 到多 Agent 系统，AI 应用的架构演进之路充满了挑战。我们今天聊的这些——从协议的慎重选择，到服务端中间件的精密设计，再到客户端的虚拟滚动渲染，以及 Nginx 的细节调优——共同构成了一个健壮的端到端流式传输体系。

这不仅仅是代码的堆砌，geng是对用户体验的极致追求。你的项目里流式输出Zuo到了哪一步？是仅仅在前端Zuo了个打字机动画，还是Yi经实现了全链路的流式架构？有没有遇到过 Nginx 缓冲导致流式失效的坑？欢迎在评论区分享你的踩坑经验。

下一篇，我们将进入geng高阶的话题——多 Agent 系统设计，探讨如何让多个 AI 智Neng体协同工作。敬请期待！