add LLMFullResponseStartFrame/LLMFullResponseEndFrame

examples/foundational/05-sync-speech-and-image.py

@@ -13,12 +13,12 @@ from dataclasses import dataclass
 
 from pipecat.frames.frames import (
     AppFrame,
+    EndFrame,
     Frame,
     ImageRawFrame,
-    TextFrame,
-    EndFrame,
+    LLMFullResponseStartFrame,
     LLMMessagesFrame,
-    LLMResponseStartFrame,
+    TextFrame
 )
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
@@ -64,7 +64,7 @@ class MonthPrepender(FrameProcessor):
         elif self.prepend_to_next_text_frame and isinstance(frame, TextFrame):
             await self.push_frame(TextFrame(f"{self.most_recent_month}: {frame.text}"))
             self.prepend_to_next_text_frame = False
-        elif isinstance(frame, LLMResponseStartFrame):
+        elif isinstance(frame, LLMFullResponseStartFrame):
             self.prepend_to_next_text_frame = True
             await self.push_frame(frame)
         else:
@@ -105,7 +105,7 @@ async def main(room_url):
 
         gated_aggregator = GatedAggregator(
             gate_open_fn=lambda frame: isinstance(frame, ImageRawFrame),
-            gate_close_fn=lambda frame: isinstance(frame, LLMResponseStartFrame),
+            gate_close_fn=lambda frame: isinstance(frame, LLMFullResponseStartFrame),
             start_open=False
         )
 
@@ -114,14 +114,14 @@ async def main(room_url):
         llm_full_response_aggregator = LLMFullResponseAggregator()
 
         pipeline = Pipeline([
-            llm,
-            sentence_aggregator,
-            ParallelTask(
-                [month_prepender, tts],
-                [llm_full_response_aggregator, imagegen]
+            llm,                     # LLM
+            sentence_aggregator,     # Aggregates LLM output into full sentences
+            ParallelTask(            # Run pipelines in parallel aggregating the result
+                [month_prepender, tts],                   # Create "Month: sentence" and output audio
+                [llm_full_response_aggregator, imagegen]  # Aggregate full LLM response
             ),
-            gated_aggregator,
-            transport.output()
+            gated_aggregator,        # Queues everything until an image is available
+            transport.output()       # Transport output
         ])
 
         frames = []

examples/foundational/07-interruptible.py

@@ -66,12 +66,12 @@ async def main(room_url: str, token):
         tma_out = LLMAssistantResponseAggregator(messages)
 
         pipeline = Pipeline([
-            transport.input(),
-            tma_in,
-            llm,
-            tts,
-            transport.output(),
-            tma_out
+            transport.input(),   # Transport user input
+            tma_in,              # User responses
+            llm,                 # LLM
+            tts,                 # TTS
+            transport.output(),  # Transport bot output
+            tma_out              # Assistant spoken responses
         ])
 
         task = PipelineTask(pipeline, allow_interruptions=True)

src/pipecat/frames/frames.py

@@ -260,6 +260,20 @@ class EndFrame(ControlFrame):
     pass
 
 
+@dataclass
+class LLMFullResponseStartFrame(ControlFrame):
+    """Used to indicate the beginning of a full LLM response. Following
+    LLMResponseStartFrame, TextFrame and LLMResponseEndFrame for each sentence
+    until a LLMFullResponseEndFrame."""
+    pass
+
+
+@dataclass
+class LLMFullResponseEndFrame(ControlFrame):
+    """Indicates the end of a full LLM response."""
+    pass
+
+
 @dataclass
 class LLMResponseStartFrame(ControlFrame):
     """Used to indicate the beginning of an LLM response. Following TextFrames

src/pipecat/processors/aggregators/llm_response.py

@@ -10,6 +10,7 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.frames.frames import (
     Frame,
     InterimTranscriptionFrame,
+    LLMFullResponseEndFrame,
     LLMMessagesFrame,
     LLMResponseStartFrame,
     TextFrame,
@@ -182,7 +183,7 @@ class LLMFullResponseAggregator(FrameProcessor):
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         if isinstance(frame, TextFrame):
             self._aggregation += frame.text
-        elif isinstance(frame, LLMResponseEndFrame):
+        elif isinstance(frame, LLMFullResponseEndFrame):
             await self.push_frame(TextFrame(self._aggregation))
             await self.push_frame(frame)
             self._aggregation = ""

src/pipecat/services/openai.py

@@ -16,6 +16,8 @@ from typing import AsyncGenerator, List, Literal
 from pipecat.frames.frames import (
     ErrorFrame,
     Frame,
+    LLMFullResponseEndFrame,
+    LLMFullResponseStartFrame,
     LLMMessagesFrame,
     LLMResponseEndFrame,
     LLMResponseStartFrame,
@@ -104,6 +106,8 @@ class BaseOpenAILLMService(LLMService):
             await self._stream_chat_completions(context)
         )
 
+        await self.push_frame(LLMFullResponseStartFrame())
+
         async for chunk in chunk_stream:
             if len(chunk.choices) == 0:
                 continue
@@ -134,6 +138,8 @@ class BaseOpenAILLMService(LLMService):
                 await self.push_frame(TextFrame(chunk.choices[0].delta.content))
                 await self.push_frame(LLMResponseEndFrame())
 
+        await self.push_frame(LLMFullResponseEndFrame())
+
         # 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: