move observer proxies to FrameProcessor

[Fix] Ultravox frame spacing issue

CHANGELOG.md

@@ -5,6 +5,34 @@ 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 support to `RimeHttpTTSService` for the `arcana` model.
+
+### Changed
+
+- Observers `on_push_frame()` now take a single argument `FramePushed` instead
+  of multiple arguments.
+
+### Removed
+
+- `StartFrame.observer` has been replaced by `StartFrame.observers`. This allows
+  a user to pass all observers instead of a single proxy one.
+
+### Deprecated
+
+- Observer `on_push_frame(src, dst, frame, direction, timestamp)` is now
+  deprecated, use `on_push_frame(data: FramePushed)` instead.
+
+### Fixed
+
+- Fixed a `UltravoxSTTService` issue that would cause the service to generate
+  all tokens as one word.
+
+- Remove custom audio tracks from `DailyTransport` before leaving.
+
 ## [0.0.66] - 2025-05-02
 
 ### Added
@@ -120,7 +148,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   not a very common case.
 
 - Added `RivaSegmentedSTTService`, which allows Riva offline/batch models, such
-  as  to be "canary-1b-asr" used in Pipecat.
+  as to be "canary-1b-asr" used in Pipecat.
 
 ### Deprecated
 

examples/foundational/07q-interruptible-rime-http.py

@@ -44,7 +44,8 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
 
         tts = RimeHttpTTSService(
             api_key=os.getenv("RIME_API_KEY", ""),
-            voice_id="rex",
+            voice_id="luna",
+            model="arcana",
             aiohttp_session=session,
         )
 

examples/foundational/13c-gladia-translation.py

@@ -4,6 +4,7 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
+import argparse
 import os
 
 from dotenv import load_dotenv
@@ -39,7 +40,7 @@ class TranscriptionLogger(FrameProcessor):
             print(f"Translation ({frame.language}): {frame.text}")
 
 
