diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index decb76d75..17ab94695 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -1439,6 +1439,27 @@ class STTMetadataFrame(ServiceMetadataFrame): ttfs_p99_latency: float +@dataclass +class RealtimeServiceMetadataFrame(ServiceMetadataFrame): + """Metadata announcing a realtime (speech-to-speech) LLM service. + + Broadcast by realtime LLM services at pipeline start so downstream + processors — notably ``LLMContextAggregatorPair`` — can detect that + a realtime service is in the pipeline. The aggregator uses this to + surface a one-time recommendation to opt in to + ``RealtimeServiceModeConfig`` when it hasn't been configured. + + Parameters: + emits_user_turn_frames: Whether this service emits + ``UserStartedSpeakingFrame`` / ``UserStoppedSpeakingFrame`` + from server-side turn signals. False for services with no + server-side turn signals (e.g. Gemini Live, AWS Nova Sonic, + Ultravox). + """ + + emits_user_turn_frames: bool = True + + @dataclass class ServiceSwitcherRequestMetadataFrame(ControlFrame): """Request a service to re-emit its metadata frames. diff --git a/src/pipecat/services/aws/nova_sonic/llm.py b/src/pipecat/services/aws/nova_sonic/llm.py index 4aeb5c848..2e6aab9ca 100644 --- a/src/pipecat/services/aws/nova_sonic/llm.py +++ b/src/pipecat/services/aws/nova_sonic/llm.py @@ -56,7 +56,7 @@ from pipecat.services.aws.nova_sonic.session_continuation import ( SessionContinuationHelper, SessionContinuationParams, ) -from pipecat.services.llm_service import LLMService +from pipecat.services.llm_service import LLMService, RealtimeServiceInfo from pipecat.services.settings import NOT_GIVEN, LLMSettings, _NotGiven, assert_given from pipecat.utils.time import time_now_iso8601 @@ -249,6 +249,10 @@ class AWSNovaSonicLLMService(LLMService[AWSNovaSonicLLMAdapter]): # Override the default adapter to use the AWSNovaSonicLLMAdapter one adapter_class = AWSNovaSonicLLMAdapter + # Realtime (speech-to-speech) service. Does NOT emit + # UserStarted/StoppedSpeakingFrame from server-side turn signals. + _realtime_service_info = RealtimeServiceInfo(emits_user_turn_frames=False) + def __init__( self, *, diff --git a/src/pipecat/services/google/gemini_live/llm.py b/src/pipecat/services/google/gemini_live/llm.py index 7194127b1..80d722d29 100644 --- a/src/pipecat/services/google/gemini_live/llm.py +++ b/src/pipecat/services/google/gemini_live/llm.py @@ -62,7 +62,7 @@ from pipecat.processors.aggregators.llm_context import LLMContext, LLMSpecificMe from pipecat.processors.frame_processor import FrameDirection from pipecat.services.google.frames import LLMSearchOrigin, LLMSearchResponseFrame, LLMSearchResult from pipecat.services.google.utils import update_google_client_http_options -from pipecat.services.llm_service import FunctionCallFromLLM, LLMService +from pipecat.services.llm_service import FunctionCallFromLLM, LLMService, RealtimeServiceInfo from pipecat.services.settings import NOT_GIVEN, LLMSettings, _NotGiven, assert_given from pipecat.transcriptions.language import Language, resolve_language from pipecat.utils.string import match_endofsentence @@ -369,6 +369,11 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]): # Overriding the default adapter to use the Gemini one. adapter_class = GeminiLLMAdapter + # Realtime (speech-to-speech) service. Does NOT emit + # UserStarted/StoppedSpeakingFrame from server-side turn signals — + # the API exposes an `interrupted` event but no turn-start/-end. + _realtime_service_info = RealtimeServiceInfo(emits_user_turn_frames=False) + @property def _is_gemini_3(self) -> bool: """Check if the current model is a Gemini 3.x model.""" diff --git a/src/pipecat/services/inworld/realtime/llm.py b/src/pipecat/services/inworld/realtime/llm.py index 023e09fbc..352e669e4 100644 --- a/src/pipecat/services/inworld/realtime/llm.py +++ b/src/pipecat/services/inworld/realtime/llm.py @@ -51,7 +51,7 @@ from pipecat.metrics.metrics import LLMTokenUsage from pipecat.processors.aggregators import async_tool_messages from pipecat.processors.aggregators.llm_context import LLMContext, LLMSpecificMessage from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.llm_service import FunctionCallFromLLM, LLMService +from pipecat.services.llm_service import FunctionCallFromLLM, LLMService, RealtimeServiceInfo from pipecat.services.settings import ( NOT_GIVEN, LLMSettings, @@ -245,6 +245,10 @@ class InworldRealtimeLLMService(LLMService[InworldRealtimeLLMAdapter]): adapter_class = InworldRealtimeLLMAdapter + # Realtime (speech-to-speech) service. Emits UserStarted/Stopped + # speaking frames from server-side VAD events. + _realtime_service_info = RealtimeServiceInfo(emits_user_turn_frames=True) + # Target ~60ms audio chunks when sending to Inworld (16-bit mono). _AUDIO_CHUNK_TARGET_MS = 60 diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 58ed54bd0..eb01d26fc 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -16,6 +16,7 @@ from collections.abc import Awaitable, Callable, Mapping, Sequence from dataclasses import dataclass from typing import ( Any, + ClassVar, Generic, Protocol, cast, @@ -48,6 +49,7 @@ from pipecat.frames.frames import ( LLMFullResponseStartFrame, LLMTextFrame, LLMUpdateSettingsFrame, + RealtimeServiceMetadataFrame, StartFrame, ) from pipecat.processors.aggregators.llm_context import ( @@ -97,6 +99,31 @@ class FunctionCallResultCallback(Protocol): ... +@dataclass(frozen=True) +class RealtimeServiceInfo: + """Per-service metadata for realtime (speech-to-speech) LLM services. + + Realtime LLM subclasses set ``LLMService._realtime_service_info`` to a + populated instance; the presence of a non-None value is what marks a + service as realtime. Non-realtime services keep the default ``None``. + + Carries the configuration ``LLMService`` and + ``LLMContextAggregatorPair`` need to wire up realtime behavior: + auto-broadcasting ``RealtimeServiceMetadataFrame`` at start, the + startup INFO log for services with no server-side turn signals, and + the aggregator's one-time recommendation log. + + Parameters: + emits_user_turn_frames: Whether the service emits + ``UserStartedSpeakingFrame`` / ``UserStoppedSpeakingFrame`` + from server-side turn signals. False for services with no + server-side turn signals (e.g. Gemini Live, AWS Nova Sonic, + Ultravox). + """ + + emits_user_turn_frames: bool = True + + @dataclass class FunctionCallParams: """Parameters for a function call. @@ -244,6 +271,15 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService, Generic[TAdapter] # However, subclasses should override this with a more specific adapter when necessary. adapter_class: type[BaseLLMAdapter] = OpenAILLMAdapter + # Marker + per-service config for realtime (speech-to-speech) LLM + # services. Realtime subclasses override this with a populated + # ``RealtimeServiceInfo`` instance — the presence of a non-None value + # is what marks the service as realtime. Non-realtime services keep + # the default ``None`` and the realtime-specific machinery + # (auto-broadcast of ``RealtimeServiceMetadataFrame``, startup INFO + # log for services without server-side turn signals) stays inert. + _realtime_service_info: ClassVar[RealtimeServiceInfo | None] = None + # Returned to the LLM as the tool result when an unavailable function is # called. Deliberately neutral about future availability so the LLM can # pick the function up again if it returns (e.g. via the @@ -363,6 +399,21 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService, Generic[TAdapter] await self._create_sequential_runner_task() if self._enable_async_tool_cancellation and self._has_async_tools(): self._setup_async_tool_cancellation() + if ( + self._realtime_service_info is not None + and not self._realtime_service_info.emits_user_turn_frames + ): + logger.info( + f"{self} does not emit UserStartedSpeakingFrame/" + "UserStoppedSpeakingFrame. Pipeline processors that depend on " + "these frames (RTVI client speech events, TurnTrackingObserver, " + "AudioBufferProcessor turn recording, UserIdleController, user " + "mute strategies, voicemail detector) will not activate. To " + "produce them locally, add `vad_analyzer=` to " + "LLMUserAggregatorParams. Note: local turn detection is a " + "heuristic; its boundaries may not match the provider's actual " + "server-side turn decisions and can desynchronize in subtle ways." + ) async def stop(self, frame: EndFrame): """Stop the LLM service. @@ -495,6 +546,23 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService, Generic[TAdapter] await super().push_frame(frame, direction) + # Broadcast realtime-service metadata immediately after the + # StartFrame propagates downstream, mirroring the order STT + # services use for STTMetadataFrame. The aggregator (upstream) + # already received its own StartFrame and is ready to process + # the broadcast; downstream processors see StartFrame then the + # metadata in their queues. + if ( + self._realtime_service_info is not None + and isinstance(frame, StartFrame) + and direction == FrameDirection.DOWNSTREAM + ): + await self.broadcast_frame( + RealtimeServiceMetadataFrame, + service_name=self.name, + emits_user_turn_frames=self._realtime_service_info.emits_user_turn_frames, + ) + async def _push_llm_text(self, text: str): """Push LLM text, using turn completion detection if enabled. diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py index 19fa0f717..9ce78ab89 100644 --- a/src/pipecat/services/openai/realtime/llm.py +++ b/src/pipecat/services/openai/realtime/llm.py @@ -51,7 +51,7 @@ from pipecat.metrics.metrics import LLMTokenUsage from pipecat.processors.aggregators import async_tool_messages from pipecat.processors.aggregators.llm_context import LLMContext, LLMSpecificMessage from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.llm_service import FunctionCallFromLLM, LLMService +from pipecat.services.llm_service import FunctionCallFromLLM, LLMService, RealtimeServiceInfo from pipecat.services.openai._constants import OPENAI_REALTIME_WHISPER_MODEL, OPENAI_SAMPLE_RATE from pipecat.services.settings import ( NOT_GIVEN, @@ -212,6 +212,10 @@ class OpenAIRealtimeLLMService(LLMService[OpenAIRealtimeLLMAdapter]): # Overriding the default adapter to use the OpenAIRealtimeLLMAdapter one. adapter_class = OpenAIRealtimeLLMAdapter + # Realtime (speech-to-speech) service. Emits UserStarted/Stopped + # speaking frames from server-side VAD events. + _realtime_service_info = RealtimeServiceInfo(emits_user_turn_frames=True) + def __init__( self, *, diff --git a/src/pipecat/services/ultravox/llm.py b/src/pipecat/services/ultravox/llm.py index 74735df6b..91aa66486 100644 --- a/src/pipecat/services/ultravox/llm.py +++ b/src/pipecat/services/ultravox/llm.py @@ -48,7 +48,7 @@ from pipecat.frames.frames import ( from pipecat.processors.aggregators import async_tool_messages from pipecat.processors.aggregators.llm_context import LLMContext, LLMSpecificMessage from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.llm_service import FunctionCallFromLLM, LLMService +from pipecat.services.llm_service import FunctionCallFromLLM, LLMService, RealtimeServiceInfo from pipecat.services.settings import NOT_GIVEN, LLMSettings, _NotGiven, assert_given from pipecat.utils.time import time_now_iso8601 @@ -179,6 +179,10 @@ class UltravoxRealtimeLLMService(LLMService): Settings = UltravoxRealtimeLLMSettings _settings: Settings + # Realtime (speech-to-speech) service. Does NOT emit + # UserStarted/StoppedSpeakingFrame from server-side turn signals. + _realtime_service_info = RealtimeServiceInfo(emits_user_turn_frames=False) + def __init__( self, *, diff --git a/src/pipecat/services/xai/realtime/llm.py b/src/pipecat/services/xai/realtime/llm.py index 7ac83d071..cbfdbaf61 100644 --- a/src/pipecat/services/xai/realtime/llm.py +++ b/src/pipecat/services/xai/realtime/llm.py @@ -50,7 +50,7 @@ from pipecat.metrics.metrics import LLMTokenUsage from pipecat.processors.aggregators import async_tool_messages from pipecat.processors.aggregators.llm_context import LLMContext, LLMSpecificMessage from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.llm_service import FunctionCallFromLLM, LLMService +from pipecat.services.llm_service import FunctionCallFromLLM, LLMService, RealtimeServiceInfo from pipecat.services.settings import ( NOT_GIVEN, LLMSettings, @@ -203,6 +203,10 @@ class GrokRealtimeLLMService(LLMService[GrokRealtimeLLMAdapter]): # Use the Grok-specific adapter adapter_class = GrokRealtimeLLMAdapter + # Realtime (speech-to-speech) service. Emits UserStarted/Stopped + # speaking frames from server-side VAD events. + _realtime_service_info = RealtimeServiceInfo(emits_user_turn_frames=True) + def __init__( self, *,