【从0搭建AI智能体·4】流式输出全链路：后端 SSE + 前端打字机效果（完整可跑）

📚 本文是《从 0 搭建你的 AI 智能体》专栏第 4 篇。
上一篇：《Function Calling 工具调用实战》　|　专栏目录：点此查看全部

标签：流式输出 SSE FastAPI 前端 打字机效果 大模型 全栈

📌 前言：为什么 ChatGPT 是「一个字一个字蹦」的

你有没有想过，为什么 ChatGPT、Claude 这些产品回答问题时，是像打字机一样一个字一个字往外冒，而不是转半天圈、然后「啪」地甩出一大段？

这不是为了炫技，而是体验的生死线。一个长回答，模型完整生成要好几秒。如果等它全部生成完再一次性返回，用户面对的就是几秒钟的白屏加载——在「3 秒定生死」的产品世界里，这足以让人关掉页面。

流式输出（Streaming） 就是解药：模型边生成、边通过 SSE（Server-Sent Events） 一小块一小块推给前端，前端收到就立刻渲染。首字延迟从「秒级」骤降到「毫秒级」，用户立刻看到反馈，哪怕整段还没说完。

这一篇，我们把流式输出从后端到前端打通成一条完整链路——这也是本专栏第一次亮出「全栈」底牌：别人的教程只到后端打印，我们直接给你一个能在浏览器里跑出打字机效果的完整页面。

💡 本文适合谁：能跑通大模型基础对话、想做出「像样的聊天界面」的开发者。后端 Python + 前端原生 JS，零框架依赖，复制即跑。阅读约 14 分钟。

目录

  1. 一张图看懂流式全链路
  2. SSE 到底是什么：和 WebSocket 的区别
  3. 后端①：最小化流式请求（解析 SSE）
  4. 后端②：用 FastAPI 把流转发给前端
  5. 前端①：用 fetch 接收流并渲染打字机
  6. 前端②：完整可跑的聊天页面（HTML 单文件）
  7. 生产级细节：断线、中断、错误兜底
  8. 常见坑与排查
  9. FAQ
  10. 总结

① 一张图看懂流式全链路

流式输出跨了三段，一张图讲清数据怎么流：

  大模型                你的后端(FastAPI)              浏览器前端
    │                        │                            │
    │  SSE 分块推送           │                            │
    │  data:{"你"}           │                            │
    │───────────────────────►│  透传/加工                  │
    │  data:{"好"}           │  data:{"你"}                │
    │───────────────────────►│───────────────────────────►│ 追加渲染「你」
    │  data:{"！"}           │  data:{"好"}                │
    │───────────────────────►│───────────────────────────►│ 追加渲染「你好」
    │  data:[DONE]           │  data:{"！"}                │
    │───────────────────────►│───────────────────────────►│ 追加渲染「你好！」
    │                        │  data:[DONE]                │
    │                        │───────────────────────────►│ 结束，光标停止

核心就一句话：模型吐一块 → 后端转一块 → 前端渲染一块，全程不等「全部生成完」。

② SSE 到底是什么：和 WebSocket 的区别

流式输出用的是 SSE（Server-Sent Events，服务器推送事件），很多人会和 WebSocket 混淆。区别一表看清：

维度SSEWebSocket
通信方向单向（服务器 → 客户端）双向
协议普通 HTTP独立的 ws 协议
实现复杂度简单，浏览器原生支持较复杂
断线重连浏览器 EventSource 自动重连要自己实现
适用场景AI 逐字输出、日志推送、通知聊天室、协同编辑、游戏

大模型输出是典型的「服务器单向往外推」，SSE 是最合适、最省事的选择，不需要上 WebSocket。

SSE 的数据格式很简单，就是一行行 data: 开头的文本：

data: {"content": "你"}

data: {"content": "好"}

data: [DONE]

💡 每条消息以 data: 开头、以两个换行 \n\n 结尾，这是 SSE 协议规定。后端拼数据时别漏了那个空行。

③ 后端①：最小化流式请求（解析 SSE）

先解决「怎么从大模型那里拿到流」。开启 stream=True 后，上游返回的是 SSE 流，要逐行解析：

import os, json, requests
from dotenv import load_dotenv

load_dotenv()
api_key = os.getenv("AGENT_KEY")
base_url = os.getenv("API_BASE_URL", "https://api.example.com/v1")
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}


