From 55fc24e9334945f0ec91306fdbf553de22c34e5e Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Mon, 12 May 2025 16:21:53 -0400
Subject: [PATCH] AWS Nova Sonic service - aggregate user transcription text;
 it was fragmented across many conversation history messages before

---
 src/pipecat/services/aws_nova_sonic/aws.py    | 44 ++++++++++++++++++-
 .../services/aws_nova_sonic/context.py        | 14 +++++-
 2 files changed, 54 insertions(+), 4 deletions(-)

diff --git a/src/pipecat/services/aws_nova_sonic/aws.py b/src/pipecat/services/aws_nova_sonic/aws.py
index 410481065..0e474fb89 100644
--- a/src/pipecat/services/aws_nova_sonic/aws.py
+++ b/src/pipecat/services/aws_nova_sonic/aws.py
@@ -26,6 +26,7 @@ from pipecat.frames.frames import (
     EndFrame,
     Frame,
     InputAudioRawFrame,
+    InterimTranscriptionFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMTextFrame,
@@ -757,6 +758,7 @@ class AWSNovaSonicLLMService(LLMService):
                 if not self._assistant_is_responding:
                     # The assistant has started responding.
                     self._assistant_is_responding = True
+                    await self._report_user_transcription_ended()  # Consider user turn over
                     await self._report_assistant_response_started()
 
     async def _handle_text_output_event(self, event_json):
@@ -790,6 +792,9 @@ class AWSNovaSonicLLMService(LLMService):
         if not self._content_being_received or not self._context:  # should never happen
             return
 
+        # Consider user turn over
+        await self._report_user_transcription_ended()
+
         # Get tool use details
         tool_use = event_json["toolUse"]
         function_name = tool_use["toolName"]
@@ -837,6 +842,14 @@ class AWSNovaSonicLLMService(LLMService):
     async def _handle_completion_end_event(self, event_json):
         pass
 
+    #
+    # assistant response reporting
+    #
+    # 1. Started
+    # 2. Text added
+    # 3. Ended
+    #
+
     async def _report_assistant_response_started(self):
         logger.debug("Assistant response started")
 
@@ -885,6 +898,15 @@ class AWSNovaSonicLLMService(LLMService):
         # For an explanation of this hack, see _report_assistant_response_text_added.
         self._context.flush_aggregated_assistant_text()
 
+    #
+    # user transcription reporting
+    #
+    # 1. Text added
+    # 2. Ended
+    #
+    # Note: "started" does not need to be reported
+    #
+
     async def _report_user_transcription_text_added(self, text):
         if not self._context:  # should never happen
             return
@@ -893,12 +915,30 @@ class AWSNovaSonicLLMService(LLMService):
 
         # Manually add new user transcription text to context.
         # We can't rely on the user context aggregator to do this since it's upstream from the LLM.
-        self._context.add_user_transcription_text(text)
+        self._context.buffer_user_text(text)
 
         # Report that some new user transcription text is available.
         if self._send_transcription_frames:
             await self.push_frame(
-                TranscriptionFrame(text=text, user_id="", timestamp=time_now_iso8601())
+                InterimTranscriptionFrame(text=text, user_id="", timestamp=time_now_iso8601())
+            )
+
+    async def _report_user_transcription_ended(self):
+        if not self._context:  # should never happen
+            return
+
+        # Manually add user transcription to context (if any has been buffered).
+        # We can't rely on the user context aggregator to do this since it's upstream from the LLM.
+        transcription = self._context.flush_aggregated_user_text()
+
+        if not transcription:
+            return
+
+        logger.debug(f"User transcription ended")
+
+        if self._send_transcription_frames:
+            await self.push_frame(
+                TranscriptionFrame(text=transcription, user_id="", timestamp=time_now_iso8601())
             )
 
     #
diff --git a/src/pipecat/services/aws_nova_sonic/context.py b/src/pipecat/services/aws_nova_sonic/context.py
index 561ae53db..95f330f61 100644
--- a/src/pipecat/services/aws_nova_sonic/context.py
+++ b/src/pipecat/services/aws_nova_sonic/context.py
@@ -60,6 +60,7 @@ class AWSNovaSonicLLMContext(OpenAILLMContext):
 
     def __setup_local(self, system_instruction: str = ""):
         self._assistant_text = ""
+        self._user_text = ""
         self._system_instruction = system_instruction
 
     @staticmethod
@@ -129,13 +130,22 @@ class AWSNovaSonicLLMContext(OpenAILLMContext):
         # NOTE: we're ignoring messages with role "tool" since they can't be loaded into AWS Nova
         # Sonic conversation history
 
-    def add_user_transcription_text(self, text):
+    def buffer_user_text(self, text):
+        self._user_text += f" {text}" if self._user_text else text
+        # logger.debug(f"User text buffered: {self._user_text}")
+
+    def flush_aggregated_user_text(self) -> str:
+        if not self._user_text:
+            return ""
+        user_text = self._user_text
         message = {
             "role": "user",
-            "content": [{"type": "text", "text": text}],
+            "content": [{"type": "text", "text": user_text}],
         }
+        self._user_text = ""
         self.add_message(message)
         # logger.debug(f"Context updated (user): {self.get_messages_for_logging()}")
+        return user_text
 
     def buffer_assistant_text(self, text):
         self._assistant_text += text