Merge pull request #1820 from pipecat-ai/tavus_video_service

Tavus improvements

CHANGELOG.md

@@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added `TavusTransport`, a new transport implementation compatible with any 
+  Pipecat pipeline. When using the `TavusTransport`the Pipecat bot will 
+  connect in the same room as the Tavus Avatar and the user.
+
 - Added `UserBotLatencyLogObserver`. This is an observer that logs the latency
   between when the user stops speaking and when the bot starts speaking. This
   gives you an initial idea on how quickly the AI services respond.
@@ -80,6 +84,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- ⚠️Refactored the `TavusVideoService`, so it acts like a proxy, sending audio to 
+  Tavus and receiving both audio and video. This will make `TavusVideoService` usable
+  with any Pipecat pipeline and with any transport. This is a **breaking change**,
+  check the `examples/foundational/21a-tavus-layer-small-webrtc.py` to see how to
+  use it.
+
 - `DailyTransport` now uses custom microphone audio tracks instead of virtual
   microphones. Now, multiple Daily transports can be used in the same process.
 

examples/foundational/21-tavus-layer-tavus-transport.py

@@ -0,0 +1,112 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+import sys
+
+import aiohttp
+from dotenv import load_dotenv
+from loguru import logger
+
+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.deepgram.stt import DeepgramSTTService
+from pipecat.services.google.llm import GoogleLLMService
+from pipecat.transports.services.tavus import TavusParams, TavusTransport
+
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+async def main():
+    async with aiohttp.ClientSession() as session:
+        transport = TavusTransport(
+            bot_name="Pipecat bot",
+            api_key=os.getenv("TAVUS_API_KEY"),
+            replica_id=os.getenv("TAVUS_REPLICA_ID"),
+            session=session,
+            params=TavusParams(
+                audio_in_enabled=True,
+                audio_out_enabled=True,
+                microphone_out_enabled=False,
+                vad_analyzer=SileroVADAnalyzer(),
+            ),
+        )
+
+        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+        tts = CartesiaTTSService(
+            api_key=os.getenv("CARTESIA_API_KEY"),
+            voice_id="a167e0f3-df7e-4d52-a9c3-f949145efdab",
+        )
+
+        llm = GoogleLLMService(api_key=os.getenv("GOOGLE_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. 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.",
+            },
+        ]
+
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
+
+        pipeline = Pipeline(
+            [
+                transport.input(),  # Transport user input
+                stt,  # STT
+                context_aggregator.user(),  # User responses
+                llm,  # LLM
+                tts,  # TTS
+                transport.output(),  # Transport bot output
+                context_aggregator.assistant(),  # Assistant spoken responses
+            ]
+        )
+
+        task = PipelineTask(
+            pipeline,
+            params=PipelineParams(
+                audio_in_sample_rate=16000,
+                audio_out_sample_rate=24000,
+                allow_interruptions=True,
+                enable_metrics=True,
+                enable_usage_metrics=True,
+                report_only_initial_ttfb=True,
+            ),
+        )
+
+        @transport.event_handler("on_client_connected")
+        async def on_client_connected(transport, participant):
+            logger.info(f"Client connected")
+            # Kick off the conversation.
+            messages.append(
+                {
+                    "role": "system",
+                    "content": "Start by greeting the user and ask how you can help.",
+                }
+            )
+            await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+        @transport.event_handler("on_client_disconnected")
+        async def on_client_disconnected(transport, participant):
+            logger.info(f"Client disconnected")
+            await task.cancel()
+
+        runner = PipelineRunner(handle_sigint=False)
+
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/foundational/21a-tavus-layer-small-webrtc.py

@@ -0,0 +1,125 @@
+#
+# 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.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.deepgram.stt import DeepgramSTTService
+from pipecat.services.google.llm import GoogleLLMService
+from pipecat.services.tavus.video import TavusVideoService
+from pipecat.transports.base_transport import TransportParams
+from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
+from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
+
+load_dotenv(override=True)
+
+
+async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):
+    logger.info(f"Starting bot")
+    async with aiohttp.ClientSession() as session:
+        transport = SmallWebRTCTransport(
+            webrtc_connection=webrtc_connection,
+            params=TransportParams(
+                audio_in_enabled=True,
+                audio_out_enabled=True,
+                video_out_enabled=True,
+                video_out_is_live=True,
+                vad_analyzer=SileroVADAnalyzer(),
+                video_out_width=1280,
+                video_out_height=720,
+            ),
+        )
+
+        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+        tts = CartesiaTTSService(
+            api_key=os.getenv("CARTESIA_API_KEY"),
+            voice_id="a167e0f3-df7e-4d52-a9c3-f949145efdab",
+        )
+
+        llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"))
+
+        tavus = TavusVideoService(
+            api_key=os.getenv("TAVUS_API_KEY"),
+            replica_id=os.getenv("TAVUS_REPLICA_ID"),
+            session=session,
+        )
+
+        messages = [
+            {
+                "role": "system",
+                "content": "You are a helpful LLM in a WebRTC call. 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.",
+            },
+        ]
+
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
+
+        pipeline = Pipeline(
+            [
+                transport.input(),  # Transport user input
+                stt,  # STT
+                context_aggregator.user(),  # User responses
+                llm,  # LLM
+                tts,  # TTS
+                tavus,  # Tavus output layer
+                transport.output(),  # Transport bot output
+                context_aggregator.assistant(),  # Assistant spoken responses
+            ]
+        )
+
+        task = PipelineTask(
+            pipeline,
+            params=PipelineParams(
+                audio_in_sample_rate=16000,
+                audio_out_sample_rate=24000,
+                allow_interruptions=True,
+                enable_metrics=True,
+                enable_usage_metrics=True,
+                report_only_initial_ttfb=True,
+            ),
+        )
+
+        @transport.event_handler("on_client_connected")
+        async def on_client_connected(transport, client):
+            logger.info(f"Client connected")
+            # Kick off the conversation.
+            messages.append(
+                {
+                    "role": "system",
+                    "content": "Start by greeting the user and ask how you can help.",
+                }
+            )
+            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()

examples/foundational/21b-tavus-layer-daily-transport.py

@@ -7,9 +7,9 @@
 import asyncio
 import os
 import sys
-from typing import Any, Mapping
 
 import aiohttp
+from daily_runner import configure
 from dotenv import load_dotenv
 from loguru import logger
 
@@ -20,7 +20,7 @@ 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.deepgram.stt import DeepgramSTTService
-from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.services.google.llm import GoogleLLMService
 from pipecat.services.tavus.video import TavusVideoService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 
@@ -32,23 +32,20 @@ logger.add(sys.stderr, level="DEBUG")
 
 async def main():
     async with aiohttp.ClientSession() as session:
-        tavus = TavusVideoService(
+        (room_url, token) = await configure(session)
-            api_key=os.getenv("TAVUS_API_KEY"),
-            replica_id=os.getenv("TAVUS_REPLICA_ID"),
-            session=session,
-        )
-
-        # get persona, look up persona_name, set this as the bot name to ignore
-        persona_name = await tavus.get_persona_name()
-        room_url = await tavus.initialize()
 
         transport = DailyTransport(
-            room_url=room_url,
+            room_url,
-            token=None,
+            token,
-            bot_name="Pipecat bot",
+            "Pipecat bot",
-            params=DailyParams(
+            DailyParams(
                 audio_in_enabled=True,
+                audio_out_enabled=True,
+                video_out_enabled=True,
+                video_out_is_live=True,
                 vad_analyzer=SileroVADAnalyzer(),
+                video_out_width=1280,
+                video_out_height=720,
             ),
         )
 
@@ -59,7 +56,13 @@ async def main():
             voice_id="a167e0f3-df7e-4d52-a9c3-f949145efdab",
         )
 
