Merge pull request #2119 from pipecat-ai/aleix/llm-messages-append-update-run-llm

add run_llm to LLMMessagesAppendFrame and LLMMessagesUpdateFrame

CHANGELOG.md

@@ -9,7 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
-- Added a new `SOXRStreamAudioResampler` for processing audio in chunks or streams.
+- Added `run_llm` field to `LLMMessagesAppendFrame` and `LLMMessagesUpdateFrame`
+  frames. If true, a context frame will be pushed triggering the LLM to respond.
+
+- Added a new `SOXRStreamAudioResampler` for processing audio in chunks or
+  streams. If you write your own processor and need to use an audio resampler,
+  use the new `create_stream_resampler()`.
 
 - Added new `DailyParams.audio_in_user_tracks` to allow receiving one track per
   user (default) or a single track from the room (all participants mixed).

src/pipecat/frames/frames.py

@@ -484,9 +484,11 @@ class LLMMessagesAppendFrame(DataFrame):
 
     Parameters:
         messages: List of message dictionaries to append.
+        run_llm: Whether the context update should be sent to the LLM.
     """
 
     messages: List[dict]
+    run_llm: Optional[bool] = None
 
 
 @dataclass
@@ -499,9 +501,11 @@ class LLMMessagesUpdateFrame(DataFrame):
 
     Parameters:
         messages: List of message dictionaries to replace current context.
+        run_llm: Whether the context update should be sent to the LLM.
     """
 
     messages: List[dict]
+    run_llm: Optional[bool] = None
 
 
 @dataclass

src/pipecat/processors/aggregators/llm_response.py

@@ -470,9 +470,9 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
         elif isinstance(frame, InterimTranscriptionFrame):
             await self._handle_interim_transcription(frame)
         elif isinstance(frame, LLMMessagesAppendFrame):
-            self.add_messages(frame.messages)
+            await self._handle_llm_messages_append(frame)
         elif isinstance(frame, LLMMessagesUpdateFrame):
-            self.set_messages(frame.messages)
+            await self._handle_llm_messages_update(frame)
         elif isinstance(frame, LLMSetToolsFrame):
             self.set_tools(frame.tools)
         elif isinstance(frame, LLMSetToolChoiceFrame):
@@ -544,6 +544,16 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
     async def _cancel(self, frame: CancelFrame):
         await self._cancel_aggregation_task()
 
+    async def _handle_llm_messages_append(self, frame: LLMMessagesAppendFrame):
+        self.add_messages(frame.messages)
+        if frame.run_llm:
+            await self.push_context_frame()
+
+    async def _handle_llm_messages_update(self, frame: LLMMessagesUpdateFrame):
+        self.set_messages(frame.messages)
+        if frame.run_llm:
+            await self.push_context_frame()
+
     async def _handle_input_audio(self, frame: InputAudioRawFrame):
         for s in self.interruption_strategies:
             await s.append_audio(frame.audio, frame.sample_rate)
@@ -767,9 +777,9 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
         elif isinstance(frame, TextFrame):
             await self._handle_text(frame)
         elif isinstance(frame, LLMMessagesAppendFrame):
-            self.add_messages(frame.messages)
+            await self._handle_llm_messages_append(frame)
         elif isinstance(frame, LLMMessagesUpdateFrame):
-            self.set_messages(frame.messages)
+            await self._handle_llm_messages_update(frame)
         elif isinstance(frame, LLMSetToolsFrame):
             self.set_tools(frame.tools)
         elif isinstance(frame, LLMSetToolChoiceFrame):
@@ -808,6 +818,16 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
         timestamp_frame = OpenAILLMContextAssistantTimestampFrame(timestamp=time_now_iso8601())
         await self.push_frame(timestamp_frame)
 
+    async def _handle_llm_messages_append(self, frame: LLMMessagesAppendFrame):
+        self.add_messages(frame.messages)
+        if frame.run_llm:
+            await self.push_context_frame(FrameDirection.UPSTREAM)
+
+    async def _handle_llm_messages_update(self, frame: LLMMessagesUpdateFrame):
+        self.set_messages(frame.messages)
+        if frame.run_llm:
+            await self.push_context_frame(FrameDirection.UPSTREAM)
+
     async def _handle_interruptions(self, frame: StartInterruptionFrame):
         await self.push_aggregation()
         self._started = 0