From 3247fd11888328778803c7dadc56a4077aa3e433 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 20 May 2026 15:08:40 -0400 Subject: [PATCH] Mark realtime LLM services with RealtimeServiceInfo + emit metadata at start MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Realtime (speech-to-speech) LLM services need to advertise themselves to the rest of the pipeline so downstream components can adapt. Add a new RealtimeServiceMetadataFrame subtype of ServiceMetadataFrame, following the STTMetadataFrame precedent. LLMService gains a single ClassVar, _realtime_service_info, typed RealtimeServiceInfo | None and defaulting to None. The presence of a populated instance is what marks a service as realtime, and the RealtimeServiceInfo dataclass carries the per-service knobs the rest of the pipeline needs — currently just emits_user_turn_frames. Keeping it all under one optional ClassVar avoids stranding realtime-only knobs on the generic LLMService surface; non-realtime services keep the default None and the realtime-specific machinery stays inert. When _realtime_service_info is set, the base service auto-broadcasts RealtimeServiceMetadataFrame right after StartFrame propagates downstream (same ordering as STT). When emits_user_turn_frames is False, a one-time INFO log at start explains which pipeline processors depend on those frames (RTVI client speech events, TurnTrackingObserver, AudioBufferProcessor turn recording, UserIdleController, user mute strategies, voicemail detector) and how to add local VAD if needed. Set the ClassVar on the seven realtime services: OpenAI Realtime, Azure Realtime (via inheritance), Inworld, Grok/xAI Realtime all emit user-turn frames; Gemini Live (and Gemini Live Vertex via inheritance), AWS Nova Sonic, Ultravox do not. In a follow-up commit, LLMContextAggregatorPair will consume RealtimeServiceMetadataFrame to surface a one-time recommendation when realtime_service_mode is not configured. --- src/pipecat/frames/frames.py | 21 ++++++ src/pipecat/services/aws/nova_sonic/llm.py | 6 +- .../services/google/gemini_live/llm.py | 7 +- src/pipecat/services/inworld/realtime/llm.py | 6 +- src/pipecat/services/llm_service.py | 68 +++++++++++++++++++ src/pipecat/services/openai/realtime/llm.py | 6 +- src/pipecat/services/ultravox/llm.py | 6 +- src/pipecat/services/xai/realtime/llm.py | 6 +- 8 files changed, 120 insertions(+), 6 deletions(-) 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, *,