Fix OTel examples to use new runner

DailyTransport: expose start_transcription/stop_transcription

CHANGELOG.md

@@ -5,6 +5,14 @@ All notable changes to **Pipecat** will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Added
+
+- Added new functions `DailyTransport.start_transcription()` and
+  `DailyTransport.stop_transcription()` to be able to start and stop Daily
+  transcription dynamically (maybe with different settings).
+
 ## [0.0.68] - 2025-05-28
 
 ### Added

MANIFEST.in

@@ -0,0 +1,4 @@
+prune docs
+prune examples
+prune scripts
+prune tests

examples/foundational/41a-text-only-webrtc.py

@@ -0,0 +1,162 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import argparse
+import os
+
+import aiohttp
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.frames.frames import (
+    LLMMessagesAppendFrame,
+)
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.frameworks.rtvi import (
+    ActionResult,
+    RTVIAction,
+    RTVIActionArgument,
+    RTVIConfig,
+    RTVIObserver,
+    RTVIProcessor,
+    RTVIServerMessageFrame,
+)
+from pipecat.services.openai import OpenAIContextAggregatorPair
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.services.daily import DailyParams
+
+load_dotenv(override=True)
+
+
+# This is an example of a text-only chatbot using small webrtc tranport.
+# It uses the small webrtc transport prebuilt web UI.
+# https://github.com/pipecat-ai/small-webrtc-prebuilt
+
+
+def create_action_llm_append_to_messages(context_aggregator: OpenAIContextAggregatorPair):
+    async def action_llm_append_to_messages_handler(
+        rtvi: RTVIProcessor, service: str, arguments: dict[str, any]
+    ) -> ActionResult:
+        run_immediately = arguments["run_immediately"] if "run_immediately" in arguments else True
+        logger.info(f"run_immediately: {run_immediately}")
+        if run_immediately:
+            await rtvi.interrupt_bot()
+            # We just interrupted the bot so it should be fine to use the
+            # context directly instead of through frame.
+            if "messages" in arguments and arguments["messages"]:
+                frame = LLMMessagesAppendFrame(messages=arguments["messages"])
+                await rtvi.push_frame(frame)
+
+            frame = context_aggregator.user().get_context_frame()
+            await rtvi.push_frame(frame)
+        return True
+
+    action_llm_append_to_messages = RTVIAction(
+        service="llm",
+        action="append_to_messages",
+        result="bool",
+        arguments=[
+            RTVIActionArgument(name="messages", type="array"),
+            RTVIActionArgument(name="run_immediately", type="bool"),
+        ],
+        handler=action_llm_append_to_messages_handler,
+    )
+    return action_llm_append_to_messages
+
+
+# We store functions so objects (e.g. SileroVADAnalyzer) don't get
+# instantiated. The function will be called when the desired transport gets
+# selected.
+transport_params = {
+    "webrtc": lambda: TransportParams(),
+}
+
+
+async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_sigint: bool):
+    logger.info(f"Starting bot")
+
+    # Create an HTTP session for API calls
+    async with aiohttp.ClientSession() as session:
+        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+
+        messages = [
+            {
+                "role": "system",
+                "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Respond to what the user said in a creative and helpful way.",
+            },
+        ]
+
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
+
+        action_llm_append_to_messages = create_action_llm_append_to_messages(context_aggregator)
+        rtvi = RTVIProcessor(config=RTVIConfig(config=[]))
+        rtvi.register_action(action_llm_append_to_messages)
+
+        pipeline = Pipeline(
+            [
+                transport.input(),
+                rtvi,
+                context_aggregator.user(),
+                llm,
+                transport.output(),
+                context_aggregator.assistant(),
+            ]
+        )
+
+        task = PipelineTask(
+            pipeline,
+            params=PipelineParams(
+                allow_interruptions=True,
+                enable_metrics=True,
+            ),
+            observers=[RTVIObserver(rtvi)],
+        )
+
+        @rtvi.event_handler("on_client_ready")
+        async def on_client_ready(rtvi):
+            logger.info("Pipecat client ready.")
+            await rtvi.set_bot_ready()
+
+            # This block is frontend UI specific
+            # These messages are intended for small webrtc UI to only handle text
+            # https://github.com/pipecat-ai/small-webrtc-prebuilt
+            messages = {
+                "show_text_container": True,
+                "show_video_container": False,
+                "show_debug_container": False,
+            }
+            rtvi_frame = RTVIServerMessageFrame(data=messages)
+            await task.queue_frames([rtvi_frame])
+
+        @transport.event_handler("on_client_connected")
+        async def on_client_connected(transport, client):
+            logger.info(f"Client connected: {client}")
+            # Kick off the conversation.
+            await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+        @transport.event_handler("on_client_disconnected")
+        async def on_client_disconnected(transport, client):
+            logger.info(f"Client disconnected")
+
+        @transport.event_handler("on_client_closed")
+        async def on_client_closed(transport, client):
+            logger.info(f"Client closed connection")
+            await task.cancel()
+
+        runner = PipelineRunner(handle_sigint=False)
+
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    from run import main
+
+    main(run_example, transport_params=transport_params)