-async def run_bot(webrtc_connection: SmallWebRTCConnection):
+async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):
     logger.info(f"Starting bot")
 
     transport = SmallWebRTCTransport(

examples/foundational/30-observer.py

@@ -14,16 +14,15 @@ from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.frames.frames import (
     BotStartedSpeakingFrame,
     BotStoppedSpeakingFrame,
-    Frame,
     StartInterruptionFrame,
 )
-from pipecat.observers.base_observer import BaseObserver
+from pipecat.observers.base_observer import BaseObserver, FramePushed
 from pipecat.observers.loggers.llm_log_observer import LLMLogObserver
 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.frame_processor import FrameDirection, FrameProcessor
+from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.cartesia.tts import CartesiaTTSService
 from pipecat.services.deepgram.stt import DeepgramSTTService
 from pipecat.services.openai.llm import OpenAILLMService
@@ -46,14 +45,13 @@ class DebugObserver(BaseObserver):
     Log format: [EVENT TYPE]: [source processor] → [destination processor] at [timestamp]s
     """
 
-    async def on_push_frame(
-        self,
-        src: FrameProcessor,
-        dst: FrameProcessor,
-        frame: Frame,
-        direction: FrameDirection,
-        timestamp: int,
-    ):
+    async def on_push_frame(self, data: FramePushed):
+        src = data.source
+        dst = data.destination
+        frame = data.frame
+        direction = data.direction
+        timestamp = data.timestamp
+
         # Convert timestamp to seconds for readability
         time_sec = timestamp / 1_000_000_000
 

examples/foundational/35-pattern-pair-voice-switching.py

@@ -45,6 +45,7 @@ Note:
 """
 
 import argparse
+import asyncio
 import os
 
 from dotenv import load_dotenv
@@ -102,8 +103,17 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
         voice_name = match.content.strip().lower()
         if voice_name in VOICE_IDS:
             voice_id = VOICE_IDS[voice_name]
-            tts.set_voice(voice_id)
-            logger.info(f"Switched to {voice_name} voice")
+
+            # Create task to reset the TTS context after voice change
+            async def change_voice():
+                # First flush any existing audio to finish the current context
+                await tts.flush_audio()
+                # Then set the new voice
+                tts.set_voice(voice_id)
+                logger.info(f"Switched to {voice_name} voice")
+
+            # Schedule the voice change task
+            asyncio.create_task(change_voice())
         else:
             logger.warning(f"Unknown voice: {voice_name}")
 

examples/phone-chatbot-daily-twilio-sip/README.md

@@ -0,0 +1,138 @@
+# Daily + Twilio SIP Voice Bot
+
+This project demonstrates how to create a voice bot that can receive phone calls via Twilio and use Daily's SIP capabilities to enable voice conversations.
+
+## How It Works
+
+1. Twilio receives an incoming call to your phone number
+2. Twilio calls your webhook server (`/call` endpoint)
+3. The server creates a Daily room with SIP capabilities
+4. The server starts the bot process with the room details
+5. The caller is put on hold with music
+6. The bot joins the Daily room and signals readiness
+7. Twilio forwards the call to Daily's SIP endpoint
+8. The caller and bot are connected, and the bot handles the conversation
+
+## Prerequisites
+
+- A Daily account with an API key
+- A Twilio account with a phone number that supports voice
+- OpenAI API key for the bot's intelligence
+- Cartesia API key for text-to-speech
+
+## Setup
+
+1. Create a virtual environment and install dependencies
+
+```bash
+python -m venv venv
+source venv/bin/activate  # On Windows: venv\Scripts\activate
+pip install -r requirements.txt
+```
+
+2. Set up environment variables
+
+Copy the example file and fill in your API keys:
+
+```bash
+cp .env.example .env
+# Edit .env with your API keys
+```
+
+3. Configure your Twilio webhook
+
+In the Twilio console:
+
+- Go to your phone number's configuration
+- Set the webhook for "A Call Comes In" to your server's URL + "/call"
+- For local testing, you can use ngrok to expose your local server
+
+```bash
+ngrok http 8000
+# Then use the provided URL (e.g., https://abc123.ngrok.io/call) in Twilio
+```
+
+## Running the Server
+
+Start the webhook server:
+
+```bash
+python server.py
+```
+
+## Testing
+
+Call your Twilio phone number. The system should answer the call, put you on hold briefly, then connect you with the bot.
+
+## Customizing the Bot
+
+You can customize the bot's behavior by modifying the system prompt in `bot.py`.
+
+### Changing the Hold Music
+
+To change the ringing sound or hold music that callers hear while waiting to be connected to the bot, update the URL in `server.py`:
+
+```python
+resp = VoiceResponse()
+resp.play(
+    url="https://your-custom-audio-file-url.mp3",
+    loop=10,
+)
+```
+
+> Read [Twilio's guide](https://www.twilio.com/en-us/blog/adding-mp3-to-voice-call-using-twilio) on how to set up an mp3 in a voice call.
+
+## Handling Multiple SIP Endpoints
+
+The bot is configured to handle multiple `on_dialin_ready` events that might occur with multiple SIP endpoints. It ensures that each call is only forwarded once using a simple flag:
+
+```python
+# Flag to track if call has been forwarded
+call_already_forwarded = False
+
+@transport.event_handler("on_dialin_ready")
+async def on_dialin_ready(transport, cdata):
+    nonlocal call_already_forwarded
+
+    # Skip if already forwarded
+    if call_already_forwarded:
+        logger.info("Call already forwarded, ignoring this event.")
+        return
+
+    # ... forwarding code ...
+    call_already_forwarded = True
+```
+
+Note that normally calls only require a single SIP endpoint. If you are planning to forward the call to a different number, you will need to set up 2 SIP endpoints: one for the initial call and one for the forwarded call. IMPORTANT: ensure that your `on_dialin_ready` handler only handles the first call.
+
+## Daily SIP Configuration
+
+The bot configures Daily rooms with SIP capabilities using these settings:
+
+```python
+sip_params = DailyRoomSipParams(
+    display_name="phone-user",  # This will show up in the Daily UI; optional display the dialer's number
+    video=False,                # Audio-only call
+    sip_mode="dial-in",         # For receiving calls (vs. dial-out)
+    num_endpoints=1,            # Number of SIP endpoints to create
+)
+```
+
+## Troubleshooting
+
+### Call is not being answered
+
+- Check that your Twilio webhook is correctly configured
+- Verify your Twilio account has sufficient funds
+- Check the logs of both the server and bot processes
+
+### Call connects but no bot is heard
+
+- Ensure your Daily API key is correct and has SIP capabilities
+- Check that the SIP endpoint is being correctly passed to the bot
+- Verify that the Cartesia API key and voice ID are correct
+
+### Bot starts but disconnects immediately
+
+- Check the Daily and Twilio logs for any error messages
+- Ensure your server has stable internet connectivity

examples/phone-chatbot-daily-twilio-sip/bot.py

@@ -0,0 +1,183 @@
+"""Twilio + Daily voice bot implementation."""
+
+import argparse
+import asyncio
+import os
+import sys
+
+from dotenv import load_dotenv
+from loguru import logger
+from twilio.rest import Client
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+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.services.cartesia.tts import CartesiaTTSService
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+
+# Setup logging
+load_dotenv()
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+# Initialize Twilio client
+twilio_client = Client(os.getenv("TWILIO_ACCOUNT_SID"), os.getenv("TWILIO_AUTH_TOKEN"))
+
+
+async def run_bot(room_url: str, token: str, call_id: str, sip_uri: str) -> None:
+    """Run the voice bot with the given parameters.
+
+    Args:
+        room_url: The Daily room URL
+        token: The Daily room token
+        call_id: The Twilio call ID
+        sip_uri: The Daily SIP URI for forwarding the call
+    """
+    logger.info(f"Starting bot with room: {room_url}")
+    logger.info(f"SIP endpoint: {sip_uri}")
+
+    call_already_forwarded = False
+
+    # Setup the Daily transport
+    transport = DailyTransport(
+        room_url,
+        token,
+        "Phone Bot",
+        DailyParams(
+            audio_in_enabled=True,
+            audio_out_enabled=True,
+            transcription_enabled=True,
+            vad_analyzer=SileroVADAnalyzer(),
+        ),
+    )
+
+    # Setup TTS service
+    tts = CartesiaTTSService(
+        api_key=os.getenv("CARTESIA_API_KEY"),
+        voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+    )
+
+    # Setup LLM service
+    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+
+    # Initialize LLM context with system prompt
+    messages = [
+        {
+            "role": "system",
+            "content": (
+                "You are a friendly phone assistant. Your responses will be read aloud, "
+                "so keep them concise and conversational. Avoid special characters or "
+                "formatting. Begin by greeting the caller and asking how you can help them today."
+            ),
+        },
+    ]
+
+    # Setup the conversational context
+    context = OpenAILLMContext(messages)
+    context_aggregator = llm.create_context_aggregator(context)
+
+    # Build the pipeline
+    pipeline = Pipeline(
+        [
+            transport.input(),
+            context_aggregator.user(),
+            llm,
+            tts,
+            transport.output(),
+            context_aggregator.assistant(),
+        ]
+    )
+
+    # Create the pipeline task
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            allow_interruptions=True  # Enable barge-in so callers can interrupt the bot
+        ),
+    )
+
+    # Handle participant joining
+    @transport.event_handler("on_first_participant_joined")
+    async def on_first_participant_joined(transport, participant):
+        logger.info(f"First participant joined: {participant['id']}")
+        await transport.capture_participant_transcription(participant["id"])
+        await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+    # Handle participant leaving
+    @transport.event_handler("on_participant_left")
+    async def on_participant_left(transport, participant, reason):
+        logger.info(f"Participant left: {participant['id']}, reason: {reason}")
+        await task.cancel()
+
+    # Handle call ready to forward
+    @transport.event_handler("on_dialin_ready")
+    async def on_dialin_ready(transport, cdata):
+        nonlocal call_already_forwarded
+
+        # We only want to forward the call once
+        # The on_dialin_ready event will be triggered for each sip endpoint provisioned
+        if call_already_forwarded:
+            logger.warning("Call already forwarded, ignoring this event.")
+            return
+
+        logger.info(f"Forwarding call {call_id} to {sip_uri}")
+
+        try:
+            # Update the Twilio call with TwiML to forward to the Daily SIP endpoint
+            twilio_client.calls(call_id).update(
+                twiml=f"<Response><Dial><Sip>{sip_uri}</Sip></Dial></Response>"
+            )
+            logger.info("Call forwarded successfully")
+            call_already_forwarded = True
+        except Exception as e:
+            logger.error(f"Failed to forward call: {str(e)}")
+            raise
+
+    @transport.event_handler("on_dialin_connected")
+    async def on_dialin_connected(transport, data):
+        logger.debug(f"Dial-in connected: {data}")
+
+    @transport.event_handler("on_dialin_stopped")
+    async def on_dialin_stopped(transport, data):
+        logger.debug(f"Dial-in stopped: {data}")
+
+    @transport.event_handler("on_dialin_error")
+    async def on_dialin_error(transport, data):
+        logger.error(f"Dial-in error: {data}")
+        # If there is an error, the bot should leave the call
+        # This may be also handled in on_participant_left with
+        # await task.cancel()
+
+    @transport.event_handler("on_dialin_warning")
+    async def on_dialin_warning(transport, data):
+        logger.warning(f"Dial-in warning: {data}")
+
+    # Run the pipeline
+    runner = PipelineRunner()
+    await runner.run(task)
+
+
+async def main():
+    """Parse command line arguments and run the bot."""
+    parser = argparse.ArgumentParser(description="Daily + Twilio Voice Bot")
+    parser.add_argument("-u", type=str, required=True, help="Daily room URL")
+    parser.add_argument("-t", type=str, required=True, help="Daily room token")
+    parser.add_argument("-i", type=str, required=True, help="Twilio call ID")
+    parser.add_argument("-s", type=str, required=True, help="Daily SIP URI")
+
+    args = parser.parse_args()
+
+    # Validate required arguments
+    if not all([args.u, args.t, args.i, args.s]):
+        logger.error("All arguments (-u, -t, -i, -s) are required")
+        parser.print_help()
+        sys.exit(1)
+
+    await run_bot(args.u, args.t, args.i, args.s)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/phone-chatbot-daily-twilio-sip/env.example

@@ -0,0 +1,11 @@
+# Daily credentials
+DAILY_API_KEY=your_daily_api_key
+DAILY_API_URL=https://api.daily.co/v1
+
+# Twilio credentials
+TWILIO_ACCOUNT_SID=your_twilio_account_sid
+TWILIO_AUTH_TOKEN=your_twilio_auth_token
+
+# Service keys
+OPENAI_API_KEY=your_openai_api_key
+CARTESIA_API_KEY=your_cartesia_api_key

examples/phone-chatbot-daily-twilio-sip/requirements.txt

@@ -0,0 +1,5 @@
+pipecat-ai[daily,elevenlabs,openai,silero]
+fastapi==0.115.6
+uvicorn
+python-dotenv
+twilio

examples/phone-chatbot-daily-twilio-sip/server.py

