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任何需要處理服務器推送、實時數據更新的場景如股票行情、體育賽事比分、協同編輯提示都可以從中獲得借鑒。