try realtime input

Author: cynthiajoan

packages/firebase_vertexai/firebase_vertexai/example/lib/pages/bidi_page.dart

@@ -221,7 +221,8 @@ class _BidiPageState extends State<BidiPage> {
     await _audioRecorder.stopRecording();
     var audioPrompt = await _audioRecorder.getAudioBytes(fromFile: true);
     //await _streamAudioChunks(audioPrompt, 'audio/pcm');
-    await _sendAudioPrompt(audioPrompt);
+    //await _sendAudioPrompt(audioPrompt);
+    await _sendAudioRealtime(audioPrompt);
   }
 
   List<Uint8List> _splitIntoChunks(Uint8List audioData, int chunkSize) {
@@ -242,9 +243,16 @@ class _BidiPageState extends State<BidiPage> {
     });
     final chunks = _splitIntoChunks(audioData, 1024);
 
-    final streamController = StreamController<Uint8List>();
+    final streamController = StreamController<InlineDataPart>();
     for (var chunk in chunks) {
-      streamController.add(chunk);
+      if (identical(chunk, chunks.last)) {
+        final lastData =
+            InlineDataPart('audio/pcm', chunk, willContinue: false);
+        streamController.add(lastData);
+      } else {
+        final data = InlineDataPart('audio/pcm', chunk, willContinue: true);
+        streamController.add(data);
+      }
     }
     streamController.close();
     print('streamController has stream closed');
@@ -261,6 +269,32 @@ class _BidiPageState extends State<BidiPage> {
     });
   }
 
+  Future<void> _sendAudioRealtime(Uint8List audio) async {
+    setState(() {
+      _loading = true;
+    });
+    final chunks = _splitIntoChunks(audio, 1024);
+
+    final media_chunks = <InlineDataPart>[];
+    for (var chunk in chunks) {
+      if (identical(chunk, chunks.last)) {
+        final lastData =
+            InlineDataPart('audio/pcm', chunk, willContinue: false);
+        media_chunks.add(lastData);
+      } else {
+        final data = InlineDataPart('audio/pcm', chunk, willContinue: true);
+        media_chunks.add(data);
+      }
+    }
+    await _session!.stream(mediaChunks: media_chunks);
+    print('Stream realtime audio chunk to server in one request');
+
+    await _handle_response_audio();
+    setState(() {
+      _loading = false;
+    });
+  }
+
   Future<void> _sendAudioPrompt(Uint8List audio) async {
     setState(() {
       _loading = true;

packages/firebase_vertexai/firebase_vertexai/lib/src/content.dart

@@ -124,17 +124,23 @@ final class TextPart implements Part {
 /// A [Part] with the byte content of a file.
 final class InlineDataPart implements Part {
   /// Constructor
-  InlineDataPart(this.mimeType, this.bytes);
+  InlineDataPart(this.mimeType, this.bytes, {this.willContinue});
 
   /// File type of the [InlineDataPart].
   /// https://cloud.google.com/vertex-ai/generative-ai/docs/multimodal/send-multimodal-prompts#media_requirements
   final String mimeType;
 
   /// Data contents in bytes.
   final Uint8List bytes;
+
+  final bool? willContinue;
   @override
   Object toJson() => {
-        'inlineData': {'data': base64Encode(bytes), 'mimeType': mimeType}
+        'inlineData': {
+          'data': base64Encode(bytes),
+          'mimeType': mimeType,
+          if (willContinue != null) 'willContinue': willContinue,
+        }
       };
 }
 

packages/firebase_vertexai/firebase_vertexai/lib/src/live.dart

@@ -39,15 +39,25 @@ class AsyncSession {
         ? LiveClientContent(turns: [input], turnComplete: turnComplete)
         : LiveClientContent(turnComplete: turnComplete);
     var clientJson = jsonEncode(clientMessage.toJson());
-    print(clientJson);
+    print('Sending $clientJson');
+    _ws.sink.add(clientJson);
+  }
+
+  Future<void> stream({
+    required List<InlineDataPart> mediaChunks,
+  }) async {
+    // var clientMessage = _parseClientMessage(input, endOfTurn);
+    var clientMessage = LiveClientRealtimeInput(mediaChunks: mediaChunks);
+    var clientJson = jsonEncode(clientMessage.toJson());
+    print('Streaming $clientJson');
     _ws.sink.add(clientJson);
   }
 
   Stream<LiveServerMessage> receive() async* {
     await for (var message in _ws.stream) {
       var jsonString = utf8.decode(message);
       var response = json.decode(jsonString);
-      //print(response);
+      print(response);
       Map<String, dynamic> responseDict;
 
       responseDict = _LiveServerMessageFromVertex(response);
@@ -63,7 +73,7 @@ class AsyncSession {
   }
 
   Stream<LiveServerMessage> startStream({
-    required Stream<Uint8List> stream,
+    required Stream<InlineDataPart> stream,
     required String mimeType,
   }) async* {
     print('beginning of startStream');
@@ -90,7 +100,7 @@ class AsyncSession {
     // }
 
     // Wait for the send loop to complete or the websocket to close.
-    // await Future.any([completer.future, _ws.stream.isEmpty]);
+    await Future.any([completer.future]);
 
     // Close the websocket if it's not already closed.
     // if (_ws.closeCode == null) {
@@ -99,19 +109,16 @@ class AsyncSession {
   }
 
   Future<void> _sendLoop(
-    Stream<Uint8List> dataStream,
+    Stream<InlineDataPart> dataStream,
     String mimeType,
     Completer completer,
   ) async {
     try {
       print('start _sendLoop');
       await for (final data in dataStream) {
-        var input = Content.inlineData(
-            mimeType, data); // No need to convert to Uint8List
-
-        print('send audio data with size ${data.length}');
+        print('send audio data with size ${data.bytes.length}');
 
-        await send(input: input);
+        await stream(mediaChunks: [data]);
         // Give a chance for the receive loop to process responses.
         await Future.delayed(Duration.zero);
       }

packages/firebase_vertexai/firebase_vertexai/lib/src/live_api.dart

@@ -170,7 +170,9 @@ class LiveClientRealtimeInput {
   final List<InlineDataPart>? mediaChunks;
 
   Map<String, dynamic> toJson() => {
-        'mediaChunks': mediaChunks?.map((e) => e.toJson()).toList(),
+        'realtime_input': {
+          'media_chunks': mediaChunks?.map((e) => e.toJson()).toList(),
+        },
       };
 }