@@ -0,0 +1,116 @@
+"""Webhook server to handle Twilio calls and start the voice bot."""
+
+import os
+import shlex
+import subprocess
+from contextlib import asynccontextmanager
+
+import aiohttp
+import uvicorn
+from dotenv import load_dotenv
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.responses import PlainTextResponse
+from twilio.twiml.voice_response import VoiceResponse
+from utils.daily_helpers import create_sip_room
+
+# Load environment variables
+load_dotenv()
+
+
+# Initialize FastAPI app with aiohttp session
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # Create aiohttp session to be used for Daily API calls
+    app.state.session = aiohttp.ClientSession()
+    yield
+    # Close session when shutting down
+    await app.state.session.close()
+
+
+app = FastAPI(lifespan=lifespan)
+
+
+@app.post("/call", response_class=PlainTextResponse)
+async def handle_call(request: Request):
+    """Handle incoming Twilio call webhook."""
+    print("Received call webhook from Twilio")
+
+    try:
+        # Get form data from Twilio webhook
+        form_data = await request.form()
+        data = dict(form_data)
+
+        # Extract call ID (required to forward the call later)
+        call_sid = data.get("CallSid")
+        if not call_sid:
+            raise HTTPException(status_code=400, detail="Missing CallSid in request")
+
+        # Extract the caller's phone number
+        caller_phone = str(data.get("From", "unknown-caller"))
+        print(f"Processing call with ID: {call_sid} from {caller_phone}")
+
+        # Create a Daily room with SIP capabilities
+        try:
+            room_details = await create_sip_room(request.app.state.session, caller_phone)
+        except Exception as e:
+            print(f"Error creating Daily room: {e}")
+            raise HTTPException(status_code=500, detail=f"Failed to create Daily room: {str(e)}")
+
+        # Extract necessary details
+        room_url = room_details["room_url"]
+        token = room_details["token"]
+        sip_endpoint = room_details["sip_endpoint"]
+
+        # Make sure we have a SIP endpoint
+        if not sip_endpoint:
+            raise HTTPException(status_code=500, detail="No SIP endpoint provided by Daily")
+
+        # Start the bot process
+        bot_cmd = f"python bot.py -u {room_url} -t {token} -i {call_sid} -s {sip_endpoint}"
+        try:
+            # Use shlex to properly split the command for subprocess
+            cmd_parts = shlex.split(bot_cmd)
+
+            # CHANGE: Keep stdout/stderr for debugging
+            # Start the bot in the background but capture output
+            subprocess.Popen(
+                cmd_parts,
+                # Don't redirect output so we can see logs
+                # stdout=subprocess.DEVNULL,
+                # stderr=subprocess.DEVNULL
+            )
+            print(f"Started bot process with command: {bot_cmd}")
+        except Exception as e:
+            print(f"Error starting bot: {e}")
+            raise HTTPException(status_code=500, detail=f"Failed to start bot: {str(e)}")
+
+        # Generate TwiML response to put the caller on hold with music
+        # You can replace the URL with your own music file
+        # or use Twilio's built-in music on hold
+        # https://www.twilio.com/docs/voice/twiml/play#music-on-hold
+        resp = VoiceResponse()
+        resp.play(
+            url="https://therapeutic-crayon-2467.twil.io/assets/US_ringback_tone.mp3",
+            loop=10,
+        )
+
+        return str(resp)
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        print(f"Unexpected error: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Server error: {str(e)}")
+
+
+@app.get("/health")
+async def health_check():
+    """Simple health check endpoint."""
+    return {"status": "healthy"}
+
+
+if __name__ == "__main__":
+    # Run the server
+    port = int(os.getenv("PORT", "8000"))
+    print(f"Starting server on port {port}")
+    uvicorn.run("server:app", host="0.0.0.0", port=port, reload=True)

examples/phone-chatbot-daily-twilio-sip/utils/daily_helpers.py

@@ -0,0 +1,76 @@
+"""Helper functions for interacting with the Daily API."""
+
+import os
+from typing import Dict, Optional
+
+import aiohttp
+from dotenv import load_dotenv
+
+from pipecat.transports.services.helpers.daily_rest import (
+    DailyRESTHelper,
+    DailyRoomParams,
+    DailyRoomProperties,
+    DailyRoomSipParams,
+)
+
+load_dotenv()
+
+
+# Initialize Daily API helper
+async def get_daily_helper(session: Optional[aiohttp.ClientSession] = None) -> DailyRESTHelper:
+    """Get a Daily REST helper with the configured API key."""
+    if session is None:
+        session = aiohttp.ClientSession()
+
+    return DailyRESTHelper(
+        daily_api_key=os.getenv("DAILY_API_KEY", ""),
+        daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
+        aiohttp_session=session,
+    )
+
+
+async def create_sip_room(
+    session: Optional[aiohttp.ClientSession] = None, caller_phone: str = "unknown-caller"
+) -> Dict[str, str]:
+    """Create a Daily room with SIP capabilities for phone calls.
+
+    Args:
+        session: Optional aiohttp session to use for API calls
+        caller_phone: The phone number of the caller to use in display name
+
+    Returns:
+        Dictionary with room URL, token, and SIP endpoint
+    """
+    daily_helper = await get_daily_helper(session)
+
+    # Configure SIP parameters
+    sip_params = DailyRoomSipParams(
+        display_name=caller_phone,
+        video=False,
+        sip_mode="dial-in",
+        num_endpoints=1,
+    )
+
+    # Create room properties with SIP enabled
+    properties = DailyRoomProperties(
+        sip=sip_params,
+        enable_dialout=True,  # Needed for outbound calls if you expand the bot
+        enable_chat=False,  # No need for chat in a voice bot
+        start_video_off=True,  # Voice only
+    )
+
+    # Create room parameters
+    params = DailyRoomParams(properties=properties)
+
+    # Create the room
+    try:
+        room = await daily_helper.create_room(params=params)
+        print(f"Created room: {room.url} with SIP endpoint: {room.config.sip_endpoint}")
+
+        # Get token for the bot to join
+        token = await daily_helper.get_token(room.url, 24 * 60 * 60)  # 24 hours validity
+
+        return {"room_url": room.url, "token": token, "sip_endpoint": room.config.sip_endpoint}
+    except Exception as e:
+        print(f"Error creating room: {e}")
+        raise

examples/phone-chatbot/README.md

@@ -235,10 +235,10 @@ For incoming calls from customers, Daily will send a webhook to your `/start` en
 
 ```json
 {
-	"From": "+CALLERS_PHONE",
-	"To": "$PURCHASED_PHONE",
-	"callId": "callid-read-only-string",
-	"callDomain": "callDomain-read-only-string"
+  "From": "+CALLERS_PHONE",
+  "To": "$PURCHASED_PHONE",
+  "callId": "callid-read-only-string",
+  "callDomain": "callDomain-read-only-string"
 }
 ```
 
@@ -266,63 +266,63 @@ When making requests to the `/start` endpoint, the config object can include:
 
 ```json
 {
-	"config": {
-		"prompts": [
-			{
-				"name": "call_transfer_initial_prompt",
-				"text": "Your custom prompt here"
-			},
-			{
-				"name": "call_transfer_prompt",
-				"text": "Your custom prompt here"
-			},
-			{
-				"name": "call_transfer_finished_prompt",
-				"text": "Your custom prompt here"
-			},
-			{
-				"name": "voicemail_detection_prompt",
-				"text": "Your custom prompt here"
-			},
-			{
-				"name": "voicemail_prompt",
-				"text": "Your custom prompt here"
-			},
-			{
-				"name": "human_conversation_prompt",
-				"text": "Your custom prompt here"
-			}
-		],
-		"dialin_settings": {
-			"From": "+CALLERS_PHONE",
-			"To": "$PURCHASED_PHONE",
-			"callId": "callid-read-only-string",
-			"callDomain": "callDomain-read-only-string"
-		},
-		"dialout_settings": [
-			{
-				"phoneNumber": "+12345678910",
-				"callerId": "caller-id-uuid",
-				"sipUri": "sip:maria@example.com"
-			}
-		],
-		"call_transfer": {
-			"mode": "dialout",
-			"speakSummary": true,
-			"storeSummary": false,
-			"operatorNumber": "+12345678910",
-			"testInPrebuilt": false
-		},
-		"voicemail_detection": {
-			"testInPrebuilt": true
-		},
-		"simple_dialin": {
-			"testInPrebuilt": true
-		},
-		"simple_dialout": {
-			"testInPrebuilt": true
-		}
-	}
+  "config": {
+    "prompts": [
+      {
+        "name": "call_transfer_initial_prompt",
+        "text": "Your custom prompt here"
+      },
+      {
+        "name": "call_transfer_prompt",
+        "text": "Your custom prompt here"
+      },
+      {
+        "name": "call_transfer_finished_prompt",
+        "text": "Your custom prompt here"
+      },
+      {
+        "name": "voicemail_detection_prompt",
+        "text": "Your custom prompt here"
+      },
+      {
+        "name": "voicemail_prompt",
+        "text": "Your custom prompt here"
+      },
+      {
+        "name": "human_conversation_prompt",
+        "text": "Your custom prompt here"
+      }
+    ],
+    "dialin_settings": {
+      "From": "+CALLERS_PHONE",
+      "To": "$PURCHASED_PHONE",
+      "callId": "callid-read-only-string",
+      "callDomain": "callDomain-read-only-string"
+    },
+    "dialout_settings": [
+      {
+        "phoneNumber": "+12345678910",
+        "callerId": "caller-id-uuid",
+        "sipUri": "sip:maria@example.com"
+      }
+    ],
+    "call_transfer": {
+      "mode": "dialout",
+      "speakSummary": true,
+      "storeSummary": false,
+      "operatorNumber": "+12345678910",
+      "testInPrebuilt": false
+    },
+    "voicemail_detection": {
+      "testInPrebuilt": true
+    },
+    "simple_dialin": {
+      "testInPrebuilt": true
+    },
+    "simple_dialout": {
+      "testInPrebuilt": true
+    }
+  }
 }
 ```
 