examples/foundational/41b-text-and-audio-webrtc.py

@@ -0,0 +1,179 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import argparse
+import os
+
+import aiohttp
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import (
+    LLMMessagesAppendFrame,
+)
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.frameworks.rtvi import (
+    ActionResult,
+    RTVIAction,
+    RTVIActionArgument,
+    RTVIConfig,
+    RTVIObserver,
+    RTVIProcessor,
+    RTVIServerMessageFrame,
+)
+from pipecat.services.cartesia.tts import CartesiaTTSService
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.openai import OpenAIContextAggregatorPair
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.services.daily import DailyParams
+
+load_dotenv(override=True)
+
+# This is an example of a chatbot in which a user can speak and/or type text to communicate with the bot.
+# It uses the small webrtc transport prebuilt web UI.
+# https://github.com/pipecat-ai/small-webrtc-prebuilt
+
+
+def create_action_llm_append_to_messages(context_aggregator: OpenAIContextAggregatorPair):
+    async def action_llm_append_to_messages_handler(
+        rtvi: RTVIProcessor, service: str, arguments: dict[str, any]
+    ) -> ActionResult:
+        run_immediately = arguments["run_immediately"] if "run_immediately" in arguments else True
+
+        if run_immediately:
+            await rtvi.interrupt_bot()
+
+            # We just interrupted the bot so it should be fine to use the
+            # context directly instead of through frame.
+            if "messages" in arguments and arguments["messages"]:
+                mess = arguments["messages"]
+                frame = LLMMessagesAppendFrame(messages=arguments["messages"])
+                await rtvi.push_frame(frame)
+
+        if run_immediately:
+            frame = context_aggregator.user().get_context_frame()
+            await rtvi.push_frame(frame)
+
+        return True
+
+    action_llm_append_to_messages = RTVIAction(
+        service="llm",
+        action="append_to_messages",
+        result="bool",
+        arguments=[
+            RTVIActionArgument(name="messages", type="array"),
+            RTVIActionArgument(name="run_immediately", type="bool"),
+        ],
+        handler=action_llm_append_to_messages_handler,
+    )
+    return action_llm_append_to_messages
+
+
+# We store functions so objects (e.g. SileroVADAnalyzer) don't get
+# instantiated. The function will be called when the desired transport gets
+# selected.
+transport_params = {
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+}
+
+
+async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_sigint: bool):
+    logger.info(f"Starting bot")
+
+    # Create an HTTP session for API calls
+    async with aiohttp.ClientSession() as session:
+        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+
+        tts = CartesiaTTSService(
+            api_key=os.getenv("CARTESIA_API_KEY"), voice_id="71a7ad14-091c-4e8e-a314-022ece01c121"
+        )
+
+        messages = [
+            {
+                "role": "system",
+                "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Respond to what the user says in a creative and helpful way. Explain to the User they can speak or type text to communicate with you.",
+            },
+        ]
+
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
+
+        action_llm_append_to_messages = create_action_llm_append_to_messages(context_aggregator)
+        rtvi = RTVIProcessor(config=RTVIConfig(config=[]))
+        rtvi.register_action(action_llm_append_to_messages)
+
+        pipeline = Pipeline(
+            [
+                transport.input(),
+                rtvi,
+                stt,
+                context_aggregator.user(),
+                llm,
+                tts,
+                transport.output(),
+                context_aggregator.assistant(),
+            ]
+        )
+
+        task = PipelineTask(
+            pipeline,
+            params=PipelineParams(
+                allow_interruptions=True,
+                enable_metrics=True,
+            ),
+            observers=[RTVIObserver(rtvi)],
+        )
+
+        @rtvi.event_handler("on_client_ready")
+        async def on_client_ready(rtvi):
+            logger.info("Pipecat client ready.")
+            await rtvi.set_bot_ready()
+
+            # This block is frontend UI specific
+            # These messages are intended for small webrtc UI to only handle text
+            # https://github.com/pipecat-ai/small-webrtc-prebuilt
+            messages = {
+                "show_text_container": True,
+                "show_debug_container": False,
+            }
+            rtvi_frame = RTVIServerMessageFrame(data=messages)
+            await task.queue_frames([rtvi_frame])
+
+        @transport.event_handler("on_client_connected")
+        async def on_client_connected(transport, client):
+            logger.info(f"Client connected: {client}")
+            # Kick off the conversation.
+            await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+        @transport.event_handler("on_client_disconnected")
+        async def on_client_disconnected(transport, client):
+            logger.info(f"Client disconnected")
+
+        @transport.event_handler("on_client_closed")
+        async def on_client_closed(transport, client):
+            logger.info(f"Client closed connection")
+            await task.cancel()
+
+        runner = PipelineRunner(handle_sigint=False)
+
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    from run import main
+
+    main(run_example, transport_params=transport_params)