-        llm = OpenAILLMService(model="gpt-4o-mini")
+        llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"))
+
+        tavus = TavusVideoService(
+            api_key=os.getenv("TAVUS_API_KEY"),
+            replica_id=os.getenv("TAVUS_REPLICA_ID"),
+            session=session,
+        )
 
         messages = [
             {
@@ -87,10 +90,8 @@ async def main():
         task = PipelineTask(
             pipeline,
             params=PipelineParams(
-                # We just use 16000 because that's what Tavus is expecting and
-                # we avoid resampling.
                 audio_in_sample_rate=16000,
-                audio_out_sample_rate=16000,
+                audio_out_sample_rate=24000,
                 allow_interruptions=True,
                 enable_metrics=True,
                 enable_usage_metrics=True,
@@ -98,33 +99,22 @@ async def main():
             ),
         )
 
-        @transport.event_handler("on_participant_joined")
+        @transport.event_handler("on_first_participant_joined")
-        async def on_participant_joined(
+        async def on_first_participant_joined(transport, participant):
-            transport: DailyTransport, participant: Mapping[str, Any]
+            # Kick off the conversation.
-        ) -> None:
+            messages.append(
-            # Ignore the Tavus replica's microphone
+                {
-            if participant.get("info", {}).get("userName", "") == persona_name:
+                    "role": "system",
-                logger.debug(f"Ignoring {participant['id']}'s microphone")
+                    "content": "Start by greeting the user and ask how you can help.",
-                await transport.update_subscriptions(
+                }
-                    participant_settings={
+            )
-                        participant["id"]: {
+            await task.queue_frames([context_aggregator.user().get_context_frame()])
-                            "media": {"microphone": "unsubscribed"},
-                        }
-                    }
-                )
-
-            if participant.get("info", {}).get("userName", "") != persona_name:
-                # Kick off the conversation.
-                messages.append(
-                    {"role": "system", "content": "Please introduce yourself to the user."}
-                )
-                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()
 
-        runner = PipelineRunner()
+        runner = PipelineRunner(handle_sigint=False)
 
         await runner.run(task)
 

examples/foundational/README.md

@@ -95,7 +95,7 @@ Depending on what you're trying to build, these learning paths will guide you th
 
 - **[18-gstreamer-filesrc.py](./18-gstreamer-filesrc.py)**: GStreamer video streaming (Video processing)
 - **[19-openai-realtime-beta.py](./19-openai-realtime-beta.py)**: OpenAI Speech-to-Speech (Direct S2S, Function calls)
-- **[21-tavus-layer.py](./21-tavus-layer.py)**: Tavus digital twin (Avatar integration)
+- **[21-tavus-layer-tavus-transport.py](./21-tavus-layer-tavus-transport.py)**: Tavus digital twin (Avatar integration)
 - **[27-simli-layer.py](./27-simli-layer.py)**: Simli avatar integration (Video synchronization)
 
 ### Performance & Optimization

src/pipecat/services/tavus/video.py

@@ -7,10 +7,11 @@
 """This module implements Tavus as a sink transport layer"""
 
 import asyncio
-import base64
+import time
 from typing import Optional
 
 import aiohttp
+from daily.daily import AudioData, VideoFrame
 from loguru import logger
 
 from pipecat.audio.utils import create_default_resampler
@@ -18,19 +19,38 @@ from pipecat.frames.frames import (
     CancelFrame,
     EndFrame,
     Frame,
+    OutputAudioRawFrame,
+    OutputImageRawFrame,
     StartFrame,
     StartInterruptionFrame,
-    TransportMessageUrgentFrame,
     TTSAudioRawFrame,
     TTSStartedFrame,
     TTSStoppedFrame,
 )
-from pipecat.processors.frame_processor import FrameDirection
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessorSetup
 from pipecat.services.ai_service import AIService
+from pipecat.transports.services.tavus import TavusCallbacks, TavusParams, TavusTransportClient
 
 
 class TavusVideoService(AIService):
-    """Class to send base64 encoded audio to Tavus"""
+    """
+    Service class that proxies audio to Tavus and receives both audio and video in return.
+
+    It uses the `TavusTransportClient` to manage the session and handle communication. When
+    audio is sent, Tavus responds with both audio and video streams, which are then routed
+    through Pipecat’s media pipeline.
+
+    In use cases such as with `DailyTransport`, this results in two distinct virtual rooms:
+        - **Tavus room**: Contains the Tavus Avatar and the Pipecat Bot.
+        - **User room**: Contains the Pipecat Bot and the user.
+
+    Args:
+        api_key (str): Tavus API key used for authentication.
+        replica_id (str): ID of the Tavus voice replica to use for speech synthesis.
+        persona_id (str): ID of the Tavus persona. Defaults to "pipecat0" to use the Pipecat TTS voice.
+        session (aiohttp.ClientSession): Async HTTP session used for communication with Tavus.
+        **kwargs: Additional arguments passed to the parent `AIService` class.
+    """
 
     def __init__(
         self,
@@ -39,54 +59,101 @@ class TavusVideoService(AIService):
         replica_id: str,
         persona_id: str = "pipecat0",  # Use `pipecat0` so that your TTS voice is used in place of the Tavus persona
         session: aiohttp.ClientSession,
-        sample_rate: int = 16000,
         **kwargs,
     ) -> None:
         super().__init__(**kwargs)
         self._api_key = api_key
+        self._session = session
         self._replica_id = replica_id
         self._persona_id = persona_id
-        self._session = session
+
-        self._sample_rate = sample_rate
+        self._other_participant_has_joined = False
+        self._client: Optional[TavusTransportClient] = None
 
         self._conversation_id: str
-
         self._resampler = create_default_resampler()
 
         self._audio_buffer = bytearray()
         self._queue = asyncio.Queue()
         self._send_task: Optional[asyncio.Task] = None
 
-    async def initialize(self) -> str:
+    async def setup(self, setup: FrameProcessorSetup):
-        url = "https://tavusapi.com/v2/conversations"
+        await super().setup(setup)
-        headers = {"Content-Type": "application/json", "x-api-key": self._api_key}
+        callbacks = TavusCallbacks(
-        payload = {
+            on_participant_joined=self._on_participant_joined,
-            "replica_id": self._replica_id,
+            on_participant_left=self._on_participant_left,
-            "persona_id": self._persona_id,
+        )
-        }
+        self._client = TavusTransportClient(
-        async with self._session.post(url, headers=headers, json=payload) as r:
+            bot_name="Pipecat",
-            r.raise_for_status()
+            callbacks=callbacks,
-            response_json = await r.json()
+            api_key=self._api_key,
+            replica_id=self._replica_id,
+            persona_id=self._persona_id,
+            session=self._session,
+            params=TavusParams(
+                audio_out_enabled=True,
+                microphone_out_enabled=False,
+                audio_in_enabled=True,
+                video_in_enabled=True,
+                video_out_enabled=True,
+            ),
+        )
+        await self._client.setup(setup)
 
-        logger.debug(f"TavusVideoService joined {response_json['conversation_url']}")
+    async def cleanup(self):
-        self._conversation_id = response_json["conversation_id"]
+        await super().cleanup()
-        return response_json["conversation_url"]
+        await self._client.cleanup()
+        self._client = None
+
+    async def _on_participant_left(self, participant, reason):
+        participant_id = participant["id"]
+        logger.info(f"Participant left {participant_id}, reason: {reason}")
+
+    async def _on_participant_joined(self, participant):
+        participant_id = participant["id"]
+        logger.info(f"Participant joined {participant_id}")
+        if not self._other_participant_has_joined:
+            self._other_participant_has_joined = True
+            await self._client.capture_participant_video(
+                participant_id, self._on_participant_video_frame, 30
+            )
+            await self._client.capture_participant_audio(
+                participant_id=participant_id,
+                callback=self._on_participant_audio_data,
+                sample_rate=self._client.out_sample_rate,
+            )
+
+    async def _on_participant_video_frame(
+        self, participant_id: str, video_frame: VideoFrame, video_source: str
+    ):
+        frame = OutputImageRawFrame(
+            image=video_frame.buffer,
+            size=(video_frame.width, video_frame.height),
+            format=video_frame.color_format,
+        )
+        frame.transport_source = video_source
+        await self.push_frame(frame)
+
+    async def _on_participant_audio_data(
+        self, participant_id: str, audio: AudioData, audio_source: str
+    ):
+        frame = OutputAudioRawFrame(
+            audio=audio.audio_frames,
+            sample_rate=audio.sample_rate,
+            num_channels=audio.num_channels,
+        )
+        frame.transport_source = audio_source
+        await self.push_frame(frame)
 
     def can_generate_metrics(self) -> bool:
         return True
 
     async def get_persona_name(self) -> str:
-        url = f"https://tavusapi.com/v2/personas/{self._persona_id}"
+        return await self._client.get_persona_name()
-        headers = {"Content-Type": "application/json", "x-api-key": self._api_key}
-        async with self._session.get(url, headers=headers) as r:
-            r.raise_for_status()
-            response_json = await r.json()
-
-        logger.debug(f"TavusVideoService persona grabbed {response_json}")
-        return response_json["persona_name"]
 
     async def start(self, frame: StartFrame):
         await super().start(frame)
+        await self._client.start(frame)
         await self._create_send_task()
 
     async def stop(self, frame: EndFrame):
@@ -112,7 +179,7 @@ class TavusVideoService(AIService):
         elif isinstance(frame, TTSAudioRawFrame):
             await self._queue_audio(frame.audio, frame.sample_rate, done=False)
         elif isinstance(frame, TTSStoppedFrame):
-            await self._queue_audio(b"\x00\x00", self._sample_rate, done=True)
+            await self._queue_audio(b"\x00\x00", self._client.in_sample_rate, done=True)
             await self.stop_ttfb_metrics()
             await self.stop_processing_metrics()
         else:
@@ -121,13 +188,11 @@ class TavusVideoService(AIService):
     async def _handle_interruptions(self):
         await self._cancel_send_task()
         await self._create_send_task()
-        await self._send_interrupt_message()
+        await self._client.send_interrupt_message()
 
     async def _end_conversation(self):
-        url = f"https://tavusapi.com/v2/conversations/{self._conversation_id}/end"
+        await self._client.stop()
-        headers = {"Content-Type": "application/json", "x-api-key": self._api_key}
+        self._other_participant_has_joined = False
-        async with self._session.post(url, headers=headers) as r:
-            r.raise_for_status()
 
     async def _queue_audio(self, audio: bytes, in_rate: int, done: bool):
         await self._queue.put((audio, in_rate, done))
@@ -142,6 +207,15 @@ class TavusVideoService(AIService):
             await self.cancel_task(self._send_task)
             self._send_task = None
 
+    # TODO (Filipi): this should be all that is needed use this Microphone Echo mode
+    # https://docs.tavus.io/sections/conversational-video-interface/layers-and-modes-overview#microphone-echo
+    # This would allow us to send an audio stream for the replica to repeat
+    # Checking with Tavus what is the right way to create the Persona to make it work
+    # async def _send_task_handler(self):
+    #    while True:
+    #        (audio, in_rate, done) = await self._queue.get()
+    #        await self._client.write_raw_audio_frames(audio)
+
     async def _send_task_handler(self):
         # Daily app-messages have a 4kb limit and also a rate limit of 20
         # messages per second. Below, we only consider the rate limit because 1
@@ -149,57 +223,39 @@ class TavusVideoService(AIService):
         # 1 channel). So, that is 48000 / 20 = 2400, which is below the 4kb
         # limit (even including base64 encoding). For a sample rate of 16000,
         # that would be 32000 / 20 = 1600.
-        MAX_CHUNK_SIZE = int((self._sample_rate * 2) / 20)
+        sample_rate = self._client.out_sample_rate
-        SLEEP_TIME = 1 / 20
+        MAX_CHUNK_SIZE = int((sample_rate * 2) / 20)
 
         audio_buffer = bytearray()
+        samples_sent = 0
+        start_time = time.time()
+
         while True:
             (audio, in_rate, done) = await self._queue.get()
 
             if done:
                 # Send any remaining audio.
                 if len(audio_buffer) > 0:
-                    await self._encode_audio_and_send(bytes(audio_buffer), done)
+                    await self._client.encode_audio_and_send(
-                await self._encode_audio_and_send(audio, done)
+                        bytes(audio_buffer), done, self._current_idx_str
+                    )
+                await self._client.encode_audio_and_send(audio, done, self._current_idx_str)
                 audio_buffer.clear()
             else:
-                audio = await self._resampler.resample(audio, in_rate, self._sample_rate)
+                audio = await self._resampler.resample(audio, in_rate, sample_rate)
                 audio_buffer.extend(audio)
                 while len(audio_buffer) >= MAX_CHUNK_SIZE:
                     chunk = audio_buffer[:MAX_CHUNK_SIZE]
                     audio_buffer = audio_buffer[MAX_CHUNK_SIZE:]
-                    await self._encode_audio_and_send(bytes(chunk), done)
-                    await asyncio.sleep(SLEEP_TIME)
 
-    async def _encode_audio_and_send(self, audio: bytes, done: bool):
+                    # Compute wait time for synchronization
-        """Encodes audio to base64 and sends it to Tavus"""
+                    wait = start_time + (samples_sent / sample_rate) - time.time()
-        audio_base64 = base64.b64encode(audio).decode("utf-8")
+                    if wait > 0:
-        logger.trace(f"{self}: sending {len(audio)} bytes")
+                        await asyncio.sleep(wait)
-        await self._send_audio_message(audio_base64, done=done)
 
-    async def _send_interrupt_message(self) -> None:
+                    await self._client.encode_audio_and_send(
-        transport_frame = TransportMessageUrgentFrame(
+                        bytes(chunk), done, self._current_idx_str
-            message={
+                    )
-                "message_type": "conversation",
-                "event_type": "conversation.interrupt",
-                "conversation_id": self._conversation_id,
-            }
-        )
-        await self.push_frame(transport_frame)
 
-    async def _send_audio_message(self, audio_base64: str, done: bool):
+                    # Update timestamp based on number of samples sent
-        transport_frame = TransportMessageUrgentFrame(
+                    samples_sent += len(chunk) // 2  # 2 bytes per sample (16-bit)
-            message={
-                "message_type": "conversation",
-                "event_type": "conversation.echo",
-                "conversation_id": self._conversation_id,
-                "properties": {
-                    "modality": "audio",
-                    "inference_id": self._current_idx_str,
-                    "audio": audio_base64,
-                    "done": done,
-                    "sample_rate": self._sample_rate,
-                },
-            }
-        )
-        await self.push_frame(transport_frame)

src/pipecat/transports/network/webrtc_connection.py

@@ -144,6 +144,7 @@ class SmallWebRTCConnection(BaseObject):
         self._renegotiation_in_progress = False
         self._last_received_time = None
         self._message_queue = []
+        self._pending_app_messages = []
 
     def _setup_listeners(self):
         @self._pc.on("datachannel")
@@ -170,7 +171,11 @@ class SmallWebRTCConnection(BaseObject):
                         if json_message["type"] == SIGNALLING_TYPE and json_message.get("message"):
                             self._handle_signalling_message(json_message["message"])
                         else:
-                            await self._call_event_handler("app-message", json_message)
+                            if self.is_connected():
+                                await self._call_event_handler("app-message", json_message)
+                            else:
+                                logger.debug("Client not connected. Queuing app-message.")
+                                self._pending_app_messages.append(json_message)
                 except Exception as e:
                     logger.exception(f"Error parsing JSON message {message}, {e}")
 
@@ -225,6 +230,9 @@ class SmallWebRTCConnection(BaseObject):
         # If we already connected, trigger again the connected event
         if self.is_connected():
             await self._call_event_handler("connected")
+            logger.debug("Flushing pending app-messages")
+            for message in self._pending_app_messages:
+                await self._call_event_handler("app-message", message)
             # We are renegotiating here, because likely we have loose the first video frames
             # and aiortc does not handle that pretty well.
             video_input_track = self.video_input_track()
@@ -293,6 +301,7 @@ class SmallWebRTCConnection(BaseObject):
         if self._pc:
             await self._pc.close()
         self._message_queue.clear()
+        self._pending_app_messages.clear()
         self._track_map = {}
 
     def get_answer(self):

src/pipecat/transports/services/tavus.py

@@ -0,0 +1,532 @@
+import asyncio
+import base64
+import time
+from functools import partial
+from typing import Any, Awaitable, Callable, Mapping, Optional
+
+import aiohttp
+from daily.daily import AudioData
+from loguru import logger
+from pydantic import BaseModel
+
+from pipecat.audio.utils import create_default_resampler
+from pipecat.frames.frames import (
+    CancelFrame,
+    EndFrame,
+    Frame,
+    InputAudioRawFrame,
+    OutputImageRawFrame,
+    StartFrame,
+    StartInterruptionFrame,
+    TransportMessageFrame,
+    TransportMessageUrgentFrame,
+    TTSStartedFrame,
+    TTSStoppedFrame,
+)
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
+from pipecat.transports.base_input import BaseInputTransport
+from pipecat.transports.base_output import BaseOutputTransport
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.services.daily import (
+    DailyCallbacks,
+    DailyParams,
+    DailyTransportClient,
+)
+
+
+class TavusApi:
+    """
+    A helper class for interacting with the Tavus API (v2).
+    """
+
+    BASE_URL = "https://tavusapi.com/v2"
+
+    def __init__(self, api_key: str, session: aiohttp.ClientSession):
+        """
+        Initialize the TavusApi client.
+
+        Args:
+            api_key (str): Tavus API key.
+            session (aiohttp.ClientSession): An aiohttp session for making HTTP requests.
+        """
+        self._api_key = api_key
+        self._session = session
+        self._headers = {"Content-Type": "application/json", "x-api-key": self._api_key}
+
+    async def create_conversation(self, replica_id: str, persona_id: str) -> dict:
+        logger.debug(f"Creating Tavus conversation: replica={replica_id}, persona={persona_id}")
+        url = f"{self.BASE_URL}/conversations"
+        payload = {
+            "replica_id": replica_id,
+            "persona_id": persona_id,
+        }
+        async with self._session.post(url, headers=self._headers, json=payload) as r:
+            r.raise_for_status()
+            response = await r.json()
+            logger.debug(f"Created Tavus conversation: {response}")
+            return response
+
+    async def end_conversation(self, conversation_id: str):
+        if conversation_id is None:
+            return
+
+        url = f"{self.BASE_URL}/conversations/{conversation_id}/end"
+        async with self._session.post(url, headers=self._headers) as r:
+            r.raise_for_status()
+            logger.debug(f"Ended Tavus conversation {conversation_id}")
+
+    async def get_persona_name(self, persona_id: str) -> str:
+        url = f"{self.BASE_URL}/personas/{persona_id}"
+        async with self._session.get(url, headers=self._headers) as r:
+            r.raise_for_status()
+            response = await r.json()
+            logger.debug(f"Fetched Tavus persona: {response}")
+            return response["persona_name"]
+
+
+class TavusCallbacks(BaseModel):
+    """Callback handlers for the Tavus events.
+
+    Attributes:
+        on_participant_joined: Called when a participant joins.
+        on_participant_left: Called when a participant leaves.
+    """
+
+    on_participant_joined: Callable[[Mapping[str, Any]], Awaitable[None]]
+    on_participant_left: Callable[[Mapping[str, Any], str], Awaitable[None]]
+
+
+class TavusParams(DailyParams):
+    """Configuration parameters for the Tavus transport."""
+
+    audio_in_enabled: bool = True
+    audio_out_enabled: bool = True
+    microphone_out_enabled: bool = False
+
+
+class TavusTransportClient:
+    """
+    A transport client that integrates a Pipecat Bot with the Tavus platform by managing
+    conversation sessions using the Tavus API.
+
+    This client uses `TavusApi` to interact with the Tavus backend services. When a conversation
+    is started via `TavusApi`, Tavus provides a `roomURL` that can be used to connect the Pipecat Bot
+    into the same virtual room where the TavusBot is operating.
+
+    Args:
+        bot_name (str): The name of the Pipecat bot instance.
+        params (TavusParams): Optional parameters for Tavus operation. Defaults to `TavusParams()`.
+        callbacks (TavusCallbacks): Callback handlers for Tavus-related events.
+        api_key (str): API key for authenticating with Tavus API.
+        replica_id (str): ID of the replica to use in the Tavus conversation.
+        persona_id (str): ID of the Tavus persona. Defaults to "pipecat0", which signals Tavus to use
+                          the TTS voice of the Pipecat bot instead of a Tavus persona voice.
+        session (aiohttp.ClientSession): The aiohttp session for making async HTTP requests.
+        sample_rate: Audio sample rate to be used by the client.
+    """
+
+    def __init__(
+        self,
+        *,
+        bot_name: str,
+        params: TavusParams = TavusParams(),
+        callbacks: TavusCallbacks,
+        api_key: str,
+        replica_id: str,
+        persona_id: str = "pipecat0",  # Use `pipecat0` so that your TTS voice is used in place of the Tavus persona
+        session: aiohttp.ClientSession,
+    ) -> None:
+        self._bot_name = bot_name
+        self._api = TavusApi(api_key, session)
+        self._replica_id = replica_id
+        self._persona_id = persona_id
+        self._conversation_id: Optional[str] = None
+        self._other_participant_has_joined = False
+        self._client: Optional[DailyTransportClient] = None
+        self._callbacks = callbacks
+        self._params = params
+
+    async def _initialize(self) -> str:
+        response = await self._api.create_conversation(self._replica_id, self._persona_id)
+        self._conversation_id = response["conversation_id"]
+        return response["conversation_url"]
+
+    async def setup(self, setup: FrameProcessorSetup):
+        if self._conversation_id is not None:
+            return
+        try:
+            room_url = await self._initialize()
+            daily_callbacks = DailyCallbacks(
+                on_active_speaker_changed=partial(
+                    self._on_handle_callback, "on_active_speaker_changed"
+                ),
+                on_joined=self._on_joined,
+                on_left=self._on_left,
+                on_error=partial(self._on_handle_callback, "on_error"),
+                on_app_message=partial(self._on_handle_callback, "on_app_message"),
+                on_call_state_updated=partial(self._on_handle_callback, "on_call_state_updated"),
+                on_client_connected=partial(self._on_handle_callback, "on_client_connected"),
+                on_client_disconnected=partial(self._on_handle_callback, "on_client_disconnected"),
+                on_dialin_connected=partial(self._on_handle_callback, "on_dialin_connected"),
+                on_dialin_ready=partial(self._on_handle_callback, "on_dialin_ready"),
+                on_dialin_stopped=partial(self._on_handle_callback, "on_dialin_stopped"),
+                on_dialin_error=partial(self._on_handle_callback, "on_dialin_error"),
+                on_dialin_warning=partial(self._on_handle_callback, "on_dialin_warning"),
+                on_dialout_answered=partial(self._on_handle_callback, "on_dialout_answered"),
+                on_dialout_connected=partial(self._on_handle_callback, "on_dialout_connected"),
+                on_dialout_stopped=partial(self._on_handle_callback, "on_dialout_stopped"),
+                on_dialout_error=partial(self._on_handle_callback, "on_dialout_error"),
+                on_dialout_warning=partial(self._on_handle_callback, "on_dialout_warning"),
+                on_participant_joined=self._callbacks.on_participant_joined,
+                on_participant_left=self._callbacks.on_participant_left,
+                on_participant_updated=partial(self._on_handle_callback, "on_participant_updated"),
+                on_transcription_message=partial(
+                    self._on_handle_callback, "on_transcription_message"
+                ),
+                on_recording_started=partial(self._on_handle_callback, "on_recording_started"),
+                on_recording_stopped=partial(self._on_handle_callback, "on_recording_stopped"),
+                on_recording_error=partial(self._on_handle_callback, "on_recording_error"),
+            )
+            self._client = DailyTransportClient(
+                room_url, None, "Pipecat", self._params, daily_callbacks, self._bot_name
+            )
+            await self._client.setup(setup)
+        except Exception as e:
+            logger.error(f"Failed to setup TavusTransportClient: {e}")
+            await self._api.end_conversation(self._conversation_id)
+
+    async def cleanup(self):
+        if self._client is None:
+            return
+        await self._client.cleanup()
+        self._client = None
+
+    async def _on_joined(self, data):
+        logger.debug("TavusTransportClient joined!")
+
+    async def _on_left(self):
+        logger.debug("TavusTransportClient left!")
+
+    async def _on_handle_callback(self, event_name, *args, **kwargs):
+        logger.trace(f"[Callback] {event_name} called with args={args}, kwargs={kwargs}")
+
+    async def get_persona_name(self) -> str:
+        return await self._api.get_persona_name(self._persona_id)
+
+    async def start(self, frame: StartFrame):
+        logger.debug("TavusTransportClient start invoked!")
+        await self._client.start(frame)
+        await self._client.join()
+
+    async def stop(self):
+        await self._client.leave()
+        await self._api.end_conversation(self._conversation_id)
+
+    async def capture_participant_video(
+        self,
+        participant_id: str,
+        callback: Callable,
+        framerate: int = 30,
+        video_source: str = "camera",
+        color_format: str = "RGB",
+    ):
+        await self._client.capture_participant_video(
+            participant_id, callback, framerate, video_source, color_format
+        )
+
+    async def capture_participant_audio(
+        self,
+        participant_id: str,
+        callback: Callable,
+        audio_source: str = "microphone",
+        sample_rate: int = 16000,
+        callback_interval_ms: int = 20,
+    ):
+        await self._client.capture_participant_audio(
+            participant_id, callback, audio_source, sample_rate, callback_interval_ms
+        )
+
+    async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame):
+        await self._client.send_message(frame)
+
+    @property
+    def out_sample_rate(self) -> int:
+        return self._client.out_sample_rate
+
+    @property
+    def in_sample_rate(self) -> int:
+        return self._client.in_sample_rate
+
+    async def encode_audio_and_send(self, audio: bytes, done: bool, inference_id: str):
+        """Encodes audio to base64 and sends it to Tavus"""
+        audio_base64 = base64.b64encode(audio).decode("utf-8")
+        await self._send_audio_message(audio_base64, done=done, inference_id=inference_id)
+
+    async def send_interrupt_message(self) -> None:
+        transport_frame = TransportMessageUrgentFrame(
+            message={
+                "message_type": "conversation",
+                "event_type": "conversation.interrupt",
+                "conversation_id": self._conversation_id,
+            }
+        )
+        await self.send_message(transport_frame)
+
+    async def _send_audio_message(self, audio_base64: str, done: bool, inference_id: str):
+        transport_frame = TransportMessageUrgentFrame(
+            message={
+                "message_type": "conversation",
+                "event_type": "conversation.echo",
+                "conversation_id": self._conversation_id,
+                "properties": {
+                    "modality": "audio",
+                    "inference_id": inference_id,
+                    "audio": audio_base64,
+                    "done": done,
+                    "sample_rate": self.out_sample_rate,
+                },
+            }
+        )
+        await self.send_message(transport_frame)
+
+    async def update_subscriptions(self, participant_settings=None, profile_settings=None):
+        await self._client.update_subscriptions(
+            participant_settings=participant_settings, profile_settings=profile_settings
+        )
+
+    async def write_raw_audio_frames(self, frames: bytes, destination: Optional[str] = None):
+        await self._client.write_raw_audio_frames(frames, destination)
+
+
+class TavusInputTransport(BaseInputTransport):
+    def __init__(
+        self,
+        client: TavusTransportClient,
+        params: TransportParams,
+        **kwargs,
+    ):
+        super().__init__(params, **kwargs)
+        self._client = client
+        self._params = params
+        self._resampler = create_default_resampler()
+
+    async def setup(self, setup: FrameProcessorSetup):
+        await super().setup(setup)
+        await self._client.setup(setup)
+
+    async def cleanup(self):
+        await super().cleanup()
+        await self._client.cleanup()
+
+    async def start(self, frame: StartFrame):
+        await super().start(frame)
+        await self._client.start(frame)
+        await self.set_transport_ready(frame)
+
+    async def stop(self, frame: EndFrame):
+        await super().stop(frame)
+        await self._client.stop()
+
+    async def cancel(self, frame: CancelFrame):
+        await super().cancel(frame)
+        await self._client.stop()
+
+    async def start_capturing_audio(self, participant):
+        if self._params.audio_in_enabled:
+            logger.info(
+                f"TavusTransportClient start capturing audio for participant {participant['id']}"
+            )
+            await self._client.capture_participant_audio(
+                participant_id=participant["id"],
+                callback=self._on_participant_audio_data,
+                sample_rate=self._client.in_sample_rate,
+            )
+
+    async def _on_participant_audio_data(
+        self, participant_id: str, audio: AudioData, audio_source: str
+    ):
+        frame = InputAudioRawFrame(
+            audio=audio.audio_frames,
+            sample_rate=audio.audio_frames,
+            num_channels=audio.num_channels,
+        )
+        frame.transport_source = audio_source
+        await self.push_audio_frame(frame)
+
+
+class TavusOutputTransport(BaseOutputTransport):
+    def __init__(
+        self,
+        client: TavusTransportClient,
+        params: TransportParams,
+        **kwargs,
+    ):
+        super().__init__(params, **kwargs)
+        self._client = client
+        self._params = params
+        self._samples_sent = 0
+        self._start_time = time.time()
+
+    async def setup(self, setup: FrameProcessorSetup):
+        await super().setup(setup)
+        await self._client.setup(setup)
+
+    async def cleanup(self):
+        await super().cleanup()
+        await self._client.cleanup()
+
+    async def start(self, frame: StartFrame):
+        await super().start(frame)
+        self._samples_sent = 0
+        self._start_time = time.time()
+        await self._client.start(frame)
+        await self.set_transport_ready(frame)
+
+    async def stop(self, frame: EndFrame):
+        await super().stop(frame)
+        await self._client.stop()
+
+    async def cancel(self, frame: CancelFrame):
+        await super().cancel(frame)
+        await self._client.stop()
+
+    async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame):
+        logger.info(f"TavusOutputTransport sending message {frame}")
+        await self._client.send_message(frame)
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+        if isinstance(frame, StartInterruptionFrame):
+            await self._handle_interruptions()
+        elif isinstance(frame, TTSStartedFrame):
+            self._current_idx_str = str(frame.id)
+        elif isinstance(frame, TTSStoppedFrame):
+            logger.debug(f"TAVUS: {self}: stopped speaking")
+            await self._client.encode_audio_and_send(b"\x00\x00", True, self._current_idx_str)
+
+    async def _handle_interruptions(self):
+        await self._client.send_interrupt_message()
+
+    async def write_raw_audio_frames(self, frames: bytes, destination: Optional[str] = None):
+        # Compute wait time for synchronization
+        wait = self._start_time + (self._samples_sent / self._sample_rate) - time.time()
+        if wait > 0:
+            await asyncio.sleep(wait)
+
+        await self._client.encode_audio_and_send(frames, False, self._current_idx_str)
+
+        # Update timestamp based on number of samples sent
+        self._samples_sent += len(frames) // 2  # 2 bytes per sample (16-bit)
+
+    async def write_raw_video_frame(
+        self, frame: OutputImageRawFrame, destination: Optional[str] = None
+    ):
+        pass
+
+
+class TavusTransport(BaseTransport):
+    """
+    Transport implementation for Tavus video calls.
+
+    When used, the Pipecat bot joins the same virtual room as the Tavus Avatar and the user.
+    This is achieved by using `TavusTransportClient`, which initiates the conversation via
+    `TavusApi` and obtains a room URL that all participants connect to.
+
+    Args:
+        bot_name (str): The name of the Pipecat bot.
+        session (aiohttp.ClientSession): aiohttp session used for async HTTP requests.
+        api_key (str): Tavus API key for authentication.
+        replica_id (str): ID of the replica model used for voice generation.
+        persona_id (str): ID of the Tavus persona. Defaults to "pipecat0" to use the Pipecat TTS voice.
+        params (TavusParams): Optional Tavus-specific configuration parameters.
+        input_name (Optional[str]): Optional name for the input transport.
+        output_name (Optional[str]): Optional name for the output transport.
+    """
+
+    def __init__(
+        self,
+        bot_name: str,
+        session: aiohttp.ClientSession,
+        api_key: str,
+        replica_id: str,
+        persona_id: str = "pipecat0",  # Use `pipecat0` so that your TTS voice is used in place of the Tavus persona
+        params: TavusParams = TavusParams(),
+        input_name: Optional[str] = None,
+        output_name: Optional[str] = None,
+    ):
+        super().__init__(input_name=input_name, output_name=output_name)
+        self._params = params
+
+        # TODO: Filipi - We can remove this if we stop sending the audio through app messages
+        # Limiting this so we don't go over 20 messages per second
+        # each message is going to have 50ms of audio
+        self._params.audio_out_10ms_chunks = 5
+
+        callbacks = TavusCallbacks(
+            on_participant_joined=self._on_participant_joined,
+            on_participant_left=self._on_participant_left,
+        )
+        self._client = TavusTransportClient(
+            bot_name="Pipecat",
+            callbacks=callbacks,
+            api_key=api_key,
+            replica_id=replica_id,
+            persona_id=persona_id,
+            session=session,
+            params=params,
+        )
+        self._input: Optional[TavusInputTransport] = None
+        self._output: Optional[TavusOutputTransport] = None
+        self._tavus_participant_id = None
+
+        # Register supported handlers. The user will only be able to register
+        # these handlers.
+        self._register_event_handler("on_client_connected")
+        self._register_event_handler("on_client_disconnected")
+
+    async def _on_participant_left(self, participant, reason):
+        persona_name = await self._client.get_persona_name()
+        if participant.get("info", {}).get("userName", "") != persona_name:
+            await self._on_client_disconnected(participant)
+
+    async def _on_participant_joined(self, participant):
+        # get persona, look up persona_name, set this as the bot name to ignore
+        persona_name = await self._client.get_persona_name()
+        # Ignore the Tavus replica's microphone
+        if participant.get("info", {}).get("userName", "") == persona_name:
+            self._tavus_participant_id = participant["id"]
+        else:
+            await self._on_client_connected(participant)
+            if self._tavus_participant_id:
+                logger.debug(f"Ignoring {self._tavus_participant_id}'s microphone")
+                await self.update_subscriptions(
+                    participant_settings={
+                        self._tavus_participant_id: {
+                            "media": {"microphone": "unsubscribed"},
+                        }
+                    }
+                )
+            if self._input:
+                await self._input.start_capturing_audio(participant)
+
+    async def update_subscriptions(self, participant_settings=None, profile_settings=None):
+        await self._client.update_subscriptions(
+            participant_settings=participant_settings,
+            profile_settings=profile_settings,
+        )
+
+    def input(self) -> FrameProcessor:
+        if not self._input:
+            self._input = TavusInputTransport(client=self._client, params=self._params)
+        return self._input
+
+    def output(self) -> FrameProcessor:
+        if not self._output:
+            self._output = TavusOutputTransport(client=self._client, params=self._params)
+        return self._output
+
+    async def _on_client_connected(self, participant: Any):
+        await self._call_event_handler("on_client_connected", participant)
+
+    async def _on_client_disconnected(self, participant: Any):
+        await self._call_event_handler("on_client_disconnected", participant)