@@ -393,19 +393,19 @@ The following table shows which feature combinations are supported when making r
 
 ```json
 {
-	"config": {
-		"dialin_settings": {
-			"from": "+12345678901",
-			"to": "+19876543210",
-			"call_id": "call-id-string",
-			"call_domain": "domain-string"
-		},
-		"call_transfer": {
-			"mode": "dialout",
-			"speakSummary": true,
-			"operatorNumber": "+12345678910"
-		}
-	}
+  "config": {
+    "dialin_settings": {
+      "from": "+12345678901",
+      "to": "+19876543210",
+      "call_id": "call-id-string",
+      "call_domain": "domain-string"
+    },
+    "call_transfer": {
+      "mode": "dialout",
+      "speakSummary": true,
+      "operatorNumber": "+12345678910"
+    }
+  }
 }
 ```
 
@@ -413,14 +413,14 @@ The following table shows which feature combinations are supported when making r
 
 ```json
 {
-	"config": {
-		"call_transfer": {
-			"mode": "dialout",
-			"speakSummary": true,
-			"operatorNumber": "+12345678910",
-			"testInPrebuilt": true
-		}
-	}
+  "config": {
+    "call_transfer": {
+      "mode": "dialout",
+      "speakSummary": true,
+      "operatorNumber": "+12345678910",
+      "testInPrebuilt": true
+    }
+  }
 }
 ```
 
@@ -428,11 +428,11 @@ The following table shows which feature combinations are supported when making r
 
 ```json
 {
-	"config": {
-		"voicemail_detection": {
-			"testInPrebuilt": true
-		}
-	}
+  "config": {
+    "voicemail_detection": {
+      "testInPrebuilt": true
+    }
+  }
 }
 ```
 
@@ -440,16 +440,16 @@ The following table shows which feature combinations are supported when making r
 
 ```json
 {
-	"config": {
-		"dialout_settings": [
-			{
-				"phoneNumber": "+12345678910"
-			}
-		],
-		"voicemail_detection": {
-			"testInPrebuilt": false
-		}
-	}
+  "config": {
+    "dialout_settings": [
+      {
+        "phoneNumber": "+12345678910"
+      }
+    ],
+    "voicemail_detection": {
+      "testInPrebuilt": false
+    }
+  }
 }
 ```
 
@@ -457,15 +457,15 @@ The following table shows which feature combinations are supported when making r
 
 ```json
 {
-	"config": {
-		"dialin_settings": {
-			"from": "+12345678901",
-			"to": "+19876543210",
-			"call_id": "call-id-string",
-			"call_domain": "domain-string"
-		},
-		"simple_dialin": {}
-	}
+  "config": {
+    "dialin_settings": {
+      "from": "+12345678901",
+      "to": "+19876543210",
+      "call_id": "call-id-string",
+      "call_domain": "domain-string"
+    },
+    "simple_dialin": {}
+  }
 }
 ```
 
@@ -473,11 +473,11 @@ The following table shows which feature combinations are supported when making r
 
 ```json
 {
-	"config": {
-		"simple_dialin": {
-			"testInPrebuilt": true
-		}
-	}
+  "config": {
+    "simple_dialin": {
+      "testInPrebuilt": true
+    }
+  }
 }
 ```
 
@@ -485,14 +485,14 @@ The following table shows which feature combinations are supported when making r
 
 ```json
 {
-	"config": {
-		"dialout_settings": [
-			{
-				"phoneNumber": "+12345678910"
-			}
-		],
-		"simple_dialout": {}
-	}
+  "config": {
+    "dialout_settings": [
+      {
+        "phoneNumber": "+12345678910"
+      }
+    ],
+    "simple_dialout": {}
+  }
 }
 ```
 
@@ -500,37 +500,14 @@ The following table shows which feature combinations are supported when making r
 
 ```json
 {
-	"config": {
-		"simple_dialout": {
-			"testInPrebuilt": true
-		}
-	}
+  "config": {
+    "simple_dialout": {
+      "testInPrebuilt": true
+    }
+  }
 }
 ```
 
-## Using Twilio (Alternative)
-
-To use Twilio for call handling:
-
-1. Start the bot runner:
-
-   ```shell
-   python bot_runner.py --host localhost
-   ```
-
-2. Start ngrok:
-
-   ```shell
-   ngrok http --domain yourdomain.ngrok.app 7860
-   ```
-
-3. In another terminal, run the Twilio bot:
-   ```shell
-   python bot_twilio.py
-   ```
-
-Make requests to `/start_twilio_bot` for Twilio-specific functionality.
-
 ## Deployment
 
 See Pipecat Cloud deployment docs for how to deploy this example: https://docs.pipecat.daily.co/agents/deploy

examples/phone-chatbot/bot_runner.py

