diff --git a/examples/foundational/05-sync-speech-and-image.py b/examples/foundational/05-sync-speech-and-image.py index 747eccb63..60dd50d07 100644 --- a/examples/foundational/05-sync-speech-and-image.py +++ b/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 = [] diff --git a/examples/foundational/07-interruptible.py b/examples/foundational/07-interruptible.py index 2349c4303..ae337f7f6 100644 --- a/examples/foundational/07-interruptible.py +++ b/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) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 9d5b15b36..8eb32664c 100644 --- a/src/pipecat/frames/frames.py +++ b/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 diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 5f23a9ed1..3b9c07fe6 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/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 = "" diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index 3e2f2ae91..56224e8fe 100644 --- a/src/pipecat/services/openai.py +++ b/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: