(LLM系列)流式输出(Streaming)实现：提升用户体验

2026-02-02
SSE Streaming Next.js Qwen AI

流式输出(Streaming)实现：提升用户体验

在现代 Web 应用中，用户体验的关键在于响应速度和交互反馈。当处理耗时操作时，传统的”等待-返回”模式往往让用户感到焦虑。流式输出（Streaming）技术通过逐步返回数据，让用户实时看到处理进度，极大提升了体验感知。本文将深入探讨如何在 Qwen Chatbot 项目中使用 SSE（Server-Sent Events）和异步处理实现流式输出。

为什么需要流式输出？

想象一个场景：用户向 AI 助手提问，传统方式需要等待完整答案生成后才能看到结果，可能需要等待数十秒。而流式输出允许答案逐字逐句地呈现，就像真人对话一样自然。这种即时反馈不仅减少了感知等待时间，还增强了应用的互动性。

流式输出的典型应用场景包括：

AI 对话系统（ChatGPT 式交互）
大文件处理进度
实时日志输出
数据分析报告生成

技术选型：为什么选择 SSE？

在实现流式数据传输时，我们有几种选择：WebSocket、HTTP/2 Server Push 和 SSE。对于单向数据流（服务器到客户端），SSE 是最优方案：

简单易用：基于 HTTP 协议，无需复杂握手
自动重连：浏览器原生支持断线重连
轻量级：相比 WebSocket 更节省资源
防火墙友好：使用标准 HTTP 端口

Qwen Chatbot 项目中的实现方案

服务端：API 路由实现

Next.js 的 Pages Router 提供了强大的 API 路由功能，非常适合实现 SSE。以下是 Qwen Chatbot 项目中的完整实现示例：

1
import type { NextApiRequest, NextApiResponse } from 'next';
2
import OpenAI from 'openai';
3

