From c5d54d06bba7c92c524dd7d171130f0e33ec4140 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 2 Jul 2025 11:12:49 -0700 Subject: [PATCH] add run_llm to LLMMessagesAppendFrame and LLMMessagesUpdateFrame --- CHANGELOG.md | 7 ++++- src/pipecat/frames/frames.py | 4 +++ .../processors/aggregators/llm_response.py | 28 ++++++++++++++++--- 3 files changed, 34 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index fcced9255..b34a057b8 100644 --- a/CHANGELOG.md +++ b/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). diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 59e862818..c7d7b4c75 100644 --- a/src/pipecat/frames/frames.py +++ b/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 diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 85c4a3b06..99834447f 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/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