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.
This commit is contained in:
Paul Kompfner
1
changelog/4480.added.2.md
Normal file
1
changelog/4480.added.2.md
Normal file
@@ -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.
|
||||
182
examples/realtime/realtime-openai-local-vad.py
Normal file
182
examples/realtime/realtime-openai-local-vad.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user