4
export default async function handler(req: NextApiRequest, res: NextApiResponse) {
5
  if (req.method !== 'POST') {
6
    return res.status(405).json({ error: 'Method not allowed' });
7
  }
8

9
  const { messages, stream = false, model, temperature = 0.7, top_p = 0.9, max_tokens = 2048 } = req.body;
10

11
  // 验证必需字段
12
  if (!messages || !Array.isArray(messages)) {
13
    return res.status(400).json({ error: 'Messages are required and must be an array' });
14
  }
15

16
  try {
17
    // 创建 OpenAI 兼容的客户端，适配通义千问
18
    const client = new OpenAI({
19
      apiKey: process.env.OPENAI_API_KEY || '',
20
      baseURL: process.env.OPENAI_API_BASE || 'https://dashscope.aliyuncs.com/compatible-mode/v1',
21
    });
22

23
    if (stream) {
24
      // 使用 TransformStream 实现流式响应
25
      const encoder = new TextEncoder();
26
      const stream = new TransformStream();
27
      const writer = stream.writable.getWriter();
28

29
      // 异步处理函数
30
      (async () => {
31
        try {
32
          // 通义千问API支持system message，直接使用原始消息
33
          const response = await client.chat.completions.create({
34
            model: model || process.env.MODEL_NAME || 'qwen-max',
35
            messages,
36
            stream: true,
37
            temperature,
38
            top_p,
39
            max_tokens,
40
            stream_options: { include_usage: true }, // 包含使用量信息
41
          });
42

43
          // 逐块发送数据
44
          for await (const chunk of response) {
45
            const content = chunk.choices[0]?.delta?.content;
46

47
            // 如果有内容，发送内容数据
48
            if (content) {
49
              const data = `data: ${JSON.stringify({ content })}\n\n`;
50
              await writer.write(encoder.encode(data));
51
            }
52

53
            // 如果有usage信息，发送token使用数据
54
            if (chunk.usage) {
55
              const tokenData = {
56
                usage: {
57
                  prompt_tokens: chunk.usage.prompt_tokens,
58
                  completion_tokens: chunk.usage.completion_tokens,
59
                  total_tokens: chunk.usage.total_tokens,
60
                }
61
              };
62
              const data = `data: ${JSON.stringify(tokenData)}\n\n`;
63
              await writer.write(encoder.encode(data));
64
            }
65
          }
66

67
          // 发送结束信号
68
          await writer.write(encoder.encode('data: [DONE]\n\n'));
69
        } catch (error: any) {
70
          // 发送错误信息
71
          await writer.write(
72
            encoder.encode(`data: ${JSON.stringify({ error: error.message || 'AI service error' })}\n\n`)
73
          );
74
        } finally {
75
          await writer.close();
76
        }
77
      })();
78

79
      // 返回 SSE 响应
80
      res.setHeader('Content-Type', 'text/event-stream');
81
      res.setHeader('Cache-Control', 'no-cache');
82
      res.setHeader('Connection', 'keep-alive');
83
      return new Response(stream.readable, {
84
        headers: {
85
          'Content-Type': 'text/event-stream',
86
          'Cache-Control': 'no-cache',
87
          'Connection': 'keep-alive',
88
        },
89
      });
90
    } else {
91
      // 非流式响应
92
      // 通义千问API支持system message，直接使用原始消息
93
      const response = await client.chat.completions.create({
94
        model: model || process.env.MODEL_NAME || 'qwen-max',
95
        messages,
96
        temperature,
97
        top_p,
98
        max_tokens,
99
      });
100

101
      const content = response.choices[0]?.message?.content || '';
102
      const usage = response.usage;
103

104
      res.status(200).json({
105
        content,
106
        usage: usage ? {
107
          prompt_tokens: usage.prompt_tokens,
108
          completion_tokens: usage.completion_tokens,
109
          total_tokens: usage.total_tokens,
110
        } : undefined
111
      });
112
    }
113
  } catch (error: any) {
114
    console.error('Error calling Qwen API:', error);
115

116
    let errorMessage = 'An error occurred while calling the API';
117
    let statusCode = 500;
118

119
    if (error.status === 401) {
120
      errorMessage = 'Authentication failed. Please check your API key.';
121
      statusCode = 401;
122
    } else if (error.status === 403) {
123
      errorMessage = 'Access forbidden. Please check your API permissions.';
124
      statusCode = 403;
125
    } else if (error.status === 429) {
126
      errorMessage = 'Rate limit exceeded. Please try again later.';
127
      statusCode = 429;
128
    } else if (error.status === 404 && error.message.includes('model')) {
129
      errorMessage = 'Model not found or access denied. Please check the model name and your API permissions. Try using "qwen-max" instead of "qwen-max-0102".';
130
      statusCode = 404;
131
    } else if (error.message) {
132
      errorMessage = error.message;
133
    }
134

135
    res.status(statusCode).json({
136
      error: errorMessage,
137
      details: process.env.NODE_ENV === 'development' ? error.toString() : undefined
138
    });
139
  }
140
}

关键点解析：

TransformStream：Next.js 推荐的流处理方式，比传统的 ReadableStream 更灵活
TextEncoder：将字符串转换为 Uint8Array，符合流传输要求
SSE 格式：数据必须以 data: 开头，以 \n\n 结尾
异步 IIFE：立即执行的异步函数，避免阻塞响应返回
通义千问适配：使用 OpenAI 兼容的 API 客户端，适配通义千问 API

客户端：React 组件实现

客户端需要处理 SSE 连接并实时更新 UI，以下是 Qwen Chatbot 项目中的实现：