def stream_chat(messages):
    """流式请求大模型，实时逐字打印，返回拼接后的完整回复"""
    payload = {"model": "standard-agent-v1", "messages": messages, "stream": True}

    full = []
    # ⚠️ requests 必须传 stream=True，否则会先缓冲整个响应，流式失效
    with requests.post(f"{base_url}/chat/completions",
                       headers=headers, json=payload, stream=True, timeout=60) as resp:
        resp.raise_for_status()
        for line in resp.iter_lines(decode_unicode=True):
            if not line or not line.startswith("data:"):
                continue                       # 跳过空行、心跳注释
            data = line[len("data:"):].strip()
            if data == "[DONE]":
                break                          # 流结束
            try:
                chunk = json.loads(data)
            except json.JSONDecodeError:
                continue                       # 半包容错
            # 增量内容在 delta.content（注意是 delta 不是 message）
            content = chunk["choices"][0].get("delta", {}).get("content")
            if content:
                print(content, end="", flush=True)   # flush 保证实时刷新
                full.append(content)
    print()
    return "".join(full)


if __name__ == "__main__":
    stream_chat([{"role": "user", "content": "用三句话介绍一下杭州"}])

⚠️ 两个高频坑：

  1. requests.post 忘传 stream=True → 它会先缓冲整个响应，流式直接失效；
  2. print 忘加 flush=True → 终端有缓冲，不是逐字而是「一坨坨」蹦。

④ 后端②：用 FastAPI 把流转发给前端

命令行能跑通后，把它变成一个 HTTP 接口，让浏览器能调。关键是用 StreamingResponse——它能边生成边发，不必等全部完成：

# app.py
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
import os, json, requests
from dotenv import load_dotenv

load_dotenv()
api_key = os.getenv("AGENT_KEY")
base_url = os.getenv("API_BASE_URL", "https://api.example.com/v1")
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}

app = FastAPI()
# 允许前端跨域访问（本地开发用 *，生产请收紧到具体域名）
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])


@app.post("/chat/stream")
def chat_stream(body: dict):
    def event_generator():
        payload = {"model": "standard-agent-v1", "messages": body["messages"], "stream": True}
        try:
            with requests.post(f"{base_url}/chat/completions", headers=headers,
                               json=payload, stream=True, timeout=60) as resp:
                resp.raise_for_status()
                for line in resp.iter_lines(decode_unicode=True):
                    if not line or not line.startswith("data:"):
                        continue
                    data = line[len("data:"):].strip()
                    if data == "[DONE]":
                        yield "data: [DONE]\n\n"       # 通知前端结束
                        break
                    try:
                        content = json.loads(data)["choices"][0].get("delta", {}).get("content", "")
                    except json.JSONDecodeError:
                        continue
                    if content:
                        # 重新包装成前端友好的 SSE 格式，注意结尾的两个换行
                        yield f"data: {json.dumps({'content': content})}\n\n"
        except Exception as e:
            yield f"data: {json.dumps({'error': str(e)})}\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")

启动：

pip install fastapi uvicorn requests python-dotenv
uvicorn app:app --reload --port 8000

🎯 为什么要「重新包装」而不是直接透传上游 SSE：上游格式（delta.content、[DONE] 等）是给你后端看的。你把它清洗成前端只关心的 {"content": "..."}，前端解析就极简，后端换模型/换供应商时前端一行都不用改。这是分层设计的价值。

⑤ 前端①：用 fetch 接收流并渲染打字机

前端接收 SSE 有两种方式：EventSource（只支持 GET，自动重连）和 fetch + ReadableStream（支持 POST，更灵活）。因为我们要 POST 发送对话历史，用 fetch 流式读取：

async function streamChat(messages, onToken, onDone, onError) {
  const resp = await fetch("http://localhost:8000/chat/stream", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ messages }),
  });

  const reader = resp.body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    buffer += decoder.decode(value, { stream: true });
    // SSE 以 \n\n 分隔每条消息
    const parts = buffer.split("\n\n");
    buffer = parts.pop();          // 最后一段可能不完整，留到下次

    for (const part of parts) {
      if (!part.startsWith("data:")) continue;
      const data = part.slice(5).trim();
      if (data === "[DONE]") { onDone(); return; }
      try {
        const obj = JSON.parse(data);
        if (obj.error) { onError(obj.error); return; }
        if (obj.content) onToken(obj.content);   // 每收到一个字就回调渲染
      } catch (_) { /* 半包，忽略 */ }
    }
  }
  onDone();
}

💡 buffer 那段是精髓：网络分片不保证「一次正好一条 SSE 消息」，可能半条、也可能两条半。用 buffer 累积、按 \n\n 切分、把不完整的尾巴留到下次——这是流式解析不出错的关键，也是最多人漏掉的地方。

⑥ 前端②：完整可跑的聊天页面（HTML 单文件）

把上面串起来，做成一个保存为 .html 双击就能用的聊天页面（配合第 ④ 步的后端）：

