From 638294c1cc273ce9c67a7e0422d0b771b6dfbef3 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Mon, 18 May 2026 11:50:16 -0400 Subject: [PATCH] Add realtime-openai-local-vad example Mirrors the Gemini Live local-VAD example for OpenAI Realtime, showing that `wait_for_transcript_to_end_user_turn=False` composes cleanly with `turn_detection=False`. The OpenAI Realtime service already wires `UserStoppedSpeakingFrame` to `input_audio_buffer.commit` + `response.create` when `turn_detection=False`, so the example is the only new code needed. --- changelog/4480.added.2.md | 1 + .../realtime/realtime-openai-local-vad.py | 182 ++++++++++++++++++ 2 files changed, 183 insertions(+) create mode 100644 changelog/4480.added.2.md create mode 100644 examples/realtime/realtime-openai-local-vad.py diff --git a/changelog/4480.added.2.md b/changelog/4480.added.2.md new file mode 100644 index 000000000..26417a9c3 --- /dev/null +++ b/changelog/4480.added.2.md @@ -0,0 +1 @@ +- The `wait_for_transcript_to_end_user_turn=False` pattern also works with OpenAI Realtime. Set `turn_detection=False` inside `OpenAIRealtimeLLMService.Settings.session_properties.audio.input` to disable OpenAI's server-side VAD; the service then drives turn boundaries from local turn detection, sending `input_audio_buffer.commit` + `response.create` on `UserStoppedSpeakingFrame`. See `examples/realtime/realtime-openai-local-vad.py` for the full pattern. diff --git a/examples/realtime/realtime-openai-local-vad.py b/examples/realtime/realtime-openai-local-vad.py new file mode 100644 index 000000000..135527573 --- /dev/null +++ b/examples/realtime/realtime-openai-local-vad.py @@ -0,0 +1,182 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + + +import os + +from dotenv import load_dotenv +from loguru import logger + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.frames.frames import LLMRunFrame +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 ( + AssistantTurnStoppedMessage, + LLMContextAggregatorPair, + LLMUserAggregatorParams, + UserMessageFinalizedMessage, + UserTurnStoppedMessage, +) +from pipecat.runner.types import RunnerArguments +from pipecat.runner.utils import create_transport +from pipecat.services.openai.realtime.events import ( + AudioConfiguration, + AudioInput, + InputAudioTranscription, + SessionProperties, +) +from pipecat.services.openai.realtime.llm import OpenAIRealtimeLLMService +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.daily.transport import DailyParams +from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams + +load_dotenv(override=True) + + +# We use lambdas to defer transport parameter creation until the transport +# type is selected at runtime. +transport_params = { + "daily": lambda: DailyParams( + audio_in_enabled=True, + audio_out_enabled=True, + ), + "twilio": lambda: FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + ), + "webrtc": lambda: TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + ), +} + + +async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): + logger.info(f"Starting bot") + + # `turn_detection=False` disables OpenAI Realtime's server-side VAD, + # so this pipeline's local turn detection drives turn boundaries. + # The service then sends `input_audio_buffer.commit` + + # `response.create` when it sees `UserStoppedSpeakingFrame`. + llm = OpenAIRealtimeLLMService( + api_key=os.environ["OPENAI_API_KEY"], + settings=OpenAIRealtimeLLMService.Settings( + session_properties=SessionProperties( + audio=AudioConfiguration( + input=AudioInput( + transcription=InputAudioTranscription(), + turn_detection=False, + ), + ), + ), + ), + ) + + context = LLMContext( + [ + { + "role": "developer", + "content": "Say hello. Then ask if I want to hear a joke.", + }, + ], + ) + # `wait_for_transcript_to_end_user_turn=False` is the right setting + # for pipelines like this one — local turn detection driving a + # realtime service. It avoids unnecessary latency from transcript + # delay: the realtime service consumes user audio directly, so + # we don't need user transcripts in context before it can respond. + # With this option: + # + # - Turn strategies do not consider user transcripts, so the user + # turn ends sooner. + # - User transcripts are handled by the aggregator: a simple + # post-turn transcript wait gives them time to arrive after the + # user turn ends, then the aggregator emits + # `on_user_turn_message_finalized` with the new user context + # message. + user_aggregator, assistant_aggregator = LLMContextAggregatorPair( + context, + user_params=LLMUserAggregatorParams( + vad_analyzer=SileroVADAnalyzer(), + wait_for_transcript_to_end_user_turn=False, + ), + ) + + pipeline = Pipeline( + [ + transport.input(), + user_aggregator, + llm, + transport.output(), + assistant_aggregator, + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_metrics=True, + enable_usage_metrics=True, + ), + idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + # Kick off the conversation. + await task.queue_frames([LLMRunFrame()]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + await task.cancel() + + # `on_user_turn_stopped` fires at the end of the user turn. With + # `wait_for_transcript_to_end_user_turn=False`, no user + # transcripts have arrived yet at this point, so + # `message.content` is empty. Logged here to make the end-of-turn + # signal visible alongside the later finalization event. + @user_aggregator.event_handler("on_user_turn_stopped") + async def on_user_turn_stopped(aggregator, strategy, message: UserTurnStoppedMessage): + logger.info(f"User turn ended (strategy: {type(strategy).__name__})") + + # `on_user_turn_message_finalized` fires when the user message has + # been finalized into the context. Here it fires later than + # `on_user_turn_stopped`, after the aggregator's post-turn + # transcript wait completes. + @user_aggregator.event_handler("on_user_turn_message_finalized") + async def on_user_turn_message_finalized( + aggregator, strategy, message: UserMessageFinalizedMessage + ): + timestamp = f"[{message.timestamp}] " if message.timestamp else "" + line = f"{timestamp}user: {message.content}" + logger.info(f"Transcript: {line}") + + @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}") + + runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + + await runner.run(task) + + +async def bot(runner_args: RunnerArguments): + """Main bot entry point compatible with Pipecat Cloud.""" + transport = await create_transport(runner_args, transport_params) + await run_bot(transport, runner_args) + + +if __name__ == "__main__": + from pipecat.runner.run import main + + main()