1
// pages/chat.tsx (SSE 处理部分)
2
const handleSubmit = async (e: React.FormEvent) => {
3
  e.preventDefault();
4
  if (!inputMessage.trim() || isLoading) return;
5

6
  // 添加用户消息
7
  const userMessage = { role: 'user', content: inputMessage };
8
  dispatch({ type: 'ADD_MESSAGE', payload: userMessage });
9
  dispatch({ type: 'SET_INPUT_MESSAGE', payload: '' });
10
  setIsLoading(true);
11

12
  try {
13
    // 准备消息数组，如果选择了角色并且该角色有系统提示，则在开头添加系统消息
14
    let messagesToSend = [...messages, userMessage];
15

16
    if (selectedRoleId) {
17
      const selectedRole = roles.find(r => r.id === selectedRoleId);
18
      if (selectedRole && selectedRole.systemPrompt) {
19
        // 检查是否已经有系统消息，如果没有则添加
20
        const hasSystemMessage = messages.some(msg => msg.role === 'system');
21
        if (!hasSystemMessage) {
22
          messagesToSend = [{ role: 'system', content: selectedRole.systemPrompt }, ...messagesToSend];
23
        }
24
      }
25
    }
26

27
    // 发送请求到后端 API
28
    // 使用流式响应获取实时token使用情况
29
    const response = await fetch('/api/qwen', {
30
      method: 'POST',
31
      headers: {
32
        'Content-Type': 'application/json',
33
      },
34
      body: JSON.stringify({
35
        messages: messagesToSend,
36
        stream: true, // 使用流式响应
37
        model: modelConfig.model,
38
        temperature: modelConfig.temperature,
39
        top_p: modelConfig.top_p,
40
        max_tokens: modelConfig.max_tokens,
41
      }),
42
    });
43

44
    if (!response.ok) {
45
      const errorData = await response.json();
46
      throw new Error(errorData.error || 'Failed to get response from API');
47
    }
48

49
    // 处理流式响应
50
    const reader = response.body?.getReader();
51
    if (!reader) {
52
      throw new Error('Could not read response body');
53
    }
54

55
    const decoder = new TextDecoder();
56
    let assistantMessage: Message = { role: 'assistant', content: '', usage: undefined };
57

58
    // 创建助手消息并添加到消息列表
59
    const newAssistantMessage: Message = { role: 'assistant', content: '', usage: undefined };
60
    dispatch({ type: 'ADD_MESSAGE', payload: newAssistantMessage });
61

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

66
      const chunk = decoder.decode(value, { stream: true });
67
      const lines = chunk.split('\n');
68

69
      for (const line of lines) {
70
        if (line.startsWith('data: ')) {
71
          const data = line.slice(6); // 移除 'data: ' 前缀
72

73
          if (data === '[DONE]') {
74
            // 流结束
75
            break;
76
          }
77

78
          try {
79
            const parsed = JSON.parse(data);
80
            if (parsed.content) {
81
              // 更新最后一条消息的内容
82
              assistantMessage.content += parsed.content;
83
              // 只更新助手消息，保留之前的消息
84
              const updatedMessages = [...messages, { ...assistantMessage }];
85
              dispatch({ type: 'SET_MESSAGES', payload: updatedMessages });
86
            } else if (parsed.usage) {
87
              // 更新最后一条消息的使用情况
88
              assistantMessage.usage = parsed.usage;
89
              const updatedMessages = [...messages, { ...assistantMessage }];
90
              dispatch({ type: 'SET_MESSAGES', payload: updatedMessages });
91
            }
92
          } catch (e) {
93
            // 忽略无法解析的数据行
94
            console.error('Error parsing data:', e);
95
          }
96
        }
97
      }
98
    }
99

100
    // 在流结束后记录对话历史
101
    const updatedMessages = [...messages, assistantMessage]; // 获取包含最新消息的完整消息列表
102
    const lastAssistantMessage = updatedMessages[updatedMessages.length - 1]; // 最后一条消息应该是助手的回复
103

104
    if (lastAssistantMessage && lastAssistantMessage.role === 'assistant') {
105
      const newHistoryEntry: ConversationHistory = {
106
        id: Date.now(), // 使用时间戳作为唯一ID
107
        timestamp: new Date().toISOString(),
108
        input: inputMessage,
109
        output: lastAssistantMessage.content,
110
        model: modelConfig.model,
111
        params: {
112
          temperature: modelConfig.temperature,
113
          top_p: modelConfig.top_p,
114
          max_tokens: modelConfig.max_tokens,
115
        },
116
        tokenUsage: assistantMessage.usage ? {
117
          prompt_tokens: assistantMessage.usage.prompt_tokens,
118
          completion_tokens: assistantMessage.usage.completion_tokens,
119
          total_tokens: assistantMessage.usage.total_tokens
120
        } : undefined,
121
        evaluation: '' // 可以让使用者手动填写或系统自动生成
122
      };
123

124
      dispatch({ type: 'ADD_TO_HISTORY', payload: newHistoryEntry }); // 添加到历史记录开头
125
    }
126
  } catch (error: any) {
127
    console.error('Error:', error);
128
    dispatch({ type: 'ADD_MESSAGE', payload: {
129
      role: 'assistant',
130
      content: `Error: ${error.message || 'An unknown error occurred'}`
131
    }});
132

133
    // 即使出错也记录历史
134
    const errorMessage = `Error: ${error.message || 'An unknown error occurred'}`;
135
    const newHistoryEntry: ConversationHistory = {
136
      id: Date.now(), // 使用时间戳作为唯一ID
137
      timestamp: new Date().toISOString(),
138
      input: inputMessage,
139
      output: errorMessage,
140
      model: modelConfig.model,
141
      params: {
142
        temperature: modelConfig.temperature,
143
        top_p: modelConfig.top_p,
144
        max_tokens: modelConfig.max_tokens,
145
      },
146
      tokenUsage: undefined, // 错误情况下无token使用数据
147
      evaluation: 'Error occurred' // 标记为错误
148
    };
149

150
    dispatch({ type: 'ADD_TO_HISTORY', payload: newHistoryEntry }); // 添加到历史记录开头
151
  } finally {
152
    setIsLoading(false);
153
  }