@@ -20,8 +20,7 @@ from bot_runner_helpers import (
 from dotenv import load_dotenv
 from fastapi import FastAPI, HTTPException, Request
 from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import JSONResponse, PlainTextResponse
-from twilio.twiml.voice_response import VoiceResponse
+from fastapi.responses import JSONResponse
 
 from pipecat.transports.services.helpers.daily_rest import (
     DailyRESTHelper,
@@ -125,32 +124,6 @@ async def start_bot(room_details: Dict[str, str], body: Dict[str, Any], example:
         raise HTTPException(status_code=500, detail=f"Failed to start subprocess: {e}")
 
 
-async def start_twilio_bot(room_details: Dict[str, str], call_id: str) -> bool:
-    """Start a Twilio bot process with the given configuration.
-
-    Args:
-        room_details: Room URL, token, and SIP endpoint
-        call_id: Twilio call ID (CallSid)
-
-    Returns:
-        Boolean indicating success
-    """
-    room_url = room_details["room"]
-    token = room_details["token"]
-    sip_endpoint = room_details["sip_endpoint"]
-
-    # Format command for Twilio bot
-    bot_proc = f"python3 -m bot_twilio -u {room_url} -t {token} -i {call_id} -s {sip_endpoint}"
-    print(f"Starting Twilio bot. Room: {room_url}")
-
-    try:
-        command_parts = shlex.split(bot_proc)
-        subprocess.Popen(command_parts, bufsize=1, cwd=os.path.dirname(os.path.abspath(__file__)))
-        return True
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=f"Failed to start subprocess: {e}")
-
-
 # ----------------- API Setup ----------------- #
 
 
@@ -180,47 +153,6 @@ app.add_middleware(
 # ----------------- API Endpoints ----------------- #
 
 
-@app.post("/twilio_start_bot", response_class=PlainTextResponse)
-async def twilio_start_bot(request: Request):
-    """Handle incoming Twilio webhook calls and start a Twilio bot.
-
-    This endpoint is called directly by Twilio as a webhook when a call is received.
-    It puts the call on hold with music and starts a bot that will handle the call.
-    """
-    print("POST /twilio_start_bot")
-
-    # Get form data from Twilio webhook
-    try:
-        form_data = await request.form()
-        data = dict(form_data)
-    except Exception as e:
-        raise HTTPException(status_code=400, detail=f"Failed to parse Twilio form data: {str(e)}")
-
-    # Get default room URL from environment
-    room_url = os.getenv("DAILY_SAMPLE_ROOM_URL", None)
-
-    # Extract call ID from Twilio data
-    call_id = data.get("CallSid")
-    if not call_id:
-        raise HTTPException(status_code=400, detail="Missing 'CallSid' in request")
-
-    print(f"CallId: {call_id}")
-
-    # Create Daily room for the Twilio call
-    room_details = await create_daily_room(room_url, None)  # No special config for Twilio rooms
-
-    # Start the Twilio bot
-    await start_twilio_bot(room_details, call_id)
-
-    # Put the call on hold until the bot is ready to handle it
-    # The bot will update the call with the SIP URI when it's ready
-    resp = VoiceResponse()
-    resp.play(
-        url="http://com.twilio.sounds.music.s3.amazonaws.com/MARKOVICHAMP-Borghestral.mp3", loop=10
-    )
-    return str(resp)
-
-
 @app.post("/start")
 async def handle_start_request(request: Request) -> JSONResponse:
     """Unified endpoint to handle bot configuration for different scenarios."""
@@ -228,21 +160,7 @@ async def handle_start_request(request: Request) -> JSONResponse:
     room_url = os.getenv("DAILY_SAMPLE_ROOM_URL", None)
 
     try:
-        # Check if this is form data (from Twilio) or JSON
-        content_type = request.headers.get("content-type", "").lower()
-
-        if "application/x-www-form-urlencoded" in content_type:
-            # Handle form data from Twilio
-            form_data = await request.form()
-            data = dict(form_data)
-
-            # Check for CallSid which indicates this is a Twilio webhook
-            if "CallSid" in data:
-                # Redirect to Twilio handler for backward compatibility
-                return await twilio_start_bot(request)
-        else:
-            # Parse JSON request data
-            data = await request.json()
+        data = await request.json()
 
         # Handle webhook test
         if "test" in data:
@@ -298,14 +216,6 @@ async def handle_start_request(request: Request) -> JSONResponse:
         return JSONResponse(response)
 
     except json.JSONDecodeError:
-        # Check if this might be form data from Twilio
-        try:
-            content_type = request.headers.get("content-type", "").lower()
-            if "application/x-www-form-urlencoded" in content_type:
-                return await twilio_start_bot(request)
-        except Exception:
-            pass
-
         raise HTTPException(status_code=400, detail="Invalid JSON in request body")
     except Exception as e:
         raise HTTPException(status_code=400, detail=f"Request processing error: {str(e)}")

examples/phone-chatbot/bot_twilio.py

@@ -1,122 +0,0 @@
-#
-# Copyright (c) 2024–2025, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import argparse
-import asyncio
-import os
-import sys
-
-from dotenv import load_dotenv
-from loguru import logger
-from twilio.rest import Client
-
-from pipecat.audio.vad.silero import SileroVADAnalyzer
-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.services.elevenlabs.tts import ElevenLabsTTSService
-from pipecat.services.openai.llm import OpenAILLMService
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-
-load_dotenv(override=True)
-
-logger.remove(0)
-logger.add(sys.stderr, level="DEBUG")
-
-
-twilio_account_sid = os.getenv("TWILIO_ACCOUNT_SID")
-twilio_auth_token = os.getenv("TWILIO_AUTH_TOKEN")
-twilioclient = Client(twilio_account_sid, twilio_auth_token)
-
-daily_api_key = os.getenv("DAILY_API_KEY", "")
-
-
-async def main(room_url: str, token: str, callId: str, sipUri: str):
-    # dialin_settings are only needed if Daily's SIP URI is used
-    # If you are handling this via Twilio, Telnyx, set this to None
-    # and handle call-forwarding when on_dialin_ready fires.
-    transport = DailyTransport(
-        room_url,
-        token,
-        "Chatbot",
-        DailyParams(
-            api_key=daily_api_key,
-            dialin_settings=None,  # Not required for Twilio
-            audio_in_enabled=True,
-            audio_out_enabled=True,
-            video_out_enabled=False,
-            vad_analyzer=SileroVADAnalyzer(),
-            transcription_enabled=True,
-        ),
-    )
-
-    tts = ElevenLabsTTSService(
-        api_key=os.getenv("ELEVENLABS_API_KEY", ""),
-        voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""),
-    )
-
-    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
-
-    messages = [
-        {
-            "role": "system",
-            "content": "You are Chatbot, a friendly, helpful robot. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way, but keep your responses brief. Start by saying 'Hello! Who dares dial me at this hour?!'.",
-        },
-    ]
-
-    context = OpenAILLMContext(messages)
-    context_aggregator = llm.create_context_aggregator(context)
-
-    pipeline = Pipeline(
-        [
-            transport.input(),
-            context_aggregator.user(),
-            llm,
-            tts,
-            transport.output(),
-            context_aggregator.assistant(),
-        ]
-    )
-
-    task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True))
-
-    @transport.event_handler("on_first_participant_joined")
-    async def on_first_participant_joined(transport, participant):
-        await transport.capture_participant_transcription(participant["id"])
-        await task.queue_frames([context_aggregator.user().get_context_frame()])
-
-    @transport.event_handler("on_participant_left")
-    async def on_participant_left(transport, participant, reason):
-        await task.cancel()
-
-    @transport.event_handler("on_dialin_ready")
-    async def on_dialin_ready(transport, cdata):
-        # For Twilio, Telnyx, etc. You need to update the state of the call
-        # and forward it to the sip_uri..
-        print(f"Forwarding call: {callId} {sipUri}")
-
-        try:
-            # The TwiML is updated using Twilio's client library
-            call = twilioclient.calls(callId).update(
-                twiml=f"<Response><Dial><Sip>{sipUri}</Sip></Dial></Response>"
-            )
-        except Exception as e:
-            raise Exception(f"Failed to forward call: {str(e)}")
-
-    runner = PipelineRunner()
-    await runner.run(task)
-
-
-if __name__ == "__main__":
-    parser = argparse.ArgumentParser(description="Pipecat Simple ChatBot")
-    parser.add_argument("-u", type=str, help="Room URL")
-    parser.add_argument("-t", type=str, help="Token")
-    parser.add_argument("-i", type=str, help="Call ID")
-    parser.add_argument("-s", type=str, help="SIP URI")
-    config = parser.parse_args()
-
-    asyncio.run(main(config.u, config.t, config.i, config.s))

examples/phone-chatbot/env.example

@@ -5,8 +5,6 @@ DEEPGRAM_API_KEY=
 OPENAI_API_KEY=
 GOOGLE_API_KEY
 CARTESIA_API_KEY=
-TWILIO_ACCOUNT_SID=
-TWILIO_AUTH_TOKEN=
 DIAL_IN_FROM_NUMBER=
 DIAL_OUT_TO_NUMBER=
 OPERATOR_NUMBER=

examples/phone-chatbot/requirements.txt

@@ -2,5 +2,4 @@ pipecat-ai[daily,cartesia,deepgram,openai,google,silero]
 fastapi==0.115.6
 uvicorn
 python-dotenv
-twilio
 python-multipart

src/pipecat/frames/frames.py

@@ -16,6 +16,7 @@ from typing import (
     Literal,
     Mapping,
     Optional,
+    Sequence,
     Tuple,
 )
 
