From 2b8f1c4cda21d9e405ad78cc60f77e3e875395e6 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Fri, 17 May 2024 17:47:33 -0700
Subject: [PATCH] services(openai): send LLMResponseStartFrame for each
 completion

---
 CHANGELOG.md                              | 4 +++-
 examples/foundational/07-interruptible.py | 3 ++-
 src/pipecat/processors/frame_processor.py | 3 +--
 src/pipecat/services/ai_services.py       | 4 +++-
 src/pipecat/services/openai.py            | 6 ++----
 5 files changed, 11 insertions(+), 9 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 69c9e9091..a26ebfbb3 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,7 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
-- Added initial interruptions support.
+- Added initial interruptions support. The assitant contexts (or aggregators)
+  should now be placed after the output transport. This way, only the completed
+  spoken context is added to the assistant context.
 
 - Added `VADParams` so you can control voice confidence level and others.
 
diff --git a/examples/foundational/07-interruptible.py b/examples/foundational/07-interruptible.py
index 71f9a22ab..2349c4303 100644
--- a/examples/foundational/07-interruptible.py
+++ b/examples/foundational/07-interruptible.py
@@ -71,7 +71,8 @@ async def main(room_url: str, token):
             llm,
             tts,
             transport.output(),
-            tma_out])
+            tma_out
+        ])
 
         task = PipelineTask(pipeline, allow_interruptions=True)
 
diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py
index a7d330680..a79352f7a 100644
--- a/src/pipecat/processors/frame_processor.py
+++ b/src/pipecat/processors/frame_processor.py
@@ -47,8 +47,7 @@ class FrameProcessor:
 
     async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
         if direction == FrameDirection.DOWNSTREAM and self._next:
-            if not isinstance(frame, AudioRawFrame):
-                logger.trace(f"Pushing {frame} from {self} to {self._next}")
+            logger.trace(f"Pushing {frame} from {self} to {self._next}")
             await self._next.process_frame(frame, direction)
         elif direction == FrameDirection.UPSTREAM and self._prev:
             logger.trace(f"Pushing {frame} upstream from {self} to {self._prev}")
diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py
index bb697a84f..52d62b7c8 100644
--- a/src/pipecat/services/ai_services.py
+++ b/src/pipecat/services/ai_services.py
@@ -74,10 +74,12 @@ class TTSService(AIService):
             await self._push_tts_frames(text)
 
     async def _push_tts_frames(self, text: str):
-        await self.push_frame(TextFrame(text))
         await self.push_frame(TTSStartedFrame())
         await self.process_generator(self.run_tts(text))
         await self.push_frame(TTSStoppedFrame())
+        # We send the original text after the audio. This way, if we are
+        # interrupted, the text is not added to the assistant context.
+        await self.push_frame(TextFrame(text))
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         if isinstance(frame, TextFrame):
diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py
index b29f3aaec..3e2f2ae91 100644
--- a/src/pipecat/services/openai.py
+++ b/src/pipecat/services/openai.py
@@ -100,8 +100,6 @@ class BaseOpenAILLMService(LLMService):
         function_name = ""
         arguments = ""
 
-        await self.push_frame(LLMResponseStartFrame())
-
         chunk_stream: AsyncStream[ChatCompletionChunk] = (
             await self._stream_chat_completions(context)
         )
@@ -132,15 +130,15 @@ class BaseOpenAILLMService(LLMService):
                     # completes
                     arguments += tool_call.function.arguments
             elif chunk.choices[0].delta.content:
+                await self.push_frame(LLMResponseStartFrame())
                 await self.push_frame(TextFrame(chunk.choices[0].delta.content))
+                await self.push_frame(LLMResponseEndFrame())
 
         # if we got a function name and arguments, yield the frame with all the info so
         # frame consumers can take action based on the function call.
         # if function_name and arguments:
         #     yield LLMFunctionCallFrame(function_name=function_name, arguments=arguments)
 
-        await self.push_frame(LLMResponseEndFrame())
-
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         context = None
         if isinstance(frame, OpenAILLMContextFrame):