services(openai): send LLMResponseStartFrame for each completion

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.
 

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)
 

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}")

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):

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):