流式输出 SSE 实现：让 AI 回答像打字机一样

LLM 生成文本的原理是一次生成一个 token（大概一个词或半个词），然后把 token 拼起来返回。

非流式做法：等所有 token 都生成完，一次性返回完整答案。用户侧感知是：提交问题 → 等待 → 突然出现一大段文字。

流式做法：每生成一个 token 就立刻发给客户端。用户侧感知是：提交问题 → 立刻开始出字 → 逐渐看到完整答案。

对于短问题（答案两三句话），差距不明显。但 AI 回答通常比较长，等待十几秒是常态，这时候流式输出能显著降低用户焦虑感，减少"是不是卡住了"的困惑。另一个好处：流式输出可以让用户在答案生成过程中就判断方向对不对，如果跑偏了可以提前打断，不用等完整回答出来再追问。

1.1 SSE vs WebSocket

SSE流式输出架构——LLM逐Token生成，经SSE事件格式通过FastAPI推送给客户端实时渲染

实现服务器向客户端推送数据，主要有两种方式。

SSE（Server-Sent Events）：

服务器向客户端单向推送的 HTTP 连接。客户端发一个普通 HTTP 请求，服务器不关闭连接，持续往里写数据。

特点：

单向（服务器→客户端）
基于 HTTP，不需要特殊协议
浏览器原生支持 EventSource API
断线自动重连
实现简单

WebSocket：

全双工通信协议，客户端和服务器可以同时互相发消息。

特点：

双向通信
需要协议升级（ws:// 或 wss://）
连接建立后延迟更低
实现相对复杂

对于 AI 回答场景，用户问一个问题，服务器持续输出回答，这是典型的单向推送。SSE 完全够用，没必要引入 WebSocket 的复杂性。除非场景需要客户端在回答过程中打断或修改，才考虑 WebSocket。

1.2 LangChain 的 stream() 方法

LangChain 的所有 LLM 封装都支持流式输出，用 stream() 方法代替 invoke()：

python

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 同步流式
for chunk in llm.stream([HumanMessage(content="解释一下什么是向量数据库")]):
    print(chunk.content, end="", flush=True)

每次循环拿到一个 chunk，chunk.content 是这一批 token 的文本内容（可能一个词，可能几个字符）。

异步版本用 astream()：

python

async def stream_answer(question: str):
    async for chunk in llm.astream([HumanMessage(content=question)]):
        yield chunk.content

Chain 也支持流式：

python

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个技术助手"),
    ("human", "{question}")
])

chain = prompt | llm | StrOutputParser()

# 流式运行 chain
async for chunk in chain.astream({"question": "什么是FastAPI"}):
    print(chunk, end="", flush=True)

1.3 FastAPI + SSE 完整实现

1.3.1 后端代码

python

# streaming_service.py
import asyncio
import json
from typing import AsyncGenerator

from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, streaming=True)


class QuestionRequest(BaseModel):
    question: str


async def generate_stream(question: str) -> AsyncGenerator[str, None]:
    """生成SSE格式的流式数据"""
    total_tokens = 0

    try:
        async for chunk in llm.astream([
            SystemMessage(content="你是一个专业的技术助手，回答要准确简洁。"),
            HumanMessage(content=question)
        ]):
            if chunk.content:
                total_tokens += 1  # 简化计数，实际见后文

                # SSE格式：data: {json}\n\n
                data = json.dumps({
                    "type": "token",
                    "content": chunk.content
                }, ensure_ascii=False)
                yield f"data: {data}\n\n"

                # 模拟打字机延迟（可选，LLM本身的延迟通常够用）
                await asyncio.sleep(0)

        # 发送结束信号
        end_data = json.dumps({
            "type": "done",
            "total_tokens": total_tokens
        })
        yield f"data: {end_data}\n\n"

    except Exception as e:
        error_data = json.dumps({
            "type": "error",
            "message": str(e)
        })
        yield f"data: {error_data}\n\n"


@app.post("/ask/stream")
async def ask_stream(request: QuestionRequest):
    return StreamingResponse(
        generate_stream(request.question),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",  # 禁用Nginx缓冲，必须加
        }
    )


@app.get("/ask/stream")
async def ask_stream_get(question: str):
    """GET版本，方便浏览器EventSource直接用"""
    return StreamingResponse(
        generate_stream(question),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
        }
    )

1.3.2 前端代码

浏览器原生 EventSource API：

html

<!DOCTYPE html>
<html>
<head><title>AI问答流式输出演示</title></head>
<body>
  <input type="text" id="question" placeholder="输入你的问题" style="width:400px">
  <button onclick="askQuestion()">提问</button>
  <div id="answer" style="white-space:pre-wrap; margin-top:20px; font-family:monospace"></div>

