Flutter SSE 流式响应：基于 Dio 实现 OpenAI 逐 Token 输出

基本信息

作者: 九狼
链接: https://juejin.cn/post/7607332124487958591

导语

在构建 AI 应用时，流式输出能显著改善用户等待体验，避免长时间白屏。本文将基于 Dio 网络库，详细讲解如何在 Flutter 中通过 SSE（Server-Sent Events）实现 OpenAI 兼容接口的逐 Token 输出。你将掌握从网络请求监听到 UI 逐字渲染的完整逻辑，从而在项目中复现类似 ChatGPT 的打字机效果。

描述

在做 AI 提示词优化器时，“等 10 秒一次性返回大段文本”的体验通常很差。更好的体验是：模型一边生成，你的 UI 一边展示（像打字机逐字出现）

摘要

本文介绍了在 Flutter 开发中，如何利用 Dio 网络库实现兼容 OpenAI 接口的 SSE（Server-Sent Events）流式响应，从而达成类似打字机的逐 Token 输出效果，提升 AI 应用的用户体验。

以下是实现步骤的简要总结：

背景与目的：为了避免 AI 生成大段文本时让用户长时间等待空白页面，采用流式传输让模型一边生成一边展示内容。
接口配置：
- Client 配置：使用 Dio 初始化网络请求，设置 ResponseType.stream 以处理流式数据。
- 请求参数：向 OpenAI 兼容接口发送请求时，需将 stream 参数设置为 true，表明开启流式响应模式。
数据解析核心逻辑：
- 监听流数据：通过监听 Dio 返回的 Stream<Uint8List>，获取服务器推送的二进制数据块。
- 数据转换：利用 utf8.decode 将二进制数据转换为字符串。
- 协议解析：SSE 数据通常以 data: 开头。代码逻辑需要识别并提取 data: 后的 JSON 字符串。
- 状态处理：
  - 过滤心跳包：忽略 [DONE] 结束标记或空数据。
  - 提取内容：解析 JSON 对象（通常位于 choices[0].delta.content 字段中）以获取具体的 Token 文本。
- 异常处理：需处理数据分片问题（例如一个 JSON 被截断在两个数据包中），通过缓冲区或特定的字符串分割逻辑（如按 data: 分割）来组装完整的 JSON 对象。
UI 展示：将解析出的文本通过 setState 或状态管理工具（如 Riverpod/Provider）实时更新到界面组件上，实现视觉上的逐字打印效果。

通过这种方式，开发者可以在 Flutter 中高效地实现 AI 对话的流式输出功能。

文章中心观点： 该文章主张在 Flutter AI 应用开发中，应摒弃传统的“请求-等待-响应”全量模式，转而利用 Dio 网络库对 Server-Sent Events (SSE) 的底层支持，实现兼容 OpenAI 接口规范的流式逐 Token 渲染，以构建具备低延迟感和高交互性的用户体验。

支撑理由与深度评价：

1. 用户体验维度的“感知性能”优化

事实陈述： 文章指出大模型生成（LLM）通常需要数秒甚至更久，一次性返回会导致用户产生“卡死”或“无响应”的焦虑感。
作者观点： 通过流式输出（Typewriter Effect），即首字极速呈现（TTFB）后的连续字符上屏，可以显著提升系统的流畅度感知。
深度分析： 这不仅是 UI 技巧，更是心理学层面的“反馈回路”构建。在 AI 应用中，用户往往容忍“慢生成”，但绝不容忍“无反馈”。文章抓住了 AI 原生应用体验的核心——即时性。这种将“计算时间”转化为“阅读时间”的策略，是目前所有主流 AI 产品（如 ChatGPT, Claude）的标准交互范式。

2. 技术实现的工程化解法

事实陈述： 文章详细描述了如何处理 SSE 协议，特别是针对 Dio 库配置 ResponseType.stream，并解析 data: [DONE] 等特定格式的数据。
你的推断： 作者试图解决 Flutter 生态中 HTTP 客户端与 SSE 适配的痛点。虽然 Dart 原生 HttpClient 或 WebSocket 也能实现，但 Dio 是 Flutter 社区最主流的网络库，基于 Dio 的方案具有极高的迁移成本优势和生态兼容性。
深度分析： 文章的技术价值在于协议兼容性。OpenAI 的 SSE 流并非标准 SSE（它带有特定的 JSON 格式和前缀），直接处理容易出现 JSON 解析错误或流截断。文章提供了一套针对 OpenAI 格式的“胶水代码”，解决了从二进制流到语义文本的转换过程。