<!DOCTYPE html>
<html lang="zh">
<head>
  <meta charset="UTF-8">
  <title>流式 AI 聊天</title>
  <style>
    body { font-family: system-ui; max-width: 720px; margin: 40px auto; padding: 0 16px; }
    #chat { border: 1px solid #ddd; border-radius: 8px; height: 420px;
            overflow-y: auto; padding: 16px; margin-bottom: 12px; }
    .msg { margin: 8px 0; line-height: 1.6; }
    .user { color: #1677ff; font-weight: 600; }
    .ai { color: #333; white-space: pre-wrap; }        /* 保留换行 */
    .cursor::after { content: "▊"; animation: blink 1s infinite; }
    @keyframes blink { 50% { opacity: 0; } }
    #row { display: flex; gap: 8px; }
    #input { flex: 1; padding: 10px; border: 1px solid #ccc; border-radius: 6px; }
    button { padding: 10px 20px; border: 0; background: #1677ff; color: #fff; border-radius: 6px; }
    button:disabled { background: #aaa; }
  </style>
</head>
<body>
  <h2>🤖 流式 AI 聊天（打字机效果）</h2>
  <div id="chat"></div>
  <div id="row">
    <input id="input" placeholder="说点什么..." autofocus>
    <button id="send">发送</button>
  </div>

  <script>
    const chat = document.getElementById("chat");
    const input = document.getElementById("input");
    const sendBtn = document.getElementById("send");
    const messages = [];   // 对话历史

    function addLine(cls, text) {
      const div = document.createElement("div");
      div.className = "msg";
      div.innerHTML = `<span class="${cls}">${text}</span>`;
      chat.appendChild(div);
      chat.scrollTop = chat.scrollHeight;
      return div.querySelector("span");
    }

    async function send() {
      const text = input.value.trim();
      if (!text) return;
      input.value = "";
      sendBtn.disabled = true;

      addLine("user", "你：" + text);
      messages.push({ role: "user", content: text });

      // AI 回复占位，带闪烁光标
      const aiSpan = addLine("ai cursor", "");
      let reply = "";

      try {
        await streamChat(messages,
          (token) => {                       // 每个字：追加
            reply += token;
            aiSpan.textContent = reply;
            chat.scrollTop = chat.scrollHeight;
          },
          () => {                            // 结束：去掉光标、存历史
            aiSpan.className = "ai";
            messages.push({ role: "assistant", content: reply });
            sendBtn.disabled = false;
          },
          (err) => {                         // 出错
            aiSpan.className = "ai";
            aiSpan.textContent = reply + "\n[出错] " + err;
            sendBtn.disabled = false;
          });
      } catch (e) {
        aiSpan.textContent = "[连接失败] " + e.message;
        sendBtn.disabled = false;
      }
    }

    sendBtn.onclick = send;
    input.onkeydown = (e) => { if (e.key === "Enter") send(); };

    // —— 把第 ⑤ 步的 streamChat 函数粘到这里 ——
    async function streamChat(messages, onToken, onDone, onError) {
      const resp = await fetch("http://localhost:8000/chat/stream", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ messages }),
      });
      const reader = resp.body.getReader();
      const decoder = new TextDecoder();
      let buffer = "";
      while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        buffer += decoder.decode(value, { stream: true });
        const parts = buffer.split("\n\n");
        buffer = parts.pop();
        for (const part of parts) {
          if (!part.startsWith("data:")) continue;
          const data = part.slice(5).trim();
          if (data === "[DONE]") { onDone(); return; }
          try {
            const obj = JSON.parse(data);
            if (obj.error) { onError(obj.error); return; }
            if (obj.content) onToken(obj.content);
          } catch (_) {}
        }
      }
      onDone();
    }
  </script>
</body>
</html>

保存为 index.html，先启动第 ④ 步的后端，再双击打开这个页面，你就得到一个带打字机效果、有历史上下文的聊天界面。这就是「全栈」——读者能直接看到成果，收藏率自然高。

🎨 打字机的「闪烁光标」纯靠 CSS 的 .cursor::after + @keyframes blink 实现，零额外依赖。回复结束时把 cursor 类去掉，光标就停了。

⑦ 生产级细节：断线、中断、错误兜底

Demo 能跑不代表能上线。三个必须处理的场景：

1. 用户中途想「停止生成」 —— 用 AbortController 掐断请求：

let controller = null;

async function streamChat(messages, onToken, onDone) {
  controller = new AbortController();
  const resp = await fetch("/chat/stream", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ messages }),
    signal: controller.signal,        // 绑定中断信号
  });
  // ...读取逻辑同前
}

// 「停止」按钮
function stop() { if (controller) controller.abort(); }

2. 流中途断开 —— 后端 event_generator 已用 try/except 把错误作为 {"error": ...} 推给前端（第 ④ 步），前端 onError 回调兜住，把已收到的半句保留、提示用户重试。

3. 后端要感知「客户端断开」 —— 用户关页面时，继续调用上游模型就是白烧钱。生产中可在生成循环里检查连接状态（如 await request.is_disconnected()），断开就停止拉流。

