From cb9fe04e0b948e505c3d89ee1006729ae60154e7 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 21 May 2026 14:14:13 -0400 Subject: [PATCH] Wire Inworld manual-mode turn detection + add local-VAD example MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Inworld Realtime's session properties accept turn_detection=None to put the service into manual mode (matching OpenAI Realtime's turn_detection=False), but the Pipecat integration hardcoded _handle_user_stopped_speaking and _handle_interruption to assume server-side VAD: both were no-ops on the client side because Inworld's server normally handles commit/cancel/response.create automatically. In manual mode the server doesn't, so local-VAD-driven turns stalled — the bot never responded after the user stopped speaking, and interruptions left the in-flight response running. Mirror the OpenAI Realtime pattern: on user-stopped-speaking in manual mode, send InputAudioBufferCommitEvent + ResponseCreateEvent; on interruption in manual mode, send InputAudioBufferClearEvent + ResponseCancelEvent. Gate both on a new _is_manual_turn_detection() helper. Add examples/realtime/realtime-inworld-local-vad.py, the matching *-local-vad.py variant for parity with the OpenAI Realtime and Grok Realtime variants, and point the Inworld service docstring at it. --- changelog/+inworld-manual-mode.fixed.md | 1 + ...ealtime-inworld-local-vad-example.added.md | 1 + .../realtime/realtime-inworld-local-vad.py | 235 ++++++++++++++++++ src/pipecat/services/inworld/realtime/llm.py | 35 ++- 4 files changed, 265 insertions(+), 7 deletions(-) create mode 100644 changelog/+inworld-manual-mode.fixed.md create mode 100644 changelog/+realtime-inworld-local-vad-example.added.md create mode 100644 examples/realtime/realtime-inworld-local-vad.py diff --git a/changelog/+inworld-manual-mode.fixed.md b/changelog/+inworld-manual-mode.fixed.md new file mode 100644 index 000000000..72fce4a92 --- /dev/null +++ b/changelog/+inworld-manual-mode.fixed.md @@ -0,0 +1 @@ +- Fixed `InworldRealtimeLLMService` not supporting manual-mode turn detection (`session_properties.audio.input.turn_detection=None`). Previously `_handle_user_stopped_speaking` and `_handle_interruption` assumed Inworld's server-side VAD handled commit/cancel/response.create automatically and were no-ops on the client side. In manual mode the server doesn't, so local-VAD-driven turns stalled: the bot never responded after the user stopped speaking, and interruptions didn't cancel the in-flight response. Wire the explicit `InputAudioBufferCommitEvent` + `ResponseCreateEvent` on user-stopped-speaking and `InputAudioBufferClearEvent` + `ResponseCancelEvent` on interruption, gated on a new `_is_manual_turn_detection()` check (mirroring the pattern in `OpenAIRealtimeLLMService`). diff --git a/changelog/+realtime-inworld-local-vad-example.added.md b/changelog/+realtime-inworld-local-vad-example.added.md new file mode 100644 index 000000000..f9567463b --- /dev/null +++ b/changelog/+realtime-inworld-local-vad-example.added.md @@ -0,0 +1 @@ +- Added `examples/realtime/realtime-inworld-local-vad.py`, a variant of the base Inworld Realtime example that disables Inworld's server-side turn detection (`turn_detection=None`, manual mode) and instead drives turn boundaries locally with `SileroVADAnalyzer` wired into the user aggregator. Mirrors the OpenAI Realtime and Grok Realtime local-VAD variants. Server-emitted turn frames are preferred when available. diff --git a/examples/realtime/realtime-inworld-local-vad.py b/examples/realtime/realtime-inworld-local-vad.py new file mode 100644 index 000000000..e9c18bbe4 --- /dev/null +++ b/examples/realtime/realtime-inworld-local-vad.py @@ -0,0 +1,235 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Inworld Realtime with locally-driven turn detection. + +By default Inworld Realtime drives the conversation with its own +server-side semantic VAD (see `realtime-inworld.py`). This variant +disables server-side turn detection (``turn_detection=None``, the +"manual" mode in Inworld's session properties) and instead drives turn +boundaries locally with ``SileroVADAnalyzer`` wired into the user +aggregator. Use this variant if you want a turn analyzer like +``LocalSmartTurnV3`` to decide when the user is done speaking, or if you +need ``UserStartedSpeakingFrame`` / ``UserStoppedSpeakingFrame`` to fire +from the same source as ``InterruptionFrame``. + +Caveat: locally-generated turn boundaries are a heuristic and may not +match the provider's actual server-side turn decisions. Prefer +server-emitted turn frames (i.e. the base `realtime-inworld.py` example) +unless you have a specific reason to drive turn detection locally. +""" + +import os +import random +from datetime import datetime + +from dotenv import load_dotenv +from loguru import logger + +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.frames.frames import LLMRunFrame +from pipecat.observers.loggers.transcription_log_observer import ( + TranscriptionLogObserver, +) +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, + RealtimeServiceModeConfig, + UserTurnStoppedMessage, +) +from pipecat.runner.types import RunnerArguments +from pipecat.runner.utils import create_transport +from pipecat.services.inworld.realtime.events import ( + AudioConfiguration, + AudioInput, + AudioOutput, + InputTranscription, + PCMAudioFormat, + SessionProperties, +) +from pipecat.services.inworld.realtime.llm import InworldRealtimeLLMService +from pipecat.services.llm_service import FunctionCallParams +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) + + +async def fetch_weather_from_api(params: FunctionCallParams): + temperature = ( + random.randint(60, 85) + if params.arguments["format"] == "fahrenheit" + else random.randint(15, 30) + ) + await params.result_callback( + { + "conditions": "nice", + "temperature": temperature, + "format": params.arguments["format"], + "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"), + } + ) + + +weather_function = FunctionSchema( + name="get_current_weather", + description="Get the current weather", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + "format": { + "type": "string", + "enum": ["celsius", "fahrenheit"], + "description": "The temperature unit to use.", + }, + }, + required=["location", "format"], +) + +tools = ToolsSchema(standard_tools=[weather_function]) + + +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("Starting Inworld Realtime bot (local VAD)") + + model = "openai/gpt-4.1-mini" + voice = "Sarah" + tts_model = "inworld-tts-2" + stt_model = "assemblyai/u3-rt-pro" + + # Setting session_properties here replaces Inworld's defaults wholesale, + # so we provide a complete SessionProperties — with turn_detection=None + # (manual mode) so local VAD drives turn boundaries instead. + session_properties = SessionProperties( + model=model, + output_modalities=["audio", "text"], + audio=AudioConfiguration( + input=AudioInput( + format=PCMAudioFormat(rate=24000), + transcription=InputTranscription(model=stt_model), + turn_detection=None, + ), + output=AudioOutput( + format=PCMAudioFormat(rate=24000), + model=tts_model, + voice=voice, + ), + ), + ) + + llm = InworldRealtimeLLMService( + api_key=os.environ["INWORLD_API_KEY"], + settings=InworldRealtimeLLMService.Settings( + system_instruction="""You are a helpful and friendly AI assistant powered by Inworld. + +Your voice and personality should be warm and engaging. Keep your responses +concise and conversational since this is a voice interaction. + +Always be helpful and proactive in offering assistance.""", + session_properties=session_properties, + ), + ) + + # Note: function calling requires a paid Inworld account and a + # function-calling-capable model + llm.register_function("get_current_weather", fetch_weather_from_api) + + context = LLMContext( + [{"role": "developer", "content": "Say hello and introduce yourself!"}], + tools, + ) + + user_aggregator, assistant_aggregator = LLMContextAggregatorPair( + context, + # Drive turn detection locally via SileroVAD wired into the user + # aggregator. realtime_service_mode keeps context-write semantics + # correct and (by default) drops the transcript wait on turn-end so + # local VAD can drive turn boundaries on the latency critical path. + realtime_service_mode=RealtimeServiceModeConfig(), + user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()), + ) + + 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, + observers=[TranscriptionLogObserver()], + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info("Client connected") + await task.queue_frames([LLMRunFrame()]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info("Client disconnected") + await task.cancel() + + @user_aggregator.event_handler("on_user_message_added") + async def on_user_message_added(aggregator, message: UserTurnStoppedMessage): + timestamp = f"[{message.timestamp}] " if message.timestamp else "" + logger.info(f"Transcript: {timestamp}user: {message.content}") + + @assistant_aggregator.event_handler("on_assistant_message_added") + async def on_assistant_message_added(aggregator, message: AssistantTurnStoppedMessage): + timestamp = f"[{message.timestamp}] " if message.timestamp else "" + logger.info(f"Transcript: {timestamp}assistant: {message.content}") + + 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() diff --git a/src/pipecat/services/inworld/realtime/llm.py b/src/pipecat/services/inworld/realtime/llm.py index 0b6aa0359..d94914dea 100644 --- a/src/pipecat/services/inworld/realtime/llm.py +++ b/src/pipecat/services/inworld/realtime/llm.py @@ -206,8 +206,10 @@ class InworldRealtimeLLMService(LLMService[InworldRealtimeLLMAdapter]): ``LLMContextAggregatorPair(..., realtime_service_mode=RealtimeServiceModeConfig())`` so context writes are decoupled from those frames. If you wire local VAD (``LLMUserAggregatorParams.vad_analyzer``) on top of this - service, disable Inworld's server-side turn detection first; - otherwise both sources broadcast duplicate user-turn frames. + service, disable Inworld's server-side turn detection first via + ``turn_detection=None`` (manual mode); otherwise both sources + broadcast duplicate user-turn frames. See + ``examples/realtime/realtime-inworld-local-vad.py``. Example:: @@ -429,12 +431,25 @@ class InworldRealtimeLLMService(LLMService[InworldRealtimeLLMAdapter]): return rate return getattr(self, "_output_sample_rate", 24000) + def _is_manual_turn_detection(self) -> bool: + """Whether server-side turn detection is disabled (manual mode).""" + session_properties = assert_given(self._settings.session_properties) + return bool( + session_properties.audio + and session_properties.audio.input + and session_properties.audio.input.turn_detection is None + ) + async def _handle_interruption(self): """Handle user interruption of assistant speech. - Inworld's server-side VAD handles response cancellation and buffer - cleanup automatically, so we only need to clean up local state. + Server-side VAD handles response cancellation and buffer cleanup + automatically; in manual mode the client must send the cancel + and clear events explicitly. """ + if self._is_manual_turn_detection(): + await self.send_client_event(events.InputAudioBufferClearEvent()) + await self.send_client_event(events.ResponseCancelEvent()) await self._truncate_current_audio_response() await self.stop_all_metrics() @@ -449,10 +464,16 @@ class InworldRealtimeLLMService(LLMService[InworldRealtimeLLMAdapter]): async def _handle_user_stopped_speaking(self, frame): """Handle user stopped speaking event. - Inworld's server-side VAD handles commit and response creation, - so this is a no-op. Metrics are started in _handle_evt_speech_stopped. + Server-side VAD handles commit and response creation + automatically; in manual mode the client must send them + explicitly. Metrics are started in _handle_evt_speech_stopped + in the server-VAD path. """ - pass + if self._is_manual_turn_detection(): + await self.start_ttfb_metrics() + await self.start_processing_metrics() + await self.send_client_event(events.InputAudioBufferCommitEvent()) + await self.send_client_event(events.ResponseCreateEvent()) async def _handle_bot_stopped_speaking(self): """Handle bot stopped speaking event."""