3. AI 应用架构的现代化演进

事实陈述： 文章背景是开发“AI 提示词优化器”，属于典型的 AI 原生应用场景。
作者观点： 流式响应是此类应用的基础设施，而非可选项。
深度分析： 随着大模型从“玩具”走向“工具”，应用架构正从 BFF（Backend for Frontend）全代理转向端侧直连模型。文章体现了这种趋势：Flutter 客户端直接处理流式协议，减少中间层转发带来的延迟损耗。这是对移动端 AI 应用架构的一种合理前瞻。

反例与边界条件：

流式传输的“不稳定性”代价：
- 事实陈述： 流式输出意味着 UI 需要频繁重绘。
- 你的推断： 在低端 Android 设备或复杂的富文本渲染场景下（如同时支持 Markdown 语法高亮），每秒数十次的 setState 或 TextSpan 更新可能导致严重的 UI 掉帧（Jank），反而比全量更新体验更差。文章未深入讨论防抖或批量渲染策略（例如每 50ms 或每 3 个 Token 更新一次 UI），这是工程落地中的常见坑点。
错误处理的复杂性：
- 事实陈述： 文章主要关注成功路径的数据解析。
- 你的推断： 在 SSE 长连接中，网络抖动、Token 超限或服务端 500 错误可能发生在流传输的任意时刻。相比一次性请求的“全有或全无”，流式请求的错误恢复逻辑（如断点续传、错误回滚）极其复杂。如果文章未提及 Stream 的异常捕获与状态管理，可能会导致生产环境下的“死循环”或数据丢失。

可验证的检查方式：

首字节延迟指标测试：
- 验证方式： 使用 Flutter DevTools 的 Network Profiling，对比“全量模式”与“流式模式”从发起请求到 UI 渲染第一个字符的时间差。在弱网环境下（3G 网络），流式模式应能将用户感知延迟降低 60% 以上。
内存与 CPU 占用曲线观察：
- 验证方式： 在应用中运行长文本生成（1000+ Tokens），使用 Flutter Performance 监控内存占用。检查是否存在内存泄漏，以及 UI 线程的 CPU 占用是否因频繁刷新而出现持续波峰。若 CPU 长期处于高位，说明缺乏批量渲染优化。
协议兼容性压力测试：
- 验证方式： 模拟网络中断（开启飞行模式）或服务端返回非标准 JSON 片段。观察代码是否会抛出未捕获的 FormatException，或者 App 是否会发生 Crash。健壮的 SSE 实现应能优雅处理流中断。

实际应用建议：

引入渲染缓冲： 不要每收到一个 Token 就更新一次 UI。建议实现一个“微缓冲区”，例如每 30-50ms 或累积 5-10 个 Token 再提交一次渲染，以平衡流畅度与性能。
Markdown 实时解析优化： 如果输出包含 Markdown 格式，直接对

学习要点

核心实现是利用 Dio 的 ResponseType.stream 类型将响应转换为二进制流，并配合 utf8.decoder 解码数据以支持 SSE 协议。
处理流式数据的关键在于自定义转换器，通过监听 Stream 逐块解析数据，并利用正则表达式匹配 SSE 格式中的 data 前缀。
实现逐 Token 输出的逻辑是维护一个完整的字符串缓冲区，将每次接收到的增量数据（delta）追加到现有内容中。
针对 OpenAI 接口，必须设置请求头 Accept: text/event-stream 并在 Body 中包含 stream: true 参数以开启流式传输。
处理网络异常和流中断的健壮性至关重要，需要在 StreamBuilder 中妥善处理 ConnectionState 的各种状态（如 waiting, active, done）。
实现打字机效果依赖于 UI 层的及时刷新，建议将流式数据与 Flutter 的 StatefulWidget 结合使用以实现界面实时更新。
SSE 数据流通常以 data: [DONE] 结尾，代码中需包含对此特定结束标记的判断以终止流处理。

常见问题

1: 使用 Dio 发起 SSE 请求时，如何配置才能正确接收流式数据？

A: 要使用 Dio 接收 SSE (Server-Sent Events) 流，核心在于配置 ResponseType。默认情况下 Dio 会等待整个响应体下载完成，必须显式指定接收类型为 stream。

关键配置步骤：

设置 options.responseType = ResponseType.stream。
设置请求头 Accept: text/event-stream。
设置请求头 Cache-Control: no-cache 以确保数据实时推送。