@@ -449,12 +450,12 @@ class StartFrame(SystemFrame):
 
     clock: BaseClock
     task_manager: BaseTaskManager
+    observers: Sequence["BaseObserver"]
     audio_in_sample_rate: int = 16000
     audio_out_sample_rate: int = 24000
     allow_interruptions: bool = False
     enable_metrics: bool = False
     enable_usage_metrics: bool = False
-    observer: Optional["BaseObserver"] = None
     report_only_initial_ttfb: bool = False
 
 

src/pipecat/observers/base_observer.py

@@ -5,9 +5,38 @@
 #
 
 from abc import ABC, abstractmethod
+from dataclasses import dataclass
+
+from typing_extensions import TYPE_CHECKING
 
 from pipecat.frames.frames import Frame
-from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+
+if TYPE_CHECKING:
+    from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+
+
+@dataclass
+class FramePushed:
+    """Represents an event where a frame is pushed from one processor to another
+    within the pipeline.
+
+    This data structure is typically used by observers to track the flow of
+    frames through the pipeline for logging, debugging, or analytics purposes.
+
+    Attributes:
+        source (FrameProcessor): The processor sending the frame.
+        destination (FrameProcessor): The processor receiving the frame.
+        frame (Frame): The frame being transferred.
+        direction (FrameDirection): The direction of the transfer (e.g., downstream or upstream).
+        timestamp (int): The time when the frame was pushed, based on the pipeline clock.
+
+    """
+
+    source: "FrameProcessor"
+    destination: "FrameProcessor"
+    frame: Frame
+    direction: "FrameDirection"
+    timestamp: int
 
 
 class BaseObserver(ABC):
@@ -19,26 +48,15 @@ class BaseObserver(ABC):
     """
 
     @abstractmethod
-    async def on_push_frame(
-        self,
-        src: FrameProcessor,
-        dst: FrameProcessor,
-        frame: Frame,
-        direction: FrameDirection,
-        timestamp: int,
-    ):
-        """Abstract method to handle the event when a frame is pushed from one
-        processor to another.
+    async def on_push_frame(self, data: FramePushed):
+        """Handle the event when a frame is pushed from one processor to another.
+
+        This method should be implemented by subclasses to define specific
+        behavior (e.g., logging, monitoring, debugging) when a frame is
+        transferred through the pipeline.
 
         Args:
-            src (FrameProcessor): The source frame processor that is sending the frame.
-            dst (FrameProcessor): The destination frame processor that will receive the frame.
-            frame (Frame): The frame being transferred between processors.
-            direction (FrameDirection): The direction of the frame transfer.
-            timestamp (int): The timestamp when the frame was pushed (based on the pipeline clock).
-
-        This method should be implemented by subclasses to define specific behavior
-        when a frame is pushed.
+            data (FramePushed): The event data containing details about the frame transfer.
 
         """
         pass

src/pipecat/observers/loggers/llm_log_observer.py

@@ -15,7 +15,7 @@ from pipecat.frames.frames import (
     LLMMessagesFrame,
     LLMTextFrame,
 )
-from pipecat.observers.base_observer import BaseObserver
+from pipecat.observers.base_observer import BaseObserver, FramePushed
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.services.llm_service import LLMService
@@ -38,14 +38,13 @@ class LLMLogObserver(BaseObserver):
 
     """
 
-    async def on_push_frame(
-        self,
-        src: FrameProcessor,
-        dst: FrameProcessor,
-        frame: Frame,
-        direction: FrameDirection,
-        timestamp: int,
-    ):
+    async def on_push_frame(self, data: FramePushed):
+        src = data.source
+        dst = data.destination
+        frame = data.frame
+        direction = data.direction
+        timestamp = data.timestamp
+
         if not isinstance(src, LLMService) and not isinstance(dst, LLMService):
             return
 

src/pipecat/observers/loggers/transcription_log_observer.py

@@ -11,7 +11,7 @@ from pipecat.frames.frames import (
     InterimTranscriptionFrame,
     TranscriptionFrame,
 )
-from pipecat.observers.base_observer import BaseObserver
+from pipecat.observers.base_observer import BaseObserver, FramePushed
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.services.stt_service import STTService
 
@@ -29,14 +29,11 @@ class TranscriptionLogObserver(BaseObserver):
 
     """
 
-    async def on_push_frame(
-        self,
-        src: FrameProcessor,
-        dst: FrameProcessor,
-        frame: Frame,
-        direction: FrameDirection,
-        timestamp: int,
-    ):
+    async def on_push_frame(self, data: FramePushed):
+        src = data.source
+        frame = data.frame
+        timestamp = data.timestamp
+
         if not isinstance(src, STTService):
             return
 

src/pipecat/pipeline/task.py

@@ -32,7 +32,6 @@ from pipecat.metrics.metrics import ProcessingMetricsData, TTFBMetricsData
 from pipecat.observers.base_observer import BaseObserver
 from pipecat.pipeline.base_pipeline import BasePipeline
 from pipecat.pipeline.base_task import BaseTask
-from pipecat.pipeline.task_observer import TaskObserver
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.utils.asyncio import BaseTaskManager, TaskManager
 
@@ -184,6 +183,7 @@ class PipelineTask(BaseTask):
         self._idle_timeout_secs = idle_timeout_secs
         self._idle_timeout_frames = idle_timeout_frames
         self._cancel_on_idle_timeout = cancel_on_idle_timeout
+        self._observers = observers
         if self._params.observers:
             import warnings
 
@@ -193,7 +193,7 @@ class PipelineTask(BaseTask):
                     "Field 'observers' is deprecated, use the 'observers' parameter instead.",
                     DeprecationWarning,
                 )
-            observers = self._params.observers
+            self._observers = self._params.observers
         self._finished = False
 
         # This queue receives frames coming from the pipeline upstream.
@@ -229,11 +229,6 @@ class PipelineTask(BaseTask):
         # PipelineTask and its frame processors.
         self._task_manager = task_manager or TaskManager()
 
-        # The task observer acts as a proxy to the provided observers. This way,
-        # we only need to pass a single observer (using the StartFrame) which
-        # then just acts as a proxy.
-        self._observer = TaskObserver(observers=observers, task_manager=self._task_manager)
-
         # These events can be used to check which frames make it to the source
         # or sink processors. Instead of calling the event handlers for every
         # frame the user needs to specify which events they are interested
@@ -347,8 +342,6 @@ class PipelineTask(BaseTask):
             self._process_push_queue(), f"{self}::_process_push_queue"
         )
 
-        await self._observer.start()
-
         return self._process_push_task
 
     def _maybe_start_heartbeat_tasks(self):
@@ -367,8 +360,6 @@ class PipelineTask(BaseTask):
             )
 
     async def _cancel_tasks(self):
-        await self._observer.stop()
-
         await self._task_manager.cancel_task(self._process_up_task)
         await self._task_manager.cancel_task(self._process_down_task)
 
@@ -425,7 +416,7 @@ class PipelineTask(BaseTask):
             audio_out_sample_rate=self._params.audio_out_sample_rate,
             enable_metrics=self._params.enable_metrics,
             enable_usage_metrics=self._params.enable_usage_metrics,
-            observer=self._observer,
+            observers=self._observers,
             report_only_initial_ttfb=self._params.report_only_initial_ttfb,
         )
         start_frame.metadata = self._params.start_metadata

src/pipecat/pipeline/task_observer.py

