diff --git a/examples/foundational/55zl-update-settings-openai-realtime.py b/examples/foundational/55zl-update-settings-openai-realtime.py index 90663a95c..9c18d528e 100644 --- a/examples/foundational/55zl-update-settings-openai-realtime.py +++ b/examples/foundational/55zl-update-settings-openai-realtime.py @@ -22,11 +22,11 @@ from pipecat.processors.aggregators.llm_response_universal import ( ) from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport +from pipecat.services.openai.realtime import events from pipecat.services.openai.realtime.llm import ( OpenAIRealtimeLLMService, OpenAIRealtimeLLMSettings, ) -from pipecat.services.openai_realtime import events from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams diff --git a/examples/foundational/55zo-update-settings-grok-realtime.py b/examples/foundational/55zo-update-settings-grok-realtime.py index 567eeae2b..7d7370f7b 100644 --- a/examples/foundational/55zo-update-settings-grok-realtime.py +++ b/examples/foundational/55zo-update-settings-grok-realtime.py @@ -16,9 +16,13 @@ from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.llm_context import LLMContext -from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair +from pipecat.processors.aggregators.llm_response_universal import ( + AssistantTurnStoppedMessage, + LLMContextAggregatorPair, +) from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport +from pipecat.services.grok.realtime import events from pipecat.services.grok.realtime.llm import ( GrokRealtimeLLMService, GrokRealtimeLLMSettings, @@ -51,7 +55,7 @@ transport_params = { async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Starting bot") - llm = GrokRealtimeLLMService(api_key=os.getenv("XAI_API_KEY")) + llm = GrokRealtimeLLMService(api_key=os.getenv("GROK_API_KEY")) messages = [ { @@ -82,15 +86,25 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, ) + @assistant_aggregator.event_handler("on_assistant_turn_stopped") + async def on_assistant_turn_stopped(aggregator, message: AssistantTurnStoppedMessage): + timestamp = f"[{message.timestamp}] " if message.timestamp else "" + line = f"{timestamp}assistant: {message.content}" + logger.info(f"Transcript: {line}") + @transport.event_handler("on_client_connected") async def on_client_connected(transport, client): logger.info(f"Client connected") await task.queue_frames([LLMRunFrame()]) await asyncio.sleep(10) - logger.info("Updating Grok Realtime LLM settings: temperature=0.1") + logger.info("Updating Grok Realtime LLM settings: voice='Rex'") await task.queue_frame( - LLMUpdateSettingsFrame(update=GrokRealtimeLLMSettings(temperature=0.1)) + LLMUpdateSettingsFrame( + update=GrokRealtimeLLMSettings( + session_properties=events.SessionProperties(voice="Rex") + ) + ) ) @transport.event_handler("on_client_disconnected") diff --git a/src/pipecat/services/grok/realtime/llm.py b/src/pipecat/services/grok/realtime/llm.py index f31769774..14c93c94a 100644 --- a/src/pipecat/services/grok/realtime/llm.py +++ b/src/pipecat/services/grok/realtime/llm.py @@ -56,7 +56,7 @@ from pipecat.processors.aggregators.llm_response_universal import ( from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.processors.frame_processor import FrameDirection from pipecat.services.llm_service import FunctionCallFromLLM, LLMService -from pipecat.services.settings import NOT_GIVEN, LLMSettings +from pipecat.services.settings import NOT_GIVEN, LLMSettings, _NotGiven from pipecat.utils.time import time_now_iso8601 from . import events @@ -94,7 +94,9 @@ class GrokRealtimeLLMSettings(LLMSettings): session_properties: Grok Realtime session configuration. """ - session_properties: Any = field(default_factory=lambda: NOT_GIVEN) + session_properties: events.SessionProperties | _NotGiven = field( + default_factory=lambda: NOT_GIVEN + ) class GrokRealtimeLLMService(LLMService): @@ -294,6 +296,27 @@ class GrokRealtimeLLMService(LLMService): # Standard AIService frame handling # + def _ensure_audio_config(self, input_sample_rate: int, output_sample_rate: int): + """Ensure session_properties.audio has input and output configs. + + Fills in any missing audio configuration using the given sample rates. + + Args: + input_sample_rate: Sample rate for audio input (Hz). + output_sample_rate: Sample rate for audio output (Hz). + """ + props = self._settings.session_properties + if not props.audio: + props.audio = events.AudioConfiguration() + if not props.audio.input: + props.audio.input = events.AudioInput( + format=events.PCMAudioFormat(rate=input_sample_rate) + ) + if not props.audio.output: + props.audio.output = events.AudioOutput( + format=events.PCMAudioFormat(rate=output_sample_rate) + ) + async def start(self, frame: StartFrame): """Start the service and establish WebSocket connection. @@ -301,23 +324,7 @@ class GrokRealtimeLLMService(LLMService): frame: The start frame triggering service initialization. """ await super().start(frame) - - # Ensure audio configuration exists with both input and output - if not self._settings.session_properties.audio: - self._settings.session_properties.audio = events.AudioConfiguration() - - # Fill in missing input configuration - if not self._settings.session_properties.audio.input: - self._settings.session_properties.audio.input = events.AudioInput( - format=events.PCMAudioFormat(rate=frame.audio_in_sample_rate) - ) - - # Fill in missing output configuration - if not self._settings.session_properties.audio.output: - self._settings.session_properties.audio.output = events.AudioOutput( - format=events.PCMAudioFormat(rate=frame.audio_out_sample_rate) - ) - + self._ensure_audio_config(frame.audio_in_sample_rate, frame.audio_out_sample_rate) await self._connect() async def stop(self, frame: EndFrame): @@ -458,9 +465,23 @@ class GrokRealtimeLLMService(LLMService): async def _update_settings(self, update): """Apply a settings update, sending a session update if needed.""" + # Capture current sample rates before the update replaces them. + input_rate = self._get_configured_sample_rate("input") + output_rate = self._get_configured_sample_rate("output") + changed = await super()._update_settings(update) + if "session_properties" in changed: + if input_rate and output_rate: + self._ensure_audio_config(input_rate, output_rate) + else: + logger.warning( + "Attempting to apply session properties update without configured sample rates. " + "Audio configuration may be incomplete." + ) await self._send_session_update() + + self._warn_unhandled_updated_settings(changed.keys() - {"session_properties"}) return changed async def _send_session_update(self):