<script>
let currentSource = null;

function askQuestion() {
  const question = document.getElementById('question').value.trim();
  if (!question) return;

  // 清空上次回答
  document.getElementById('answer').textContent = '';

  // 关闭上一个连接
  if (currentSource) {
    currentSource.close();
  }

  // EventSource只支持GET请求
  const url = `http://localhost:8000/ask/stream?question=${encodeURIComponent(question)}`;
  currentSource = new EventSource(url);

  currentSource.onmessage = function(event) {
    const data = JSON.parse(event.data);

    if (data.type === 'token') {
      // 追加token到显示区域
      document.getElementById('answer').textContent += data.content;
    } else if (data.type === 'done') {
      console.log('回答完成，token数：', data.total_tokens);
      currentSource.close();
    } else if (data.type === 'error') {
      console.error('错误：', data.message);
      currentSource.close();
    }
  };

  currentSource.onerror = function(event) {
    console.error('SSE连接出错', event);
    currentSource.close();
  };
}
</script>
</body>
</html>

如果前端用 fetch API（支持 POST）：

javascript

async function askWithFetch(question) {
  const response = await fetch('http://localhost:8000/ask/stream', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ question })
  });

  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  const answerEl = document.getElementById('answer');

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

    const text = decoder.decode(value);
    // 解析SSE格式
    const lines = text.split('\n');
    for (const line of lines) {
      if (line.startsWith('data: ')) {
        try {
          const data = JSON.parse(line.slice(6));
          if (data.type === 'token') {
            answerEl.textContent += data.content;
          }
        } catch (e) {
          // 忽略解析错误
        }
      }
    }
  }
}

1.4 处理中途断开

用户可能在回答生成到一半时关闭页面，或者网络中断。服务端需要能感知到这个情况，停止 LLM 调用，避免浪费 token。

FastAPI 通过 request.is_disconnected() 检测客户端是否断开：

python

from fastapi import Request

@app.post("/ask/stream")
async def ask_stream(request: QuestionRequest, http_request: Request):
    async def generate():
        async for chunk in llm.astream([HumanMessage(content=request.question)]):
            # 每次yield前检查客户端是否还连着
            if await http_request.is_disconnected():
                print("客户端已断开，停止生成")
                break

            if chunk.content:
                data = json.dumps({"type": "token", "content": chunk.content})
                yield f"data: {data}\n\n"

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

1.5 流式输出时统计 Token 用量

流式输出时，LLM 不会在 chunk 里返回完整的 token 统计，需要特殊处理。

LangChain 的做法是在最后一个 chunk 里附带 usage 信息：

python

from langchain_core.outputs import LLMResult
from langchain.callbacks.base import BaseCallbackHandler

class TokenCountCallback(BaseCallbackHandler):
    """自定义回调，统计token用量"""
    def __init__(self):
        self.input_tokens = 0
        self.output_tokens = 0

    def on_llm_end(self, response: LLMResult, **kwargs):
        if response.llm_output:
            usage = response.llm_output.get("token_usage", {})
            self.input_tokens = usage.get("prompt_tokens", 0)
            self.output_tokens = usage.get("completion_tokens", 0)


async def generate_with_token_count(question: str):
    callback = TokenCountCallback()
    llm_with_callback = ChatOpenAI(
        model="gpt-4o-mini",
        streaming=True,
        callbacks=[callback]
    )

    async for chunk in llm_with_callback.astream([HumanMessage(content=question)]):
        if chunk.content:
            data = json.dumps({"type": "token", "content": chunk.content})
            yield f"data: {data}\n\n"

    # 流结束后，callback里已经有token统计了
    end_data = json.dumps({
        "type": "done",
        "input_tokens": callback.input_tokens,
        "output_tokens": callback.output_tokens
    })
    yield f"data: {end_data}\n\n"

1.6 实际效果的对比

假设一个问题，LLM 生成回答需要 8 秒，答案 500 字。

非流式：

用户等待时间感知：8 秒空白 → 突然出现 500 字
平均"感知等待"：8 秒
用户焦虑指数：高（不知道有没有在处理）

流式：

用户等待时间感知：0.3 秒后开始出字 → 8 秒内逐渐看到完整答案
平均"感知等待"：0.3 秒
用户焦虑指数：低（看到在处理了）

实际 LLM 处理时间完全一样，但用户体验截然不同。

1.7 小结

流式输出的实现链路：LangChain astream() → FastAPI StreamingResponse → SSE 协议 → 浏览器 EventSource。

每个环节都不复杂，串起来也就一百多行代码。

流式输出不是锦上添花。用户对等待时间非常敏感，流式输出能在不改变 LLM 性能的前提下，让体验好很多。