Wire Inworld manual-mode turn detection + add local-VAD example

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

@@ -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`).

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.

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()

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."""