🎯 成本提醒：流式场景下，「用户关页面」比你想的频繁。若不处理客户端断开，后端会傻傻地把整段生成完——Token 照扣。这是流式上线后最容易被忽视的隐性成本。

⑧ 常见坑与排查

现象原因解决
前端一次性收到全部、没有逐字效果后端/代理开了缓冲后端确认 stream=True；Nginx 加 proxy_buffering off;
命令行不逐字print 没 flush加 flush=True
前端 JSON 解析报错按行切了 SSE，但一条被网络拆成两片用 buffer 累积 + \n\n 切分（第 ⑤ 步）
跨域报错 CORS后端没开 CORSFastAPI 加 CORSMiddleware
中文乱码解码没处理分片TextDecoder 的 {stream:true}

🔍 Nginx 是隐形杀手：本地跑得好好的，一上线过了 Nginx 就变「不逐字」，十有八九是 proxy_buffering 没关。记住 proxy_buffering off; 这一行。

⑨ FAQ

Q1：为什么不用现成的 EventSource？
EventSource 只支持 GET 请求，没法在 body 里 POST 对话历史。要传复杂数据就得用 fetch + ReadableStream，代价是自己处理重连。

Q2：流式和 Function Calling 能一起用吗？
能，但复杂：流式下工具调用的参数是分片到达的，要先把碎片拼完整再执行。建议先用非流式把工具逻辑跑通，再叠流式。

Q3：SSE 需要保持长连接，会不会占用很多资源？
每个流式请求确实占一个连接直到生成结束。高并发时注意后端的连接数/协程上限，FastAPI（异步）比同步框架更扛得住。

Q4：想要 Markdown 渲染（代码高亮、表格）怎么办？
把 aiSpan.textContent = reply 换成用 marked.js 之类库把 reply 渲染成 HTML。注意做 XSS 转义。

Q5：移动端 / 微信里流式失效？
部分环境会强制缓冲。可在响应头加 X-Accel-Buffering: no，并确认中间代理未缓冲。

⑩ 总结

这一篇我们把流式输出从模型到浏览器打通成一条完整链路：

环节关键点
协议选型单向推送用 SSE，别上 WebSocket
后端stream=True + StreamingResponse，清洗成前端友好格式
前端fetch 流式读取 + buffer 拼包 + CSS 打字机光标
生产AbortController 中断、客户端断开检测、Nginx 关缓冲

流式输出是「AI 产品感」的第一道门槛。掌握它，你的智能体就从「能对话」升级到了「用起来舒服」。而这一篇的「后端 + 前端」全链路打法，正是本专栏和满屏「只到后端」教程的分水岭。

下一篇我们讲 Prompt 工程：System Prompt 的 10 个实战模板——同样一个模型，好的 System Prompt 能让效果差出一个量级。

🔜 下一篇预告：《Prompt 工程：System Prompt 的 10 个实战模板》——直接抄的人设/约束/格式模板合集。
👍 如果本文帮到你，点赞 / 收藏 / 关注，追更不迷路。

# 人工智能 # 前端 # 状态模式
Logo
亚马逊云科技技术品牌专区

更多推荐

open-code-review Codex 插件教程：对话式调用评审技能指南

本文是一份面向新手的 **open-code-review Codex 插件教程**，教你在 Codex 中通过对话式方式调用 open-code-review 的 AI 代码评审技能。open-code-review 是阿里巴巴开源的 AI 代码评审工具，采用"确定性流水线 + LLM Agent"混合架构，内置多语言评审规则（NPE、线程安全、XSS、SQL 注入等），可在 Codex 里用一

avatar 亚马逊云科技技术品牌专区

OBS Studio 完整指南：免费开源录屏与直播软件从零上手

想做一场像模像样的直播，或者把屏幕内容录成教程视频，但一想到商用软件的订阅费就有些犹豫？其实你不需要花一分钱。OBS Studio 是一款免费开源的直播与录屏软件，支持 Windows、macOS 和 Linux，采集、混音、编码、推流、录制，一条流水线全包了。## 先搞清楚 OBS Studio 是什么OBS Studio 是一个"场景化"的采集合成工具：你把摄像头、窗口、浏览器、图片、

avatar 亚马逊云科技技术品牌专区

如何用MediaCrawler爬取小红书抖音数据？7大平台爬虫从安装到实战完整指南

MediaCrawler 是一个支持小红书、抖音、快手、B站、微博、百度贴吧、知乎的开源自媒体数据采集工具。它能按关键词、按帖子、按创作者主页三种方式抓取笔记/视频内容、评论甚至二级评论，并把结果存成 JSON、CSV、Excel 或数据库。适合做市场调研、竞品监控、内容选题的运营人员，以及想要现成爬虫框架而不想从零写 JS 逆向的开发者。## 从一次"手动抄数据"说起想象这个场景：你负责

avatar 亚马逊云科技技术品牌专区