examples/open-telemetry/daily_runner.py

@@ -0,0 +1,64 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import argparse
+import os
+from typing import Optional
+
+import aiohttp
+
+from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
+
+
+async def configure(aiohttp_session: aiohttp.ClientSession):
+    (url, token, _) = await configure_with_args(aiohttp_session)
+    return (url, token)
+
+
+async def configure_with_args(
+    aiohttp_session: aiohttp.ClientSession, parser: Optional[argparse.ArgumentParser] = None
+):
+    if not parser:
+        parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
+    parser.add_argument(
+        "-u", "--url", type=str, required=False, help="URL of the Daily room to join"
+    )
+    parser.add_argument(
+        "-k",
+        "--apikey",
+        type=str,
+        required=False,
+        help="Daily API Key (needed to create an owner token for the room)",
+    )
+
+    args, unknown = parser.parse_known_args()
+
+    url = args.url or os.getenv("DAILY_SAMPLE_ROOM_URL")
+    key = args.apikey or os.getenv("DAILY_API_KEY")
+
+    if not url:
+        raise Exception(
+            "No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL."
+        )
+
+    if not key:
+        raise Exception(
+            "No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers."
+        )
+
+    daily_rest_helper = DailyRESTHelper(
+        daily_api_key=key,
+        daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
+        aiohttp_session=aiohttp_session,
+    )
+
+    # Create a meeting token for the given room with an expiration 1 hour in
+    # the future.
+    expiry_time: float = 60 * 60
+
+    token = await daily_rest_helper.get_token(url, expiry_time)
+
+    return (url, token, args)

examples/open-telemetry/jaeger/bot.py

@@ -6,6 +6,7 @@
 
 import argparse
 import os
+import sys
 
 from dotenv import load_dotenv
 from loguru import logger
@@ -162,6 +163,7 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si
 
 
 if __name__ == "__main__":
-    from ..run import main
+    sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+    from run import main
 
     main(run_example, transport_params=transport_params)

examples/open-telemetry/langfuse/bot.py

@@ -6,6 +6,7 @@
 
 import argparse
 import os