@@ -1,103 +0,0 @@
-#
-# Copyright (c) 2024–2025, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-from typing import List
-
-from attr import dataclass
-
-from pipecat.frames.frames import Frame
-from pipecat.observers.base_observer import BaseObserver
-from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.utils.asyncio import BaseTaskManager
-
-
-@dataclass
-class Proxy:
-    """This is the data we receive from the main observer and that we put into
-    a queue for later processing.
-
-    """
-
-    queue: asyncio.Queue
-    task: asyncio.Task
-    observer: BaseObserver
-
-
-@dataclass
-class ObserverData:
-    """This is the data we receive from the main observer and that we put into a
-    proxy queue for later processing.
-
-    """
-
-    src: FrameProcessor
-    dst: FrameProcessor
-    frame: Frame
-    direction: FrameDirection
-    timestamp: int
-
-
-class TaskObserver(BaseObserver):
-    """This is a pipeline frame observer that is meant to be used as a proxy to
-    the user provided observers. That is, this is the observer that should be
-    passed to the frame processors. Then, every time a frame is pushed this
-    observer will call all the observers registered to the pipeline task.
-
-    This observer makes sure that passing frames to observers doesn't block the
-    pipeline by creating a queue and a task for each user observer. When a frame
-    is received, it will be put in a queue for efficiency and later processed by
-    each task.
-
-    """
-
-    def __init__(self, *, observers: List[BaseObserver] = [], task_manager: BaseTaskManager):
-        self._observers = observers
-        self._task_manager = task_manager
-        self._proxies: List[Proxy] = []
-
-    async def start(self):
-        """Starts all proxy observer tasks."""
-        self._proxies = self._create_proxies(self._observers)
-
-    async def stop(self):
-        """Stops all proxy observer tasks."""
-        for proxy in self._proxies:
-            await self._task_manager.cancel_task(proxy.task)
-
-    async def on_push_frame(
-        self,
-        src: FrameProcessor,
-        dst: FrameProcessor,
-        frame: Frame,
-        direction: FrameDirection,
-        timestamp: int,
-    ):
-        for proxy in self._proxies:
-            await proxy.queue.put(
-                ObserverData(
-                    src=src, dst=dst, frame=frame, direction=direction, timestamp=timestamp
-                )
-            )
-
-    def _create_proxies(self, observers) -> List[Proxy]:
-        proxies = []
-        for observer in observers:
-            queue = asyncio.Queue()
-            task = self._task_manager.create_task(
-                self._proxy_task_handler(queue, observer),
-                f"TaskObserver::{observer.__class__.__name__}::_proxy_task_handler",
-            )
-            proxy = Proxy(queue=queue, task=task, observer=observer)
-            proxies.append(proxy)
-        return proxies
-
-    async def _proxy_task_handler(self, queue: asyncio.Queue, observer: BaseObserver):
-        while True:
-            data = await queue.get()
-            await observer.on_push_frame(
-                data.src, data.dst, data.frame, data.direction, data.timestamp
-            )

src/pipecat/processors/frame_processor.py

@@ -5,8 +5,9 @@
 #
 
 import asyncio
+import inspect
 from enum import Enum
-from typing import Awaitable, Callable, Coroutine, Optional
+from typing import Awaitable, Callable, Coroutine, List, Optional, Sequence
 
 from loguru import logger
 
@@ -21,8 +22,9 @@ from pipecat.frames.frames import (
     SystemFrame,
 )
 from pipecat.metrics.metrics import LLMTokenUsage, MetricsData
+from pipecat.observers.base_observer import BaseObserver, FramePushed
 from pipecat.processors.metrics.frame_processor_metrics import FrameProcessorMetrics
-from pipecat.utils.asyncio import BaseTaskManager
+from pipecat.utils.asyncio import BaseTaskManager, TaskManager
 from pipecat.utils.base_object import BaseObject
 
 
@@ -31,6 +33,52 @@ class FrameDirection(Enum):
     UPSTREAM = 2
 
 