代码示例：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
final dio = Dio();
final response = await dio.post<ResponseBody>(
  'https://api.openai.com/v1/chat/completions',
  data: {
    "model": "gpt-3.5-turbo",
    "messages": [{"role": "user", "content": "Hello"}],
    "stream": true, // 关键：开启流式模式
  },
  options: Options(
    headers: {
      "Authorization": "Bearer $apiKey",
      "Accept": "text/event-stream",
      "Cache-Control": "no-cache",
    },
    responseType: ResponseType.stream, // 关键：指定响应类型为流
  ),
);

2: 如何从 Dio 返回的 `ResponseBody` 中逐行解析出 SSE 数据？

2: 如何从 Dio 返回的 ResponseBody 中逐行解析出 SSE 数据？

A: OpenAI 的 SSE 接口返回的是 text/event-stream 格式，数据以 data: {...} 的形式逐行发送。你需要监听 response.data.stream，处理字节流并将其转换为 UTF-8 字符串，然后按行分割。

解析逻辑：

监听 Stream：response.data.stream.listen((data) { ... })。
编码转换：使用 utf8.decode(data) 将字节列表转为字符串。
数据清洗：SSE 数据可能包含多个换行符，需要先按 \n 分割，再过滤掉空行。
前缀处理：每行有效数据通常以 data: 开头，需要去除该前缀。
结束标记：OpenAI 会在流结束时发送 [DONE]，需要判断并终止流。

代码片段：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
response.data.stream.listen((data) {
  final decoded = utf8.decode(data);
  final lines = decoded.split('\n');
  
  for (final line in lines) {
    final trimmed = line.trim();
    if (trimmed.isEmpty) continue;
    
    if (trimmed.startsWith('data: ')) {
      final jsonStr = trimmed.substring(6); // 去除 'data: '
      if (jsonStr == '[DONE]') {
        // 流结束逻辑
        return;
      }
      // 解析 JSON
      final jsonData = jsonDecode(jsonStr);
      // 处理 choices.delta.content
    }
  }
});

3: 在处理 SSE 流时，如何解决 JSON 数据被截断或跨行传输的问题？

A: 这是网络流式传输中最常见的问题。由于 TCP 分包机制，一次 stream.listen 接收到的数据块可能不包含完整的 JSON，或者包含半个 JSON。直接 jsonDecode 会抛出格式异常。

解决方案：使用缓冲区 你需要维护一个字符串缓冲区，将接收到的数据追加到缓冲区中，然后尝试从缓冲区中提取完整的行进行处理。

实现思路：

定义全局变量 String buffer = ''。
在 listen 中将新数据追加到 buffer。
检查 buffer 中是否存在换行符 \n。
如果存在，截取第一个换行符之前的内容作为“一行”进行处理。
处理完毕后，将该行从 buffer 中移除，保留剩余部分供下次使用。

4: OpenAI 返回的流式 JSON 结构是怎样的？如何提取具体的 Token 文本？

A: OpenAI 兼容接口在流式模式下，返回的 JSON 对象结构略有不同。文本内容不是直接返回的，而是嵌套在 choices 数组的 delta 对象中。

数据结构示例：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
{
  "id": "chatcmpl-123",
  "object": "chat.completion.chunk",
  "created": 1677652288,
  "model": "gpt-3.5-turbo",
  "choices": [
    {
      "index": 0,
      "delta": {
        "content": "Hello"
      },
      "finish_reason": null
    }
  ]
}

提取逻辑： 你需要访问 jsonData['choices'][0]['delta']['content']。 注意：

content 字段在某些情况下可能不存在（例如仅发送角色信息时），访问前需判空。
finish_reason 为 null 表示流未结束，不为 null（如 stop）表示该条消息生成结束。

引用

掘金原文: https://juejin.cn/post/7607332124487958591

注：文中事实性信息以以上引用为准；观点与推断为 AI Stack 的分析。

站内链接

分类：前端 / AI 工程
标签： Flutter / Dio / SSE / 流式响应 / OpenAI / LLM / 逐字输出 / 移动端开发
场景： AI/ML项目 / 大语言模型

Spring AI 多模型对话实战：统一接口与 Redis 记忆
Flutter计划支持Packaged AI Assets以提升AI理解能力
kirara-ai：支持多平台接入的多模态AI聊天机器人框架
Codex 应用：基于 GPT-3 的代码生成工具
Codex 应用：基于 OpenAI 模型的代码生成工具 本文由 AI Stack 自动生成，提供深度内容分析。

AI Stack

Flutter SSE 流式响应：基于 Dio 实现 OpenAI 逐 Token 输出