154
};

核心实现要点：

ReadableStream Reader：使用 getReader() 逐块读取数据
TextDecoder：将二进制数据解码为字符串
状态更新：通过 Redux-like 状态管理更新消息
错误处理：妥善处理解析错误和网络异常
Token 使用情况：实时更新 API 调用的 token 使用情况

前端打字机效果实现

为了让流式输出看起来更自然，我们在前端实现了打字机效果：

1
import React, { useState, useEffect, useRef } from 'react';
2
import styles from '../styles/TypeWriterEffect.module.css';
3

4
interface TypeWriterEffectProps {
5
  text: string;
6
  speed?: number; // 打字速度，毫秒/字符
7
  className?: string; // 自定义类名
8
}
9

10
const TypeWriterEffect: React.FC<TypeWriterEffectProps> = ({
11
  text,
12
  speed = 50, // 放慢速度到50ms/字符，让效果更明显
13
  className = ''
14
}) => {
15
  const [displayedText, setDisplayedText] = useState('');
16
  const [isTyping, setIsTyping] = useState(true);
17
  const timeoutRef = useRef<NodeJS.Timeout | null>(null);
18

19
  useEffect(() => {
20
    // 每次text变化时重置
21
    setDisplayedText('');
22
    setIsTyping(true);
23

24
    // 清除之前的定时器
25
    if (timeoutRef.current) {
26
      clearTimeout(timeoutRef.current);
27
    }
28

29
    // 如果文本为空，直接返回
30
    if (!text) {
31
      setIsTyping(false);
32
      return;
33
    }
34

35
    // 开始打字
36
    let index = 0;
37
    const typeNextChar = () => {
38
      if (index < text.length) {
39
        const char = text[index];
40
        // 确保字符不是undefined或null
41
        if (char !== undefined && char !== null) {
42
          // 强制更新，避免React优化
43
          setDisplayedText(prev => prev + String(char));
44
        }
45
        index++;
46
        timeoutRef.current = setTimeout(typeNextChar, speed);
47
      } else {
48
        setIsTyping(false);
49
      }
50
    };
51

52
    timeoutRef.current = setTimeout(typeNextChar, speed);
53

54
    // 清理
55
    return () => {
56
      if (timeoutRef.current) {
57
        clearTimeout(timeoutRef.current);
58
      }
59
    };
60
  }, [text, speed]);
61

62
  return (
63
    <span className={`${styles.typeWriterText} ${className}`}>
64
      {displayedText}
65
      {isTyping && <span className={styles.cursor}>|</span>}
66
    </span>
67
  );
68
};
69

