1. 从“玩具”到“产品”为什么要在Flutter里做流式AI最近几个月我身边不少做移动端的朋友都在聊一个话题怎么把现在火热的AI能力特别是那种能打字、能说话、能实时生成内容的“流式AI”真正塞进自己的App里。大家试过各种方案有的用WebView套壳有的调系统浏览器但体验总是不尽如人意——要么交互生硬要么性能拉胯要么就是完全脱离了App的原生体验。这让我想起了几年前刚接触Flutter时的情景。当时大家争论的是“用Flutter能不能做出媲美原生的流畅度”而现在问题变成了“用Flutter能不能做出媲美大厂的原生AI体验”。我花了些时间用Flutter 3.41完整走通了一个“App版流式AI系统”的实战项目从网络请求、状态管理、UI渲染到性能优化踩了不少坑也总结出了一套相对可靠的方案。这篇文章我就来聊聊怎么从零开始把一个听起来很“未来”的流式AI概念落地成一个用户感知流畅、开发者维护顺手的真实Flutter功能模块。这不是一个简单的API调用教程而是一次关于如何用Flutter技术栈去“驯服”流式数据、构建复杂交互的完整实践。所谓“流式AI”在App语境下核心体验就是“边生成边显示”。比如你问AI一个问题它不是等全部答案在服务器端生成好了再一股脑丢给你而是一个字一个字、或者一个词一个词地“流”到你的手机屏幕上。这种“实时感”对用户体验的提升是巨大的但背后对客户端的技术要求也更高你需要处理不完整的数据、管理复杂的渲染状态、还要保证在数据持续到达时UI依然流畅不卡顿。Flutter 3.41在Dart语言特性、异步编程模型和渲染管线上的诸多改进让我们有了更好的武器库来应对这些挑战。2. 技术选型与架构设计不止是调用一个API在动手写代码之前我们先得把架子搭好。一个常见的误区是认为实现流式AI就是找到一个支持流式响应的API然后在Flutter里用http包发起一个请求接着在setState里更新文本。这么做很快就能看到效果但一旦需求稍微复杂比如需要支持中途停止、重新生成、历史会话、错误重试代码就会迅速变成一团乱麻。因此一个清晰的分层架构至关重要。2.1 核心分层数据流、业务逻辑与UI的分离我采用的是一种改良后的MVVM模式结合Flutter的响应式特性具体分为四层数据层Repository职责是纯粹的数据获取。它不关心数据怎么用只负责以最原始的形式从网络或本地缓存拿到数据。对于流式AI这里的关键是处理Server-Sent EventsSSE或WebSocket等流式协议。我强烈推荐使用dart:io中的HttpClient来手动处理SSE而不是依赖一些封装过度的第三方包因为我们需要对数据流的生命周期连接、接收、关闭、错误有绝对的控制权。模型层Model定义数据结构。除了常规的请求参数如prompt、model和完整的响应模型必须专门为流式数据设计一个“数据块”模型。这个模型需要包含当前收到的文本片段、该片段是否是最后一个isFinish、以及可能携带的额外信息如本次生成的token数、思考过程等元数据。视图模型层ViewModel/Bloc/Cubit这是业务逻辑的核心。它持有数据层实例接收UI层的动作如用户发送消息然后指挥数据层工作并将原始数据流转换为UI层能够直接消费的状态流。这里我们会大量使用Stream和StreamController。一个健壮的ViewModel需要处理以下状态空闲、连接中、流式接收中、完成、错误、用户手动停止。UI层View根据视图模型提供的状态流来构建界面。它不应该包含任何业务逻辑只负责“显示什么”和“转发用户操作”。对于流式文本的显示我们需要一个能够优雅处理文本内容不断增长的Widget。2.2 为什么选择SSE而非WebSocket目前绝大多数提供流式响应的AI服务如OpenAI的Chat Completions、国内各大模型的流式接口都支持SSE协议。SSE是基于HTTP的单向通信服务器可以主动推送数据片段到客户端。相比于WebSocketSSE有几个优势在移动端场景下尤为突出更简单它就是HTTP复用现有HTTP基础设施无需额外的协议握手和连接管理逻辑。自动重连浏览器环境下的SSE实现自带重连机制虽然我们在Dart中需要自己实现但逻辑依然比WebSocket简单。更利于调试你甚至可以直接用curl命令来测试SSE接口数据格式一目了然。在Dart中处理SSE的核心在于监听HttpClientResponse的stream。下面是一个最简化的数据层方法原型它揭示了如何处理分块传输编码chunked的数据import dart:async; import dart:convert; import dart:io; class AIService { final HttpClient _client HttpClient(); StreamString streamCompletion({ required String prompt, required String apiKey, }) async* { final request await _client.postUrl(Uri.parse(https://api.example.com/v1/chat/completions)); // 设置Headers request.headers.set(Authorization, Bearer $apiKey); request.headers.set(Content-Type, application/json); request.headers.set(Accept, text/event-stream); // 关键声明接受SSE流 final body jsonEncode({ model: gpt-3.5-turbo, messages: [{role: user, content: prompt}], stream: true, // 关键开启流式 }); request.write(body); final response await request.close(); if (response.statusCode ! 200) { throw Exception(请求失败: ${response.statusCode}); } // 核心逐块读取响应流 await for (final chunk in response.transform(utf8.decoder)) { // SSE数据格式为 data: {...}\n\n需要按行解析 final lines chunk.split(\n); for (final line in lines) { if (line.startsWith(data: ) line.length 6) { final dataStr line.substring(6); if (dataStr [DONE]) { // 流结束标志 return; } try { final data jsonDecode(dataStr); final content data[choices][0][delta][content]; if (content ! null) { yield content; // 使用yield将每个内容片段输出为Stream } } catch (e) { // 忽略解析中的非致命错误可能是不完整的json片段 } } } } } }这段代码是数据层的核心。async*和yield关键字让我们能轻松地创建一个异步数据流。transform(utf8.decoder)将字节流转换为字符串流然后我们按照SSE的规范data:前缀和\n\n分隔来解析出每一个有效的JSON数据块。2.3 状态管理方案Riverpod的优雅实践对于视图模型层状态管理方案的选择直接决定了代码的整洁度和可维护性。经过对比我选择了Riverpod因为它提供了无与伦比的灵活性和编译安全性。我们将使用StreamProvider和StateNotifierProvider或AsyncNotifierProvider来组合我们的状态。StreamProvider用于直接暴露从AIService获取的原始文本流。这个流是“热”的一旦被监听就开始接收数据。StateNotifierProvider用于管理更高级的UI状态比如当前是否正在生成、已生成的完整历史消息列表、错误信息等。它会监听StreamProvider并将新的文本片段整合到历史消息中。这种分离的好处是UI可以同时监听多个Provider一个用于获取最新的动态文本片段用于实时显示另一个用于获取完整的、稳定的对话历史用于展示和持久化。3. 构建响应式视图模型处理流式状态与业务逻辑有了数据层我们就可以构建视图模型了。视图模型是连接“原始数据流”和“UI状态”的桥梁。它的核心任务是将一个StreamString零散的文本片段转换成一个StreamConversationStateUI可以直接渲染的完整状态。3.1 定义状态类首先我们需要一个精细的状态类来描述对话可能处于的各种情况。part conversation_state.freezed.dart; // 使用freezed生成不可变类 freezed class ConversationState with _$ConversationState { const factory ConversationState.initial() _Initial; const factory ConversationState.loading() _Loading; const factory ConversationState.streaming({ required ListMessage messages, // 完整的对话历史 required String currentDelta, // 当前正在接收的增量文本 }) _Streaming; const factory ConversationState.complete({ required ListMessage messages, }) _Complete; const factory ConversationState.error({ required String message, ListMessage? messages, }) _Error; } class Message { final String role; // user or assistant final String content; final DateTime timestamp; Message({required this.role, required this.content, required this.timestamp}); }使用freezed包可以让我们轻松创建不可变immutable的数据类并自带copyWith、值相等、toString等方法这在管理状态时非常安全且方便。3.2 实现视图模型Notifier接下来我们实现一个ConversationNotifier它继承自StateNotifierConversationState并负责管理整个对话的生命周期。import package:flutter_riverpod/flutter_riverpod.dart; import package:uuid/uuid.dart; class ConversationNotifier extends StateNotifierConversationState { ConversationNotifier(this._aiService) : super(const ConversationState.initial()); final AIService _aiService; StreamSubscriptionString? _streamSubscription; // 用于取消订阅 final ListMessage _messageHistory []; final String _currentAssistantMessageId const Uuid().v4(); // 为本次AI回复生成唯一ID Futurevoid sendMessage(String userInput) async { if (state is _Loading || state is _Streaming) { return; // 防止重复发送 } // 1. 添加用户消息到历史 _messageHistory.add(Message( role: user, content: userInput, timestamp: DateTime.now(), )); // 2. 进入Loading状态UI可以显示“正在思考”之类的指示 state const ConversationState.loading(); // 3. 添加一个初始为空的AI消息占位符到历史 _messageHistory.add(Message( role: assistant, content: , // 初始内容为空 timestamp: DateTime.now(), )); // 4. 进入Streaming状态并开始接收流 state ConversationState.streaming( messages: List.from(_messageHistory), currentDelta: , ); try { // 5. 发起流式请求并订阅 _streamSubscription _aiService .streamCompletion(prompt: userInput) .listen(_onDataReceived, onError: _onError, onDone: _onDone); } catch (e) { state ConversationState.error(message: 连接失败: $e, messages: _messageHistory); } } void _onDataReceived(String textDelta) { // 1. 更新当前增量文本 final lastMessageIndex _messageHistory.length - 1; final oldMessage _messageHistory[lastMessageIndex]; final newContent oldMessage.content textDelta; // 2. 更新历史中最后一条AI消息的内容 _messageHistory[lastMessageIndex] oldMessage.copyWith(content: newContent); // 3. 更新状态通知UI刷新 state ConversationState.streaming( messages: List.from(_messageHistory), currentDelta: textDelta, // 可以只传递增量UI用于特殊效果如打字机动画 ); } void _onError(Object error) { _streamSubscription?.cancel(); state ConversationState.error(message: 生成过程出错: $error, messages: _messageHistory); } void _onDone() { _streamSubscription?.cancel(); // 流式接收完毕转换为完成状态 state ConversationState.complete(messages: List.from(_messageHistory)); } // 提供手动停止生成的方法 void stopGeneration() { _streamSubscription?.cancel(); state ConversationState.complete(messages: _messageHistory); } override void dispose() { _streamSubscription?.cancel(); // 非常重要防止内存泄漏 super.dispose(); } }这个Notifier是大脑。它管理着消息历史协调着加载、流式接收、完成、错误等各种状态切换。_onDataReceived方法是关键它每次接收到一个文本片段就更新历史记录的最后一条消息并产生一个新的streaming状态通知UI更新。这里使用List.from(...)来创建历史列表的新副本这对于遵循不可变数据原则、确保Riverpod能正确检测到状态变化至关重要。4. UI层的魔法打造流畅的流式文本渲染体验UI层的目标是将视图模型提供的状态转化为用户能感知到的、流畅的交互。这里有两个核心挑战一是如何平滑地显示不断增长的文本二是如何实现“打字机”效果以增强流式体验。4.1 构建对话界面骨架我们首先构建一个基本的对话界面它监听ConversationNotifier的状态。class ConversationScreen extends ConsumerWidget { const ConversationScreen({super.key}); override Widget build(BuildContext context, WidgetRef ref) { final conversationState ref.watch(conversationNotifierProvider); final scrollController ScrollController(); return Scaffold( appBar: AppBar(title: const Text(AI对话)), body: Column( children: [ // 消息列表 Expanded( child: ListView.builder( controller: scrollController, padding: const EdgeInsets.all(8.0), itemCount: _getMessageCount(conversationState), itemBuilder: (context, index) { return _buildMessageItem(index, conversationState, ref); }, ), ), // 输入框和发送按钮 _buildInputArea(ref), ], ), ); } }4.2 关键流式消息项的构建_buildMessageItem是渲染的核心。对于已经完成的历史消息我们可以直接用TextWidget显示。但对于正在接收中的AI消息即ConversationState.streaming状态下的最后一条消息我们需要特殊处理。一个朴素的做法是直接在setState或状态更新时重建整个TextWidget。但对于长文本频繁重建整个文本块可能不够高效尤其是当文本包含复杂样式如Markdown时。更优的方案是使用StreamBuilder直接监听一个只包含当前增量文本的Stream或者使用AnimatedBuilder配合ValueNotifier。这里我分享一个在实践中效果很好的“混合方案”Widget _buildMessageItem(int index, ConversationState state, WidgetRef ref) { final messages state.messages; final message messages[index]; final isUser message.role user; final isLastMessage index messages.length - 1; final isStreaming state is _Streaming isLastMessage; return Container( margin: const EdgeInsets.symmetric(vertical: 4.0), alignment: isUser ? Alignment.centerRight : Alignment.centerLeft, child: Container( constraints: BoxConstraints(maxWidth: MediaQuery.of(context).size.width * 0.7), padding: const EdgeInsets.all(12.0), decoration: BoxDecoration( color: isUser ? Colors.blue[100] : Colors.grey[200], borderRadius: BorderRadius.circular(16.0), ), child: Column( crossAxisAlignment: CrossAxisAlignment.start, children: [ // 如果是正在流式接收的最后一条消息使用特殊的StreamingTextWidget if (isStreaming) StreamingTextWidget( key: ValueKey(message.id), // 使用唯一Key确保动画重置 fullText: message.content, stream: _getCurrentDeltaStream(ref), // 从Provider获取增量文本流 ) else SelectableText( message.content, style: Theme.of(context).textTheme.bodyMedium, ), const SizedBox(height: 4), Text( DateFormat(HH:mm).format(message.timestamp), style: Theme.of(context).textTheme.caption, ), ], ), ), ); } // 专门的Widget来处理流式文本显示和打字机动画 class StreamingTextWidget extends StatefulWidget { final String fullText; final StreamString stream; const StreamingTextWidget({super.key, required this.fullText, required this.stream}); override StateStreamingTextWidget createState() _StreamingTextWidgetState(); } class _StreamingTextWidgetState extends StateStreamingTextWidget with SingleTickerProviderStateMixin { final _displayText ValueNotifierString(); late final AnimationController _cursorController; override void initState() { super.initState(); _cursorController AnimationController( vsync: this, duration: const Duration(milliseconds: 500), )..repeat(reverse: true); // 光标闪烁动画 // 初始化显示文本 _displayText.value widget.fullText; // 监听外部传入的流更新显示文本 widget.stream.listen((delta) { _displayText.value delta; }); } override Widget build(BuildContext context) { return Row( mainAxisSize: MainAxisSize.min, crossAxisAlignment: CrossAxisAlignment.end, children: [ // 使用ValueListenableBuilder局部重建文本避免重建整个Widget树 ValueListenableBuilderString( valueListenable: _displayText, builder: (context, text, child) { return Expanded( child: SelectableText( text, style: Theme.of(context).textTheme.bodyMedium, ), ); }, ), const SizedBox(width: 2), // 闪烁的光标 AnimatedBuilder( animation: _cursorController, builder: (context, child) { return Opacity( opacity: _cursorController.value, child: Container( width: 2, height: 20, color: Colors.black, ), ); }, ), ], ); } override void dispose() { _cursorController.dispose(); super.dispose(); } }这个StreamingTextWidget的精髓在于ValueNotifierValueListenableBuilder我们将动态变化的文本存储在ValueNotifier中然后使用ValueListenableBuilder来监听它。ValueListenableBuilder只会重建其builder方法返回的Widget在这里就是SelectableText而不是整个StreamingTextWidget甚至整个消息气泡。这极大地提高了渲染效率。独立的光标动画使用AnimationController控制一个独立Widget的透明度来实现光标闪烁与文本更新逻辑解耦动画流畅。外部流监听在initState中监听传入的stream每当有新的文本增量delta到达就更新_displayText.value触发UI更新。4.3 自动滚动与性能优化当新消息到来或AI消息不断变长时我们需要自动滚动列表到底部。这应该在StreamingTextWidget的ValueListenableBuilder中或者在与conversationState关联的ListView.builder外层通过WidgetsBinding的addPostFrameCallback来实现以确保在UI帧渲染完成后执行滚动。// 在ConversationScreen的build方法中或在一个监听state变化的Listener中 void _scrollToBottom(ScrollController scrollController) { WidgetsBinding.instance.addPostFrameCallback((_) { if (scrollController.hasClients) { scrollController.animateTo( scrollController.position.maxScrollExtent, duration: const Duration(milliseconds: 300), curve: Curves.easeOut, ); } }); }关于性能还有一点至关重要对于很长的流式响应要避免在每次文本更新时都将完整的、不断变长的字符串传递给TextWidget进行布局计算。虽然Flutter的文本渲染性能很好但极端情况下仍可能造成界面卡顿。我们的ValueListenableBuilder方案已经优化了重建范围。更进一步可以考虑将超长文本分页或者使用AutomaticKeepAliveClientMixin来保存已滚出屏幕的复杂消息项的状态避免重复解析和布局。5. 进阶优化与实战避坑指南把基础功能跑通只是第一步要让这个功能真正达到“产品级”体验还需要处理一系列边界情况和进行深度优化。5.1 网络稳定性与错误处理流式连接天生比单次请求更脆弱。网络抖动、服务器中断、应用退到后台等都可能导致连接断开。心跳与超时虽然SSE协议本身有重连机制但在Dart客户端我们需要自己实现。可以在建立连接后启动一个定时器定期检查最后收到数据的时间。如果超过一定阈值如15秒则主动断开并尝试重连或者通知用户网络不稳定。后台处理当App进入后台大多数网络活动会被暂停。你需要根据产品需求决定策略是温和地中断生成并保存进度还是使用background_fetch之类的插件尝试保持连接通常对于非即时通讯场景中断并提示用户“连接已断开点击继续”是更合理的做法。错误状态细分不要只用一种“错误”状态。区分“网络错误”、“服务器错误5xx”、“内容过滤错误4xx”、“生成超时”等并在UI上给予用户明确的、可操作的反馈。5.2 对话历史管理与持久化一个完整的AI对话功能必然需要历史记录。我们需要将_messageHistory列表持久化到本地。推荐使用isar或hive这类高性能的本地数据库而不是简单的shared_preferences不适合存储大量结构化数据。在Notifier初始化时从数据库加载历史在每次对话状态变为complete或error时保存历史。注意对于未完成的流式消息通常不进行持久化除非要实现“草稿”功能。5.3 流式中断与重新生成用户有权在任何时候停止AI的“滔滔不绝”。我们在Notifier中已经提供了stopGeneration方法它取消StreamSubscription并将状态置为complete。调用它后当前这条不完整的AI消息会被视为最终消息保存下来。“重新生成”功能则稍微复杂一些。它意味着要删除上一条AI消息可能是不完整的然后用相同的用户问题再次发起请求。这要求我们的Notifier能处理消息的删除和替换而不是简单的追加。5.4 一个隐蔽的性能陷阱Stream的多次监听在Riverpod架构下一个常见的错误是在多个地方watch同一个由StreamProvider提供的流。默认情况下每次watch都会导致一个新的流订阅这意味着会发起一次新的网络请求这绝对是灾难性的。我们必须确保流是广播流并且被正确地共享。解决方案是使用StreamProvider的.autoDispose家族时格外小心或者更推荐的方式是不在UI层直接watch数据层的原始流。而是像我们之前设计的那样让ConversationNotifier作为唯一的数据消费者它内部监听数据流并将其转化为状态。UI只watch这个Notifier提供的状态。这样就保证了数据流只有一个订阅源。5.5 文本渲染的增强Markdown与代码高亮纯文本的AI回复是乏味的。大多数AI模型返回的答案都包含Markdown格式。我们需要在渲染时解析Markdown。可以使用flutter_markdown包但要注意其性能。对于流式文本频繁地解析和渲染整个Markdown文档是不可取的。一个折中的优化方案是在流式接收过程中先以纯文本形式显示但可以识别简单的换行和段落。当流式接收完成后再将完整的文本交给Markdown渲染引擎进行格式化渲染。对于代码块可以集成highlight这样的包进行语法高亮这能极大提升程序员用户的体验。6. 从功能到体验动画、音效与无障碍技术实现稳固后我们可以追求更极致的用户体验。打字机动画曲线上面实现的光标闪烁是基础。更高级的“打字机效果”是让文字逐个出现而不是一段段出现。这可以通过一个Animation来控制显示文本的长度并随着时间推移逐渐增加_displayText.value.substring(0, length)中的length值来实现。使用Curves.easeOut等缓动曲线会让动画更自然。音效反馈在收到新的文本片段时可以播放一个微弱的、短促的打字机音效但务必提供开关且不宜频繁播放。这能强化“AI正在为你思考”的感知。无障碍支持为动态更新的文本区域添加SemanticsWidget并设置liveRegion属性为LiveRegion.polite。这样屏幕阅读器如TalkBack/VoiceOver会在文本更新时自动朗读新增的内容让视障用户也能跟上AI的思考节奏。这是很多AI应用忽略但至关重要的细节。7. 测试策略如何验证流式交互测试流式UI比测试静态UI复杂得多。你需要模拟一个能按特定节奏发送数据块的“假”数据源。单元测试Notifier使用mocktail来模拟AIService让你可以精确控制何时发出数据、发出什么数据、何时抛出错误。然后验证你的ConversationNotifier在各种情况下正常流、中途错误、用户停止是否产生了正确的状态序列。Widget测试使用fake_async包来控制时间让你能在测试中“快进”动画。你可以构建StreamingTextWidget并模拟一个每100毫秒发送一个字的流然后验证UI是否正确更新光标动画是否运行。集成测试可以启动一个本地的模拟服务器使用shelf或aqueduct快速搭建一个能返回SSE的端点然后在真机或模拟器上运行完整的集成测试流程从输入到看到流式输出。整个项目走下来最大的体会是在Flutter中构建流式AI功能技术难点并不在于某个高深的算法而在于如何将异步数据流、响应式状态管理和细腻的UI动画有机地编织在一起形成一个稳定、流畅、可维护的整体。它考验的是开发者对Flutter响应式编程范式的理解深度以及对产品细节的打磨耐心。当你看到文字一个接一个平滑地出现在屏幕上光标在恰当的位置闪烁整个交互如德芙般丝滑时你就会觉得这些复杂的设计和优化都是值得的。这套架构不仅适用于聊天AI任何需要处理服务器推送、实时数据更新的场景如股票行情、体育赛事比分、协同编辑提示都可以从中获得借鉴。