+import sys
 
 from dotenv import load_dotenv
 from loguru import logger
@@ -159,6 +160,7 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si
 
 
 if __name__ == "__main__":
-    from ..run import main
+    sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+    from run import main
 
     main(run_example, transport_params=transport_params)

examples/open-telemetry/run.py

@@ -10,11 +10,10 @@ import json
 import os
 import sys
 from contextlib import asynccontextmanager
-from typing import Callable, Dict, Mapping, Optional
+from typing import Any, Callable, Dict, Mapping, Optional
 
 import aiohttp
 import uvicorn
-from daily_runner import configure
 from dotenv import load_dotenv
 from fastapi import BackgroundTasks, FastAPI, WebSocket
 from fastapi.middleware.cors import CORSMiddleware
@@ -22,7 +21,7 @@ from fastapi.responses import HTMLResponse, RedirectResponse
 from loguru import logger
 
 from pipecat.serializers.twilio import TwilioFrameSerializer
-from pipecat.transports.base_transport import TransportParams
+from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.network.fastapi_websocket import (
     FastAPIWebsocketParams,
     FastAPIWebsocketTransport,
@@ -35,6 +34,23 @@ from pipecat.transports.services.daily import DailyParams, DailyTransport
 load_dotenv(override=True)
 
 
+def get_transport_client_id(transport: BaseTransport, client: Any) -> str:
+    if isinstance(transport, SmallWebRTCTransport):
+        return client.pc_id
+    elif isinstance(transport, DailyTransport):
+        return client["id"]
+    logger.warning(f"Unable to get client id from unsupported transport {type(transport)}")
+    return ""
+
+
+async def maybe_capture_participant_video(transport: BaseTransport, client: Any):
+    if isinstance(transport, DailyTransport):
+        await transport.capture_participant_video(client["id"], framerate=0, video_source="camera")
+        await transport.capture_participant_video(
+            client["id"], framerate=0, video_source="screenVideo"
+        )
+
+
 def run_example_daily(
     run_example: Callable,
     args: argparse.Namespace,
@@ -42,6 +58,8 @@ def run_example_daily(
 ):
     logger.info("Running example with DailyTransport...")
 
+    from daily_runner import configure
+
     async def run():
         async with aiohttp.ClientSession() as session:
             (room_url, token) = await configure(session)

src/pipecat/audio/vad/vad_analyzer.py

@@ -71,7 +71,7 @@ class VADAnalyzer(ABC):
         self.set_params(self._params)
 
     def set_params(self, params: VADParams):
-        logger.info(f"Setting VAD params to: {params}")
+        logger.debug(f"Setting VAD params to: {params}")
         self._params = params
         self._vad_frames = self.num_frames_required()
         self._vad_frames_num_bytes = self._vad_frames * self._num_channels * 2

src/pipecat/transports/services/daily.py

@@ -476,7 +476,7 @@ class DailyTransportClient(EventHandler):
                 logger.info(f"Joined {self._room_url}")
 
                 if self._params.transcription_enabled:
-                    await self._start_transcription()
+                    await self.start_transcription(self._params.transcription_settings)
 
                 await self._callbacks.on_joined(data)
 
@@ -491,23 +491,6 @@ class DailyTransportClient(EventHandler):
             self._joining = False
             await self._callbacks.on_error(error_msg)
 
-    async def _start_transcription(self):
-        if not self._token:
-            logger.warning("Transcription can't be started without a room token")
-            return
-
-        logger.info(f"Enabling transcription with settings {self._params.transcription_settings}")
-
-        future = self._get_event_loop().create_future()
-        self._client.start_transcription(
-            settings=self._params.transcription_settings.model_dump(exclude_none=True),
-            completion=completion_callback(future),
-        )
-        error = await future
-        if error:
-            logger.error(f"Unable to start transcription: {error}")
-            return
-
     async def _join(self):
         future = self._get_event_loop().create_future()
 
@@ -577,7 +560,7 @@ class DailyTransportClient(EventHandler):
         logger.info(f"Leaving {self._room_url}")
 
         if self._params.transcription_enabled:
-            await self._stop_transcription()
+            await self.stop_transcription()
 
         # Remove any custom tracks, if any.
         for track_name, _ in self._custom_audio_tracks.items():
@@ -597,15 +580,6 @@ class DailyTransportClient(EventHandler):
             logger.error(error_msg)
             await self._callbacks.on_error(error_msg)
 
-    async def _stop_transcription(self):
-        if not self._token:
-            return
-        future = self._get_event_loop().create_future()
-        self._client.stop_transcription(completion=completion_callback(future))
-        error = await future
-        if error:
-            logger.error(f"Unable to stop transcription: {error}")
-
     async def _leave(self):
         future = self._get_event_loop().create_future()
         self._client.leave(completion=completion_callback(future))
@@ -623,14 +597,22 @@ class DailyTransportClient(EventHandler):
         return self._client.participant_counts()
 
     async def start_dialout(self, settings):
+        logger.debug(f"Starting dialout: settings={settings}")
+
         future = self._get_event_loop().create_future()
         self._client.start_dialout(settings, completion=completion_callback(future))
-        await future
+        error = await future
+        if error:
+            logger.error(f"Unable to start dialout: {error}")
 
     async def stop_dialout(self, participant_id):
+        logger.debug(f"Stopping dialout: participant_id={participant_id}")
+
         future = self._get_event_loop().create_future()
         self._client.stop_dialout(participant_id, completion=completion_callback(future))
-        await future
+        error = await future
+        if error:
+            logger.error(f"Unable to stop dialout: {error}")
 
     async def send_dtmf(self, settings):
         future = self._get_event_loop().create_future()
@@ -648,16 +630,54 @@ class DailyTransportClient(EventHandler):
         await future
 
     async def start_recording(self, streaming_settings, stream_id, force_new):
+        logger.debug(
+            f"Starting recording: stream_id={stream_id} force_new={force_new} settings={streaming_settings}"
+        )
+
         future = self._get_event_loop().create_future()
         self._client.start_recording(
             streaming_settings, stream_id, force_new, completion=completion_callback(future)
         )
-        await future
+        error = await future
+        if error:
+            logger.error(f"Unable to start recording: {error}")
 
     async def stop_recording(self, stream_id):
+        logger.debug(f"Stopping recording: stream_id={stream_id}")
+
         future = self._get_event_loop().create_future()
         self._client.stop_recording(stream_id, completion=completion_callback(future))
-        await future
+        error = await future
+        if error:
+            logger.error(f"Unable to stop recording: {error}")
+
+    async def start_transcription(self, settings):
+        if not self._token:
+            logger.warning("Transcription can't be started without a room token")
+            return
+
+        logger.debug(f"Starting transcription: settings={settings}")
+
+        future = self._get_event_loop().create_future()
+        self._client.start_transcription(
+            settings=self._params.transcription_settings.model_dump(exclude_none=True),
+            completion=completion_callback(future),
+        )
+        error = await future
+        if error:
+            logger.error(f"Unable to start transcription: {error}")
+
+    async def stop_transcription(self):
+        if not self._token:
+            return
+
+        logger.debug(f"Stopping transcription")
+
+        future = self._get_event_loop().create_future()
+        self._client.stop_transcription(completion=completion_callback(future))
+        error = await future
+        if error:
+            logger.error(f"Unable to stop transcription: {error}")
 
     async def send_prebuilt_chat_message(self, message: str, user_name: Optional[str] = None):
         if not self._joined:
@@ -1366,6 +1386,12 @@ class DailyTransport(BaseTransport):
     async def stop_recording(self, stream_id=None):
         await self._client.stop_recording(stream_id)
 
+    async def start_transcription(self, settings=None):
+        await self._client.start_transcription(settings)
+
+    async def stop_transcription(self):
+        await self._client.stop_transcription()
+
     async def send_prebuilt_chat_message(self, message: str, user_name: Optional[str] = None):
         """Sends a chat message to Daily's Prebuilt main room.