用户在等待AI回答时的体验差距是巨大的等待5秒后看到完整回答vs 0.5秒内开始看到文字逐字出现——后者的用户满意度高出70%以上。流式输出Streaming不仅是体验优化更是AI应用架构的核心能力。本文深入解析LLM流式输出的工程实现覆盖后端到前端的完整技术链路。一、流式输出的工作原理LLM生成文本是逐token的自回归过程——模型每次只预测下一个token而非一次生成完整句子。流式输出利用这一特性在每个token生成后立即推送给客户端而非等到全部完成。技术实现依赖Server-Sent EventsSSE或WebSocket客户端 服务器 LLM API | | | |-- HTTP请求 -----------| | | |-- streamTrue请求 -------| |-- data: {text:你} |-- token: 你 ----------| |-- data: {text:好} |-- token: 好 ----------| |-- data: {text:} |-- token: ----------| |-- data: [DONE] |-- [DONE] ---------------|## 二、后端流式实现### 2.1 FastAPI SSE 实现pythonfrom fastapi import FastAPI, Requestfrom fastapi.responses import StreamingResponseimport anthropicimport jsonimport asyncioapp FastAPI()client anthropic.AsyncAnthropic()app.post(/api/chat/stream)async def chat_stream(request: Request): 流式聊天接口SSE格式 body await request.json() messages body.get(messages, []) model body.get(model, claude-opus-4-7) async def generate(): try: async with client.messages.stream( modelmodel, max_tokens2000, messagesmessages ) as stream: async for text in stream.text_stream: # SSE格式data: {json}\n\n chunk json.dumps({ type: text_delta, text: text }, ensure_asciiFalse) yield fdata: {chunk}\n\n # 发送完成信号 final_message await stream.get_final_message() done_data json.dumps({ type: done, usage: { input_tokens: final_message.usage.input_tokens, output_tokens: final_message.usage.output_tokens } }) yield fdata: {done_data}\n\n except anthropic.APIError as e: error_data json.dumps({type: error, message: str(e)}) yield fdata: {error_data}\n\n except asyncio.CancelledError: # 客户端断开连接正常终止 pass return StreamingResponse( generate(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no # 禁用Nginx缓冲 } )### 2.2 工具调用的流式处理工具调用的流式处理更复杂——需要在流中解析工具调用参数执行工具再继续流式生成pythonapp.post(/api/agent/stream)async def agent_stream(request: Request): 带工具调用的Agent流式接口 body await request.json() messages body.get(messages, []) tools body.get(tools, []) async def generate(): current_messages messages.copy() while True: tool_calls [] current_tool_input {} current_tool_id None current_tool_name None async with client.messages.stream( modelclaude-opus-4-7, max_tokens2000, toolstools, messagescurrent_messages ) as stream: async for event in stream: if event.type content_block_start: if event.content_block.type text: # 文本块开始推送信号 yield fdata: {json.dumps({type: text_start})}\n\n elif event.content_block.type tool_use: current_tool_id event.content_block.id current_tool_name event.content_block.name current_tool_input {} # 告知客户端工具调用开始 yield fdata: {json.dumps({type: tool_start, name: current_tool_name})}\n\n elif event.type content_block_delta: if event.delta.type text_delta: # 推送文本增量 chunk json.dumps({ type: text_delta, text: event.delta.text }, ensure_asciiFalse) yield fdata: {chunk}\n\n elif event.delta.type input_json_delta: # 累积工具输入参数流式JSON片段 # 实际上在content_block_stop时才能解析完整JSON pass elif event.type content_block_stop: if current_tool_id: # 工具调用参数接收完整执行工具 yield fdata: {json.dumps({type: tool_executing, name: current_tool_name})}\n\n tool_calls.append({ id: current_tool_id, name: current_tool_name }) current_tool_id None # 检查停止原因 final_message await stream.get_final_message() if final_message.stop_reason end_turn: yield fdata: {json.dumps({type: done})}\n\n break elif final_message.stop_reason tool_use: # 执行工具并继续对话 tool_results [] for tc in final_message.content: if tc.type tool_use: result await execute_tool(tc.name, tc.input) tool_results.append({ type: tool_result, tool_use_id: tc.id, content: json.dumps(result, ensure_asciiFalse) }) # 推送工具结果给客户端 yield fdata: {json.dumps({type: tool_result, name: tc.name, result: str(result)[:200]})}\n\n # 更新消息历史继续下一轮 current_messages.append({role: assistant, content: final_message.content}) current_messages.append({role: user, content: tool_results}) return StreamingResponse(generate(), media_typetext/event-stream, headers{Cache-Control: no-cache, X-Accel-Buffering: no})### 2.3 流式输出的错误处理与重试pythonimport asynciofrom typing import AsyncGeneratorclass ResilientStreamClient: 带重试机制的流式客户端 def __init__(self, max_retries: int 3, retry_delay: float 1.0): self.client anthropic.AsyncAnthropic() self.max_retries max_retries self.retry_delay retry_delay async def stream_with_retry( self, messages: list, model: str claude-opus-4-7, max_tokens: int 2000 ) - AsyncGenerator[str, None]: 带自动重试的流式生成 last_error None accumulated_text # 记录已收到的文本断点续传 for attempt in range(self.max_retries): try: async with self.client.messages.stream( modelmodel, max_tokensmax_tokens, messagesmessages ) as stream: async for text in stream.text_stream: accumulated_text text yield text return # 成功完成退出 except anthropic.RateLimitError as e: # 频率限制等待后重试 wait_time self.retry_delay * (2 ** attempt) yield f\n[等待重试: {wait_time:.0f}秒] await asyncio.sleep(wait_time) last_error e except anthropic.APITimeoutError as e: if accumulated_text: # 有部分内容可以续写 messages messages [ {role: assistant, content: accumulated_text}, {role: user, content: 请继续} ] yield \n[连接中断正在续写...] last_error e except anthropic.APIConnectionError as e: await asyncio.sleep(self.retry_delay * (attempt 1)) last_error e yield f\n[重试{self.max_retries}次后失败: {str(last_error)}]## 三、前端流式消费### 3.1 React HookuseStreamingChattypescriptimport { useState, useCallback, useRef } from react;interface Message { role: user | assistant; content: string;}interface StreamChunk { type: text_delta | tool_start | tool_result | done | error; text?: string; name?: string; result?: string; message?: string;}export function useStreamingChat() { const [messages, setMessages] useStateMessage[]([]); const [isStreaming, setIsStreaming] useState(false); const [currentResponse, setCurrentResponse] useState(); const abortControllerRef useRefAbortController | null(null); const sendMessage useCallback(async (userMessage: string) { const newMessages: Message[] [ ...messages, { role: user, content: userMessage } ]; setMessages(newMessages); setIsStreaming(true); setCurrentResponse(); abortControllerRef.current new AbortController(); try { const response await fetch(/api/chat/stream, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ messages: newMessages }), signal: abortControllerRef.current.signal }); if (!response.ok) { throw new Error(HTTP ${response.status}); } const reader response.body!.getReader(); const decoder new TextDecoder(); let fullResponse ; while (true) { const { done, value } await reader.read(); if (done) break; const chunk decoder.decode(value, { stream: true }); const lines chunk.split(\n); for (const line of lines) { if (!line.startsWith(data: )) continue; const data line.slice(6); if (data [DONE]) break; try { const parsed: StreamChunk JSON.parse(data); if (parsed.type text_delta parsed.text) { fullResponse parsed.text; setCurrentResponse(fullResponse); } else if (parsed.type done) { setMessages(prev [ ...prev, { role: assistant, content: fullResponse } ]); setCurrentResponse(); } } catch (e) { // 忽略解析错误 } } } } catch (error: any) { if (error.name ! AbortError) { setCurrentResponse(prev prev \n[发生错误请重试]); } } finally { setIsStreaming(false); } }, [messages]); const stopStreaming useCallback(() { abortControllerRef.current?.abort(); setIsStreaming(false); }, []); return { messages, currentResponse, isStreaming, sendMessage, stopStreaming };}### 3.2 流式Markdown渲染流式输出时Markdown文本是逐渐到来的直接渲染会导致频繁的DOM更新和闪烁。解决方案typescriptimport { useEffect, useState } from react;import { marked } from marked;export function StreamingMarkdown({ text }: { text: string }) { const [html, setHtml] useState(); useEffect(() { // 使用requestAnimationFrame避免过度渲染 const rafId requestAnimationFrame(() { // 智能处理未闭合的Markdown语法 const processedText fixIncompleteMarkdown(text); setHtml(marked.parse(processedText) as string); }); return () cancelAnimationFrame(rafId); }, [text]); return ( div classNameprose max-w-none dangerouslySetInnerHTML{{ __html: html }} / );}function fixIncompleteMarkdown(text: string): string { // 处理未闭合的代码块 const codeBlockCount (text.match(//g) || []).length; if (codeBlockCount % 2 ! 0) { return text \n; } // 处理未闭合的粗体/斜体 const boldCount (text.match(/\*\*/g) || []).length; if (boldCount % 2 ! 0) { return text **; } return text;}## 四、性能优化批量合并与节流typescript// 节流Streaming更新避免过于频繁的状态更新导致UI卡顿class StreamingBuffer { private buffer: string ; private onFlush: (text: string) void; private flushInterval: number; private timer: NodeJS.Timeout | null null; constructor(onFlush: (text: string) void, flushIntervalMs: number 50) { this.onFlush onFlush; this.flushInterval flushIntervalMs; } append(text: string): void { this.buffer text; if (!this.timer) { this.timer setInterval(() this.flush(), this.flushInterval); } } flush(): void { if (this.buffer) { this.onFlush(this.buffer); this.buffer ; } } destroy(): void { if (this.timer) { clearInterval(this.timer); this.flush(); } }}## 五、生产部署注意事项Nginx配置默认Nginx会缓冲上游响应导致流式输出变成批量输出。必须禁用缓冲nginxlocation /api/chat/stream { proxy_pass http://backend:8000; proxy_buffering off; # 禁用代理缓冲 proxy_cache off; # 禁用缓存 proxy_read_timeout 300s; # 长超时流式响应时间长 proxy_send_timeout 300s; # SSE必要头 proxy_set_header Connection ; proxy_http_version 1.1; chunked_transfer_encoding on;}超时配置流式请求时间通常比普通请求长10-30倍需要相应调整所有层级的超时设置LB、反向代理、后端框架。## 六、总结LLM流式输出的工程关键点1.后端使用async生成器实现SSE正确处理工具调用的流式解析配置重试与断点续传2.前端用EventSource/fetchReadableStream消费SSE节流UI更新智能处理未闭合Markdown3.基础设施Nginx禁用缓冲配置合理的长超时健康检查需要单独处理流式连接4.错误处理网络中断时的优雅降级频率限制时的退避重试流式输出不只是打字机效果的视觉优化——它是AI应用架构中直接影响用户感知质量的核心工程能力。