70
export default TypeWriterEffect;

性能优化技巧

1. 背压处理（Backpressure）

当客户端处理速度跟不上服务端发送速度时，需要实现背压机制：

1
const writer = stream.writable.getWriter();
2

3
async function writeWithBackpressure(data: string) {
4
  await writer.ready; // 等待缓冲区可写
5
  await writer.write(encoder.encode(data));
6
}

2. 分块策略

合理控制每次发送的数据量，避免过小（频繁网络开销）或过大（失去流式效果）：

1
let buffer = '';
2
const CHUNK_SIZE = 50; // 每 50 个字符发送一次
3

4
for (const char of response) {
5
  buffer += char;
6
  if (buffer.length >= CHUNK_SIZE) {
7
    await writeWithBackpressure(`data: ${JSON.stringify({ content: buffer })}\n\n`);
8
    buffer = '';
9
  }
10
}

3. 连接管理

实现心跳检测，防止连接意外断开：

1
// 服务端定期发送心跳
2
const heartbeatInterval = setInterval(() => {
3
  writer.write(encoder.encode(': heartbeat\n\n'));
4
}, 30000);
5

6
// 清理
7
process.on('exit', () => clearInterval(heartbeatInterval));

实战案例：Qwen Chatbot 中的集成

在 Qwen Chatbot 项目中，我们将以上技术整合到了真实的 AI 对话系统中：

消息组件集成：在 ChatWindow 组件中使用 TypeWriterEffect 显示助手回复
状态管理：使用全局状态管理器跟踪消息流
实时更新：SSE 流实时更新助手消息内容
打字效果：前端实现的打字机效果增强用户体验

1
import TypeWriterEffect from './TypeWriterEffect';
2

3
// ...
4

5
{messages.map((msg, index) => (
6
  <div key={index} className={`${styles.message} ${styles[msg.role]}`}>
7
    <div className={styles.avatar}>
8
      {msg.role === 'user' ? '👤' : '🤖'}
9
    </div>
10
    <div className={styles.content}>
11
      {msg.role === 'assistant' ? (
12
        <TypeWriterEffect text={msg.content} speed={20} />
13
      ) : (
14
        msg.content
15
      )}
16
      {msg.usage && (
17
        <div className={styles.tokenInfo}>
18
          Tokens: {msg.usage.total_tokens} (Prompt: {msg.usage.prompt_tokens}, Completion: {msg.usage.completion_tokens})
19
        </div>
20
      )}
21
    </div>
22
  </div>
23
))}

注意事项与最佳实践

超时处理：设置合理的超时时间，避免连接永久挂起
错误恢复：客户端应实现重试机制，处理网络波动
资源清理：确保 writer 和 reader 正确关闭，防止内存泄漏
CORS 配置：跨域场景需要正确配置响应头
进度指示：提供明确的加载状态，让用户知道系统正在工作
打字机效果优化：不能依赖 SSE 返回粒度，必须在前端主动控制显示节奏
API 兼容性：适配不同 LLM 提供商的 API 格式差异

总结

流式输出通过 SSE 和异步处理技术，将”等待-返回”的交互模式转变为”实时反馈”的体验。在 Qwen Chatbot 项目中，借助 Next.js 和 Web Streams API，我们优雅地实现了这一功能。无论是 AI 对话、数据处理还是实时日志，流式输出都能显著提升用户体验。

通过结合后端流式传输和前端打字机效果，我们实现了既高效又直观的用户交互体验。随着 Web 技术的发展，流式处理将成为构建现代 AI 应用的标配能力。掌握这项技术，让你的应用更加流畅、响应更加迅速，为用户带来更好的交互体验。