+class FrameObserverProxy:
+    def __init__(self, task_manager: TaskManager, observer: BaseObserver) -> None:
+        self._task_manager = task_manager
+        self._observer = observer
+        self._queue = asyncio.Queue()
+        self._task: Optional[asyncio.Task] = None
+        self._warning_reported = False
+
+    def start(self):
+        if not self._task:
+            self._task = self._task_manager.create_task(
+                self._observer_task_handler(), f"ObserverProxy::{self._observer.__class__.__name__}"
+            )
+
+    async def stop(self):
+        if self._task:
+            await self._task_manager.cancel_task(self._task)
+            self._task = None
+
+    async def observe(self, data: FramePushed):
+        await self._queue.put(data)
+
+    async def _observer_task_handler(self):
+        while True:
+            data = await self._queue.get()
+
+            signature = inspect.signature(self._observer.on_push_frame)
+            if len(signature.parameters) > 1:
+                if not self._warning_reported:
+                    import warnings
+
+                    with warnings.catch_warnings():
+                        warnings.simplefilter("always")
+                        warnings.warn(
+                            "Observer `on_push_frame(source, destination, frame, direction, timestamp)` is deprecated, use `on_push_frame(data: FramePushed)` instead.",
+                            DeprecationWarning,
+                        )
+                    self._warning_reported = True
+                await self._observer.on_push_frame(
+                    data.source, data.destination, data.frame, data.direction, data.timestamp
+                )
+            else:
+                await self._observer.on_push_frame(data)
+            self._queue.task_done()
+
+
 class FrameProcessor(BaseObject):
     def __init__(
         self,
@@ -55,7 +103,7 @@ class FrameProcessor(BaseObject):
         self._enable_metrics = False
         self._enable_usage_metrics = False
         self._report_only_initial_ttfb = False
-        self._observer = None
+        self._observer_proxies: List[FrameObserverProxy] = []
 
         # Cancellation is done through CancelFrame (a system frame). This could
         # cause other events being triggered (e.g. closing a transport) which
@@ -77,10 +125,16 @@ class FrameProcessor(BaseObject):
         self.__input_frame_task: Optional[asyncio.Task] = None
 
         # Every processor in Pipecat should only output frames from a single
-        # task. This avoid problems like audio overlapping. System frames are the
-        # exception to this rule. This create this task.
+        # task. This avoids problems like audio overlapping. System frames are
+        # the exception to this rule.
         self.__push_frame_task: Optional[asyncio.Task] = None
 
+        # The observers task will push observed frames to the observers'
+        # proxies. This task avoids the pipeline to block. Then, there is one
+        # proxy per observer and each proxy has its own task. This way each
+        # observer is independent to the rest.
+        self.__observers_task: Optional[asyncio.Task] = None
+
     @property
     def id(self) -> int:
         return self._id
@@ -170,6 +224,7 @@ class FrameProcessor(BaseObject):
         await super().cleanup()
         await self.__cancel_input_task()
         await self.__cancel_push_task()
+        await self.__cancel_observers_task()
 
     def link(self, processor: "FrameProcessor"):
         self._next = processor
@@ -232,7 +287,9 @@ class FrameProcessor(BaseObject):
             self._enable_metrics = frame.enable_metrics
             self._enable_usage_metrics = frame.enable_usage_metrics
             self._report_only_initial_ttfb = frame.report_only_initial_ttfb
-            self._observer = frame.observer
+            self._observer_proxies = self._create_observer_proxies(
+                frame.task_manager, frame.observers
+            )
             await self.__start(frame)
         elif isinstance(frame, StartInterruptionFrame):
             await self._start_interruption()
@@ -255,6 +312,7 @@ class FrameProcessor(BaseObject):
             await self.__push_queue.put((frame, direction))
 
     async def __start(self, frame: StartFrame):
+        self.__create_observers_task()
         self.__create_input_task()
         self.__create_push_task()
 
@@ -262,6 +320,7 @@ class FrameProcessor(BaseObject):
         self._cancelling = True
         await self.__cancel_input_task()
         await self.__cancel_push_task()
+        await self.__cancel_observers_task()
 
     #
     # Handle interruptions
@@ -294,17 +353,25 @@ class FrameProcessor(BaseObject):
             timestamp = self._clock.get_time() if self._clock else 0
             if direction == FrameDirection.DOWNSTREAM and self._next:
                 logger.trace(f"Pushing {frame} from {self} to {self._next}")
-                if self._observer:
-                    await self._observer.on_push_frame(
-                        self, self._next, frame, direction, timestamp
-                    )
+                data = FramePushed(
+                    source=self,
+                    destination=self._next,
+                    frame=frame,
+                    direction=direction,
+                    timestamp=timestamp,
+                )
+                await self.__observe_frame(data)
                 await self._next.queue_frame(frame, direction)
             elif direction == FrameDirection.UPSTREAM and self._prev:
                 logger.trace(f"Pushing {frame} upstream from {self} to {self._prev}")
-                if self._observer:
-                    await self._observer.on_push_frame(
-                        self, self._prev, frame, direction, timestamp
-                    )
+                data = FramePushed(
+                    source=self,
+                    destination=self._prev,
+                    frame=frame,
+                    direction=direction,
+                    timestamp=timestamp,
+                )
+                await self.__observe_frame(data)
                 await self._prev.queue_frame(frame, direction)
         except Exception as e:
             logger.exception(f"Uncaught exception in {self}: {e}")
@@ -368,3 +435,36 @@ class FrameProcessor(BaseObject):
             (frame, direction) = await self.__push_queue.get()
             await self.__internal_push_frame(frame, direction)
             self.__push_queue.task_done()
+
+    def _create_observer_proxies(
+        self, task_manager: TaskManager, observers: Sequence[BaseObserver]
+    ) -> List[FrameObserverProxy]:
+        result = []
+        for observer in observers:
+            proxy = FrameObserverProxy(task_manager, observer)
+            proxy.start()
+            result.append(proxy)
+        return result
+
+    def __create_observers_task(self):
+        if not self.__observers_task:
+            self.__observers_queue = asyncio.Queue()
+            self.__observers_task = self.create_task(self.__observers_task_handler())
+
+    async def __cancel_observers_task(self):
+        if self.__observers_task:
+            for proxy in self._observer_proxies:
+                await proxy.stop()
+            await self.cancel_task(self.__observers_task)
+            self.__observers_task = None
+
+    async def __observe_frame(self, data: FramePushed):
+        await self.__observers_queue.put(data)
+
+    async def __observers_task_handler(self):
+        while True:
+            data = await self.__observers_queue.get()
+            # Proxy observation to all observers.
+            for observer in self._observer_proxies:
+                await observer.observe(data)
+            self.__observers_queue.task_done()

src/pipecat/processors/frameworks/rtvi.py

@@ -55,7 +55,7 @@ from pipecat.metrics.metrics import (
     TTFBMetricsData,
     TTSUsageMetricsData,
 )
-from pipecat.observers.base_observer import BaseObserver
+from pipecat.observers.base_observer import BaseObserver, FramePushed
 from pipecat.processors.aggregators.openai_llm_context import (
     OpenAILLMContext,
     OpenAILLMContextFrame,
@@ -445,14 +445,7 @@ class RTVIObserver(BaseObserver):
         self._frames_seen = set()
         rtvi.set_errors_enabled(self._params.errors_enabled)
 
-    async def on_push_frame(
-        self,
-        src: FrameProcessor,
-        dst: FrameProcessor,
-        frame: Frame,
-        direction: FrameDirection,
-        timestamp: int,
-    ):
+    async def on_push_frame(self, data: FramePushed):
         """Process a frame being pushed through the pipeline.
 
         Args:
@@ -462,6 +455,10 @@ class RTVIObserver(BaseObserver):
             direction: Direction of frame flow in pipeline
             timestamp: Time when frame was pushed
         """
+        src = data.source
+        frame = data.frame
+        direction = data.direction
+
         # If we have already seen this frame, let's skip it.
         if frame.id in self._frames_seen:
             return

src/pipecat/services/google/rtvi.py

@@ -9,8 +9,9 @@ from typing import List, Literal, Optional
 from pydantic import BaseModel
 
 from pipecat.frames.frames import Frame
+from pipecat.observers.base_observer import FramePushed
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.processors.frameworks.rtvi import RTVIObserver
+from pipecat.processors.frameworks.rtvi import RTVIObserver, RTVIProcessor
 from pipecat.services.google.frames import LLMSearchOrigin, LLMSearchResponseFrame
 
 
@@ -27,18 +28,13 @@ class RTVIBotLLMSearchResponseMessage(BaseModel):
 
 
 class GoogleRTVIObserver(RTVIObserver):
-    def __init__(self, rtvi: FrameProcessor):
+    def __init__(self, rtvi: RTVIProcessor):
         super().__init__(rtvi)
 
-    async def on_push_frame(
-        self,
-        src: FrameProcessor,
-        dst: FrameProcessor,
-        frame: Frame,
-        direction: FrameDirection,
-        timestamp: int,
-    ):
-        await super().on_push_frame(src, dst, frame, direction, timestamp)
+    async def on_push_frame(self, data: FramePushed):
+        await super().on_push_frame(data)
+
+        frame = data.frame
 
         if isinstance(frame, LLMSearchResponseFrame):
             await self._handle_llm_search_response_frame(frame)

src/pipecat/services/rime/tts.py

@@ -400,6 +400,13 @@ class RimeHttpTTSService(TTSService):
         payload["modelId"] = self._model_name
         payload["samplingRate"] = self.sample_rate
 
+        # Arcana does not support PCM audio
+        if payload["modelId"] == "arcana":
+            headers["Accept"] = "audio/wav"
+            need_to_strip_wav_header = True
+        else:
+            need_to_strip_wav_header = False
+
         try:
             await self.start_ttfb_metrics()
 
@@ -420,6 +427,10 @@ class RimeHttpTTSService(TTSService):
                 CHUNK_SIZE = 1024
 
                 async for chunk in response.content.iter_chunked(CHUNK_SIZE):
+                    if need_to_strip_wav_header and chunk.startswith(b"RIFF"):
+                        chunk = chunk[44:]
+                        need_to_strip_wav_header = False
+
                     if len(chunk) > 0:
                         await self.stop_ttfb_metrics()
                         frame = TTSAudioRawFrame(chunk, self.sample_rate, 1)

src/pipecat/services/ultravox/stt.py

@@ -425,7 +425,7 @@ class UltravoxSTTService(AIService):
                             if "content" in delta:
                                 new_text = delta["content"]
                                 if new_text:
-                                    yield LLMTextFrame(text=new_text.strip())
+                                    yield LLMTextFrame(text=new_text)
 
                     # Stop processing metrics after completion
                     await self.stop_processing_metrics()

src/pipecat/tests/utils.py

@@ -15,7 +15,7 @@ from pipecat.frames.frames import (
     StartFrame,
     SystemFrame,
 )
-from pipecat.observers.base_observer import BaseObserver
+from pipecat.observers.base_observer import BaseObserver, FramePushed
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
@@ -42,14 +42,10 @@ class HeartbeatsObserver(BaseObserver):
         self._target = target
         self._callback = heartbeat_callback
 
-    async def on_push_frame(
-        self,
-        src: FrameProcessor,
-        dst: FrameProcessor,
-        frame: Frame,
-        direction: FrameDirection,
-        timestamp: int,
-    ):
+    async def on_push_frame(self, data: FramePushed):
+        src = data.source
+        frame = data.frame
+
         if src == self._target and isinstance(frame, HeartbeatFrame):
             await self._callback(self._target, frame)
 

src/pipecat/transports/services/daily.py

@@ -578,6 +578,10 @@ class DailyTransportClient(EventHandler):
         if self._params.transcription_enabled:
             await self._stop_transcription()
 
+        # Remove any custom tracks, if any.
+        for track_name, _ in self._audio_sources.items():
+            await self.remove_custom_audio_track(track_name)
+
         try:
             error = await self._leave()
             if not error:
@@ -741,6 +745,14 @@ class DailyTransportClient(EventHandler):
 
         return audio_source
 
+    async def remove_custom_audio_track(self, track_name: str):
+        future = self._get_event_loop().create_future()
+        self._client.remove_custom_audio_track(
+            track_name=track_name,
+            completion=completion_callback(future),
+        )
+        await future
+
     async def update_transcription(self, participants=None, instance_id=None):
         future = self._get_event_loop().create_future()
         self._client.update_transcription(