wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'main' into google-streaming-tts

Browse Source
This commit is contained in:
aristid
2025-05-24 17:16:22 +02:00
committed by GitHub
parent d3e2a9e5c0 f86d002ceb
commit 09ba319f3e
40 changed files with 1795 additions and 824 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/pipecat/audio/vad/silero.py
View File

@@ -138,7 +138,9 @@ class SileroVADAnalyzer(VADAnalyzer):
def set_sample_rate(self, sample_rate: int):
if sample_rate != 16000 and sample_rate != 8000:
raise ValueError("Silero VAD sample rate needs to be 16000 or 8000")
raise ValueError(
f"Silero VAD sample rate needs to be 16000 or 8000 (sample rate: {sample_rate})"
)
super().set_sample_rate(sample_rate)

50
src/pipecat/observers/loggers/user_bot_latency_log_observer.py Normal file
View File

@@ -0,0 +1,50 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import time
from loguru import logger
from pipecat.frames.frames import (
BotStartedSpeakingFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
)
from pipecat.observers.base_observer import BaseObserver, FramePushed
from pipecat.processors.frame_processor import FrameDirection
class UserBotLatencyLogObserver(BaseObserver):
"""Observer that logs the latency between when the user stops speaking and
when the bot starts speaking.
This helps measure how quickly the AI services respond.
"""
def __init__(self):
super().__init__()
self._processed_frames = set()
self._user_stopped_time = 0
async def on_push_frame(self, data: FramePushed):
# Only process downstream frames
if data.direction != FrameDirection.DOWNSTREAM:
return
# Skip already processed frames
if data.frame.id in self._processed_frames:
return
self._processed_frames.add(data.frame.id)
if isinstance(data.frame, UserStartedSpeakingFrame):
self._user_stopped_time = 0
elif isinstance(data.frame, UserStoppedSpeakingFrame):
self._user_stopped_time = time.time()
elif isinstance(data.frame, BotStartedSpeakingFrame) and self._user_stopped_time:
latency = time.time() - self._user_stopped_time
logger.debug(f"⏱️ LATENCY FROM USER STOPPED SPEAKING TO BOT STARTED SPEAKING: {latency}")

2
src/pipecat/processors/frameworks/rtvi.py
View File

@@ -843,7 +843,7 @@ class RTVIProcessor(FrameProcessor):
async def _handle_client_ready(self, request_id: str):
logger.debug("Received client-ready")
if self._input_transport:
self._input_transport.start_audio_in_streaming()
await self._input_transport.start_audio_in_streaming()
self._client_ready_id = request_id
await self.set_client_ready()

2
src/pipecat/services/anthropic/llm.py
View File

@@ -90,7 +90,7 @@ class AnthropicLLMService(LLMService):
self,
*,
api_key: str,
model: str = "claude-3-7-sonnet-20250219",
model: str = "claude-sonnet-4-20250514",
params: Optional[InputParams] = None,
client=None,
**kwargs,

6
src/pipecat/services/elevenlabs/tts.py
View File

@@ -254,14 +254,16 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
async def set_model(self, model: str):
await super().set_model(model)
logger.info(f"Switching TTS model to: [{model}]")
# No need to disconnect/reconnect for model changes with multi-context API
await self._disconnect()
await self._connect()
async def _update_settings(self, settings: Mapping[str, Any]):
prev_voice = self._voice_id
await super()._update_settings(settings)
# If voice changes, we don't need to reconnect, just use a new context
if not prev_voice == self._voice_id:
logger.info(f"Switching TTS voice to: [{self._voice_id}]")
await self._disconnect()
await self._connect()
async def start(self, frame: StartFrame):
await super().start(frame)

2
src/pipecat/services/gemini_multimodal_live/gemini.py
View File

@@ -335,7 +335,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
*,
api_key: str,
base_url: str = "generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent",
model="models/gemini-2.0-flash-live-001",
model="models/gemini-2.5-flash-preview-native-audio-dialog",
voice_id: str = "Charon",
start_audio_paused: bool = False,
start_video_paused: bool = False,

8
src/pipecat/services/sarvam/__init__.py Normal file
View File

@@ -0,0 +1,8 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
from .tts import *

195
src/pipecat/services/sarvam/tts.py Normal file
View File

@@ -0,0 +1,195 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import base64
from typing import AsyncGenerator, Optional
import aiohttp
from loguru import logger
from pydantic import BaseModel, Field
from pipecat.frames.frames import (
ErrorFrame,
Frame,
StartFrame,
TTSAudioRawFrame,
TTSStartedFrame,
TTSStoppedFrame,
)
from pipecat.services.tts_service import TTSService
from pipecat.transcriptions.language import Language
from pipecat.utils.tracing.service_decorators import traced_tts
def language_to_sarvam_language(language: Language) -> Optional[str]:
"""Convert Pipecat Language enum to Sarvam AI language codes."""
LANGUAGE_MAP = {
Language.BN: "bn-IN", # Bengali
Language.EN: "en-IN", # English (India)
Language.GU: "gu-IN", # Gujarati
Language.HI: "hi-IN", # Hindi
Language.KN: "kn-IN", # Kannada
Language.ML: "ml-IN", # Malayalam
Language.MR: "mr-IN", # Marathi
Language.OR: "od-IN", # Odia
Language.PA: "pa-IN", # Punjabi
Language.TA: "ta-IN", # Tamil
Language.TE: "te-IN", # Telugu
}
return LANGUAGE_MAP.get(language)
class SarvamTTSService(TTSService):
"""Text-to-Speech service using Sarvam AI's API.
Converts text to speech using Sarvam AI's TTS models with support for multiple
Indian languages. Provides control over voice characteristics like pitch, pace,
and loudness.
Args:
api_key: Sarvam AI API subscription key.
voice_id: Speaker voice ID (e.g., "anushka", "meera").
model: TTS model to use ("bulbul:v1" or "bulbul:v2").
aiohttp_session: Shared aiohttp session for making requests.
base_url: Sarvam AI API base URL.
sample_rate: Audio sample rate in Hz (8000, 16000, 22050, 24000).
params: Additional voice and preprocessing parameters.
Example:
```python
tts = SarvamTTSService(
api_key="your-api-key",
voice_id="anushka",
model="bulbul:v2",
aiohttp_session=session,
params=SarvamTTSService.InputParams(
language=Language.HI,
pitch=0.1,
pace=1.2
)
)
```
"""
class InputParams(BaseModel):
language: Optional[Language] = Language.EN
pitch: Optional[float] = Field(default=0.0, ge=-0.75, le=0.75)
pace: Optional[float] = Field(default=1.0, ge=0.3, le=3.0)
loudness: Optional[float] = Field(default=1.0, ge=0.1, le=3.0)
enable_preprocessing: Optional[bool] = False
def __init__(
self,
*,
api_key: str,
voice_id: str = "anushka",
model: str = "bulbul:v2",
aiohttp_session: aiohttp.ClientSession,
base_url: str = "https://api.sarvam.ai",
sample_rate: Optional[int] = None,
params: Optional[InputParams] = None,
**kwargs,
):
super().__init__(sample_rate=sample_rate, **kwargs)
params = params or SarvamTTSService.InputParams()
self._api_key = api_key
self._base_url = base_url
self._session = aiohttp_session
self._settings = {
"language": self.language_to_service_language(params.language)
if params.language
else "en-IN",
"pitch": params.pitch,
"pace": params.pace,
"loudness": params.loudness,
"enable_preprocessing": params.enable_preprocessing,
}
self.set_model_name(model)
self.set_voice(voice_id)
def can_generate_metrics(self) -> bool:
return True
def language_to_service_language(self, language: Language) -> Optional[str]:
return language_to_sarvam_language(language)
async def start(self, frame: StartFrame):
await super().start(frame)
self._settings["sample_rate"] = self.sample_rate
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"{self}: Generating TTS [{text}]")
try:
await self.start_ttfb_metrics()
payload = {
"text": text,
"target_language_code": self._settings["language"],
"speaker": self._voice_id,
"pitch": self._settings["pitch"],
"pace": self._settings["pace"],
"loudness": self._settings["loudness"],
"speech_sample_rate": self.sample_rate,
"enable_preprocessing": self._settings["enable_preprocessing"],
"model": self._model_name,
}
headers = {
"api-subscription-key": self._api_key,
"Content-Type": "application/json",
}
url = f"{self._base_url}/text-to-speech"
yield TTSStartedFrame()
async with self._session.post(url, json=payload, headers=headers) as response:
if response.status != 200:
error_text = await response.text()
logger.error(f"Sarvam API error: {error_text}")
await self.push_error(ErrorFrame(f"Sarvam API error: {error_text}"))
return
response_data = await response.json()
await self.start_tts_usage_metrics(text)
# Decode base64 audio data
if "audios" not in response_data or not response_data["audios"]:
logger.error("No audio data received from Sarvam API")
await self.push_error(ErrorFrame("No audio data received"))
return
# Get the first audio (there should be only one for single text input)
base64_audio = response_data["audios"][0]
audio_data = base64.b64decode(base64_audio)
# Strip WAV header (first 44 bytes) if present
if audio_data.startswith(b"RIFF"):
logger.debug("Stripping WAV header from Sarvam audio data")
audio_data = audio_data[44:]
frame = TTSAudioRawFrame(
audio=audio_data,
sample_rate=self.sample_rate,
num_channels=1,
)
yield frame
except Exception as e:
logger.error(f"{self} exception: {e}")
await self.push_error(ErrorFrame(f"Error generating TTS: {e}"))
finally:
await self.stop_ttfb_metrics()
yield TTSStoppedFrame()

197
src/pipecat/services/tavus/video.py
View File

@@ -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,98 @@ 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:
url = "https://tavusapi.com/v2/conversations"
headers = {"Content-Type": "application/json", "x-api-key": self._api_key}
payload = {
"replica_id": self._replica_id,
"persona_id": self._persona_id,
}
async with self._session.post(url, headers=headers, json=payload) as r:
r.raise_for_status()
response_json = await r.json()
async def setup(self, setup: FrameProcessorSetup):
await super().setup(setup)
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=self._api_key,
replica_id=self._replica_id,
persona_id=self._persona_id,
session=self._session,
params=TavusParams(
audio_in_enabled=True,
video_in_enabled=True,
),
)
await self._client.setup(setup)
logger.debug(f"TavusVideoService joined {response_json['conversation_url']}")
self._conversation_id = response_json["conversation_id"]
return response_json["conversation_url"]
async def cleanup(self):
await super().cleanup()
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}"
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"]
return await self._client.get_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 +176,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 +185,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"
headers = {"Content-Type": "application/json", "x-api-key": self._api_key}
async with self._session.post(url, headers=headers) as r:
r.raise_for_status()
await self._client.stop()
self._other_participant_has_joined = False
async def _queue_audio(self, audio: bytes, in_rate: int, done: bool):
await self._queue.put((audio, in_rate, done))
@@ -142,6 +204,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 +220,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)
SLEEP_TIME = 1 / 20
sample_rate = self._client.out_sample_rate
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._encode_audio_and_send(audio, done)
await self._client.encode_audio_and_send(
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):
"""Encodes audio to base64 and sends it to Tavus"""
audio_base64 = base64.b64encode(audio).decode("utf-8")
logger.trace(f"{self}: sending {len(audio)} bytes")
await self._send_audio_message(audio_base64, done=done)
# Compute wait time for synchronization
wait = start_time + (samples_sent / sample_rate) - time.time()
if wait > 0:
await asyncio.sleep(wait)
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.push_frame(transport_frame)
await self._client.encode_audio_and_send(
bytes(chunk), done, self._current_idx_str
)
async def _send_audio_message(self, audio_base64: str, done: bool):
transport_frame = TransportMessageUrgentFrame(
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)
# Update timestamp based on number of samples sent
samples_sent += len(chunk) // 2 # 2 bytes per sample (16-bit)

2
src/pipecat/transports/base_input.py
View File

@@ -101,7 +101,7 @@ class BaseInputTransport(FrameProcessor):
logger.debug(f"Enabling audio on start. {enabled}")
self._params.audio_in_stream_on_start = enabled
def start_audio_in_streaming(self):
async def start_audio_in_streaming(self):
pass
@property

33
src/pipecat/transports/base_output.py
View File

@@ -8,6 +8,7 @@ import asyncio
import itertools
import sys
import time
from concurrent.futures import ThreadPoolExecutor
from typing import Any, AsyncGenerator, Dict, List, Mapping, Optional
from loguru import logger
@@ -234,6 +235,9 @@ class BaseOutputTransport(FrameProcessor):
self._audio_chunk_size = audio_chunk_size
self._params = params
# This is to resize images. We only need to resize one image at a time.
self._executor = ThreadPoolExecutor(max_workers=1)
# Buffer to keep track of incoming audio.
self._audio_buffer = bytearray()
@@ -558,18 +562,25 @@ class BaseOutputTransport(FrameProcessor):
self._video_queue.task_done()
async def _draw_image(self, frame: OutputImageRawFrame):
desired_size = (self._params.video_out_width, self._params.video_out_height)
def resize_frame(frame: OutputImageRawFrame) -> OutputImageRawFrame:
desired_size = (self._params.video_out_width, self._params.video_out_height)
# TODO: we should refactor in the future to support dynamic resolutions
# which is kind of what happens in P2P connections.
# We need to add support for that inside the DailyTransport
if frame.size != desired_size:
image = Image.frombytes(frame.format, frame.size, frame.image)
resized_image = image.resize(desired_size)
# logger.warning(f"{frame} does not have the expected size {desired_size}, resizing")
frame = OutputImageRawFrame(
resized_image.tobytes(), resized_image.size, resized_image.format
)
# TODO: we should refactor in the future to support dynamic resolutions
# which is kind of what happens in P2P connections.
# We need to add support for that inside the DailyTransport
if frame.size != desired_size:
image = Image.frombytes(frame.format, frame.size, frame.image)
resized_image = image.resize(desired_size)
# logger.warning(f"{frame} does not have the expected size {desired_size}, resizing")
frame = OutputImageRawFrame(
resized_image.tobytes(), resized_image.size, resized_image.format
)
return frame
frame = await self._transport.get_event_loop().run_in_executor(
self._executor, resize_frame, frame
)
await self._transport.write_raw_video_frame(frame, self._destination)

11
src/pipecat/transports/network/webrtc_connection.py
View File

@@ -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):

306
src/pipecat/transports/services/daily.py
View File

@@ -14,14 +14,12 @@ import aiohttp
from loguru import logger
from pydantic import BaseModel
from pipecat.audio.utils import create_default_resampler
from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADParams
from pipecat.frames.frames import (
CancelFrame,
EndFrame,
ErrorFrame,
Frame,
InputAudioRawFrame,
InterimTranscriptionFrame,
OutputAudioRawFrame,
OutputImageRawFrame,
@@ -46,12 +44,11 @@ try:
AudioData,
CallClient,
CustomAudioSource,
CustomAudioTrack,
Daily,
EventHandler,
VideoFrame,
VirtualCameraDevice,
VirtualMicrophoneDevice,
VirtualSpeakerDevice,
)
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
@@ -245,6 +242,12 @@ def completion_callback(future):
return _callback
@dataclass
class DailyAudioTrack:
source: CustomAudioSource
track: CustomAudioTrack
class DailyTransportClient(EventHandler):
"""Core client for interacting with Daily's API.
@@ -306,35 +309,33 @@ class DailyTransportClient(EventHandler):
self._client: CallClient = CallClient(event_handler=self)
# We use a separate task to execute the callbacks, otherwise if we call
# a `CallClient` function and wait for its completion this will
# currently result in a deadlock. This is because `_call_async_callback`
# can be used inside `CallClient` event handlers which are holding the
# GIL in `daily-python`. So if the `callback` passed here makes a
# `CallClient` call and waits for it to finish using completions (and a
# future) we will deadlock because completions use event handlers (which
# are holding the GIL).
self._callback_queue = asyncio.Queue()
self._callback_task = None
# We use separate tasks to execute callbacks (events, audio or
# video). In the case of events, if we call a `CallClient` function
# inside the callback and wait for its completion this will result in a
# deadlock (because we haven't exited the event callback). The deadlocks
# occur because `daily-python` is holding the GIL when calling the
# callbacks. So, if our callback handler makes a `CallClient` call and
# waits for it to finish using completions (and a future) we will
# deadlock because completions use event handlers (which are holding the
# GIL).
self._event_queue = asyncio.Queue()
self._audio_queue = asyncio.Queue()
self._video_queue = asyncio.Queue()
self._event_task = None
self._audio_task = None
self._video_task = None
# Input and ouput sample rates. They will be initialize on setup().
self._in_sample_rate = 0
self._out_sample_rate = 0
self._camera: Optional[VirtualCameraDevice] = None
self._mic: Optional[VirtualMicrophoneDevice] = None
self._speaker: Optional[VirtualSpeakerDevice] = None
self._audio_sources: Dict[str, CustomAudioSource] = {}
self._microphone_track: Optional[DailyAudioTrack] = None
self._custom_audio_tracks: Dict[str, DailyAudioTrack] = {}
def _camera_name(self):
return f"camera-{self}"
def _mic_name(self):
return f"mic-{self}"
def _speaker_name(self):
return f"speaker-{self}"
@property
def room_url(self) -> str:
return self._room_url
@@ -365,43 +366,26 @@ class DailyTransportClient(EventHandler):
)
await future
async def read_next_audio_frame(self) -> Optional[InputAudioRawFrame]:
if not self._speaker:
return None
sample_rate = self._in_sample_rate
num_channels = self._params.audio_in_channels
num_frames = int(sample_rate / 100) * 2 # 20ms of audio
future = self._get_event_loop().create_future()
self._speaker.read_frames(num_frames, completion=completion_callback(future))
audio = await future
if len(audio) > 0:
return InputAudioRawFrame(
audio=audio, sample_rate=sample_rate, num_channels=num_channels
)
else:
# If we don't read any audio it could be there's no participant
# connected. daily-python will return immediately if that's the
# case, so let's sleep for a little bit (i.e. busy wait).
await asyncio.sleep(0.01)
return None
async def register_audio_destination(self, destination: str):
self._audio_sources[destination] = await self.add_custom_audio_track(destination)
self._custom_audio_tracks[destination] = await self.add_custom_audio_track(destination)
self._client.update_publishing({"customAudio": {destination: True}})
async def write_raw_audio_frames(self, frames: bytes, destination: Optional[str] = None):
future = self._get_event_loop().create_future()
if not destination and self._mic:
self._mic.write_frames(frames, completion=completion_callback(future))
elif destination and destination in self._audio_sources:
source = self._audio_sources[destination]
source.write_frames(frames, completion=completion_callback(future))
audio_source: Optional[CustomAudioSource] = None
if not destination and self._microphone_track:
audio_source = self._microphone_track.source
elif destination and destination in self._custom_audio_tracks:
track = self._custom_audio_tracks[destination]
audio_source = track.source
if audio_source:
audio_source.write_frames(frames, completion=completion_callback(future))
else:
logger.warning(f"{self} unable to write audio frames to destination [{destination}]")
future.set_result(None)
await future
async def write_raw_video_frame(
@@ -415,15 +399,21 @@ class DailyTransportClient(EventHandler):
return
self._task_manager = setup.task_manager
self._callback_task = self._task_manager.create_task(
self._callback_task_handler(),
f"{self}::callback_task",
self._event_task = self._task_manager.create_task(
self._callback_task_handler(self._event_queue),
f"{self}::event_callback_task",
)
async def cleanup(self):
if self._callback_task and self._task_manager:
await self._task_manager.cancel_task(self._callback_task)
self._callback_task = None
if self._event_task and self._task_manager:
await self._task_manager.cancel_task(self._event_task)
self._event_task = None
if self._audio_task and self._task_manager:
await self._task_manager.cancel_task(self._audio_task)
self._audio_task = None
if self._video_task and self._task_manager:
await self._task_manager.cancel_task(self._video_task)
self._video_task = None
# Make sure we don't block the event loop in case `client.release()`
# takes extra time.
await self._get_event_loop().run_in_executor(self._executor, self._cleanup)
@@ -432,6 +422,17 @@ class DailyTransportClient(EventHandler):
self._in_sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate
self._out_sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate
if self._params.audio_in_enabled and not self._audio_task and self._task_manager:
self._audio_task = self._task_manager.create_task(
self._callback_task_handler(self._audio_queue),
f"{self}::audio_callback_task",
)
if self._params.video_in_enabled and not self._video_task and self._task_manager:
self._video_task = self._task_manager.create_task(
self._callback_task_handler(self._video_queue),
f"{self}::video_callback_task",
)
if self._params.video_out_enabled and not self._camera:
self._camera = Daily.create_camera_device(
self._camera_name(),
@@ -440,22 +441,10 @@ class DailyTransportClient(EventHandler):
color_format=self._params.video_out_color_format,
)
if self._params.audio_out_enabled and not self._mic:
self._mic = Daily.create_microphone_device(
self._mic_name(),
sample_rate=self._out_sample_rate,
channels=self._params.audio_out_channels,
non_blocking=True,
)
if self._params.audio_in_enabled and not self._speaker:
self._speaker = Daily.create_speaker_device(
self._speaker_name(),
sample_rate=self._in_sample_rate,
channels=self._params.audio_in_channels,
non_blocking=True,
)
Daily.select_speaker_device(self._speaker_name())
if self._params.audio_out_enabled and not self._microphone_track:
audio_source = CustomAudioSource(self._out_sample_rate, self._params.audio_out_channels)
audio_track = CustomAudioTrack(audio_source)
self._microphone_track = DailyAudioTrack(source=audio_source, track=audio_track)
async def join(self):
# Transport already joined or joining, ignore.
@@ -540,12 +529,11 @@ class DailyTransportClient(EventHandler):
"microphone": {
"isEnabled": microphone_enabled,
"settings": {
"deviceId": self._mic_name(),
"customConstraints": {
"autoGainControl": {"exact": False},
"echoCancellation": {"exact": False},
"noiseSuppression": {"exact": False},
},
"customTrack": {
"id": self._microphone_track.track.id
if self._microphone_track
else "no-microphone-track"
}
},
},
},
@@ -592,7 +580,7 @@ class DailyTransportClient(EventHandler):
await self._stop_transcription()
# Remove any custom tracks, if any.
for track_name, _ in self._audio_sources.items():
for track_name, _ in self._custom_audio_tracks.items():
await self.remove_custom_audio_track(track_name)
try:
@@ -694,6 +682,8 @@ class DailyTransportClient(EventHandler):
participant_id: str,
callback: Callable,
audio_source: str = "microphone",
sample_rate: int = 16000,
callback_interval_ms: int = 20,
):
# Only enable the desired audio source subscription on this participant.
if audio_source in ("microphone", "screenAudio"):
@@ -705,14 +695,14 @@ class DailyTransportClient(EventHandler):
self._audio_renderers.setdefault(participant_id, {})[audio_source] = callback
logger.info(
f"Starting to capture audio from participant {participant_id} to {audio_source}"
)
logger.info(f"Starting to capture [{audio_source}] audio from participant {participant_id}")
self._client.set_audio_renderer(
participant_id,
self._audio_data_received,
audio_source=audio_source,
sample_rate=sample_rate,
callback_interval_ms=callback_interval_ms,
)
async def capture_participant_video(
@@ -740,19 +730,24 @@ class DailyTransportClient(EventHandler):
color_format=color_format,
)
async def add_custom_audio_track(self, track_name: str) -> CustomAudioSource:
async def add_custom_audio_track(self, track_name: str) -> DailyAudioTrack:
future = self._get_event_loop().create_future()
audio_source = CustomAudioSource(self._out_sample_rate, 1)
audio_track = CustomAudioTrack(audio_source)
self._client.add_custom_audio_track(
track_name=track_name,
audio_source=audio_source,
audio_track=audio_track,
completion=completion_callback(future),
)
await future
return audio_source
track = DailyAudioTrack(source=audio_source, track=audio_track)
return track
async def remove_custom_audio_track(self, track_name: str):
future = self._get_event_loop().create_future()
@@ -799,57 +794,57 @@ class DailyTransportClient(EventHandler):
#
def on_active_speaker_changed(self, participant):
self._call_async_callback(self._callbacks.on_active_speaker_changed, participant)
self._call_event_callback(self._callbacks.on_active_speaker_changed, participant)
def on_app_message(self, message: Any, sender: str):
self._call_async_callback(self._callbacks.on_app_message, message, sender)
self._call_event_callback(self._callbacks.on_app_message, message, sender)
def on_call_state_updated(self, state: str):
self._call_async_callback(self._callbacks.on_call_state_updated, state)
self._call_event_callback(self._callbacks.on_call_state_updated, state)
def on_dialin_connected(self, data: Any):
self._call_async_callback(self._callbacks.on_dialin_connected, data)
self._call_event_callback(self._callbacks.on_dialin_connected, data)
def on_dialin_ready(self, sip_endpoint: str):
self._call_async_callback(self._callbacks.on_dialin_ready, sip_endpoint)
self._call_event_callback(self._callbacks.on_dialin_ready, sip_endpoint)
def on_dialin_stopped(self, data: Any):
self._call_async_callback(self._callbacks.on_dialin_stopped, data)
self._call_event_callback(self._callbacks.on_dialin_stopped, data)
def on_dialin_error(self, data: Any):
self._call_async_callback(self._callbacks.on_dialin_error, data)
self._call_event_callback(self._callbacks.on_dialin_error, data)
def on_dialin_warning(self, data: Any):
self._call_async_callback(self._callbacks.on_dialin_warning, data)
self._call_event_callback(self._callbacks.on_dialin_warning, data)
def on_dialout_answered(self, data: Any):
self._call_async_callback(self._callbacks.on_dialout_answered, data)
self._call_event_callback(self._callbacks.on_dialout_answered, data)
def on_dialout_connected(self, data: Any):
self._call_async_callback(self._callbacks.on_dialout_connected, data)
self._call_event_callback(self._callbacks.on_dialout_connected, data)
def on_dialout_stopped(self, data: Any):
self._call_async_callback(self._callbacks.on_dialout_stopped, data)
self._call_event_callback(self._callbacks.on_dialout_stopped, data)
def on_dialout_error(self, data: Any):
self._call_async_callback(self._callbacks.on_dialout_error, data)
self._call_event_callback(self._callbacks.on_dialout_error, data)
def on_dialout_warning(self, data: Any):
self._call_async_callback(self._callbacks.on_dialout_warning, data)
self._call_event_callback(self._callbacks.on_dialout_warning, data)
def on_participant_joined(self, participant):
self._call_async_callback(self._callbacks.on_participant_joined, participant)
self._call_event_callback(self._callbacks.on_participant_joined, participant)
def on_participant_left(self, participant, reason):
self._call_async_callback(self._callbacks.on_participant_left, participant, reason)
self._call_event_callback(self._callbacks.on_participant_left, participant, reason)
def on_participant_updated(self, participant):
self._call_async_callback(self._callbacks.on_participant_updated, participant)
self._call_event_callback(self._callbacks.on_participant_updated, participant)
def on_transcription_started(self, status):
logger.debug(f"Transcription started: {status}")
self._transcription_status = status
self._call_async_callback(self.update_transcription, self._transcription_ids)
self._call_event_callback(self.update_transcription, self._transcription_ids)
def on_transcription_stopped(self, stopped_by, stopped_by_error):
logger.debug("Transcription stopped")
@@ -858,19 +853,19 @@ class DailyTransportClient(EventHandler):
logger.error(f"Transcription error: {message}")
def on_transcription_message(self, message):
self._call_async_callback(self._callbacks.on_transcription_message, message)
self._call_event_callback(self._callbacks.on_transcription_message, message)
def on_recording_started(self, status):
logger.debug(f"Recording started: {status}")
self._call_async_callback(self._callbacks.on_recording_started, status)
self._call_event_callback(self._callbacks.on_recording_started, status)
def on_recording_stopped(self, stream_id):
logger.debug(f"Recording stopped: {stream_id}")
self._call_async_callback(self._callbacks.on_recording_stopped, stream_id)
self._call_event_callback(self._callbacks.on_recording_stopped, stream_id)
def on_recording_error(self, stream_id, message):
logger.error(f"Recording error for {stream_id}: {message}")
self._call_async_callback(self._callbacks.on_recording_error, stream_id, message)
self._call_event_callback(self._callbacks.on_recording_error, stream_id, message)
#
# Daily (CallClient callbacks)
@@ -878,25 +873,38 @@ class DailyTransportClient(EventHandler):
def _audio_data_received(self, participant_id: str, audio_data: AudioData, audio_source: str):
callback = self._audio_renderers[participant_id][audio_source]
self._call_async_callback(callback, participant_id, audio_data, audio_source)
self._call_audio_callback(callback, participant_id, audio_data, audio_source)
def _video_frame_received(
self, participant_id: str, video_frame: VideoFrame, video_source: str
):
callback = self._video_renderers[participant_id][video_source]
self._call_async_callback(callback, participant_id, video_frame, video_source)
self._call_video_callback(callback, participant_id, video_frame, video_source)
def _call_async_callback(self, callback, *args):
#
# Queue callbacks handling
#
def _call_audio_callback(self, callback, *args):
self._call_async_callback(self._audio_queue, callback, *args)
def _call_video_callback(self, callback, *args):
self._call_async_callback(self._video_queue, callback, *args)
def _call_event_callback(self, callback, *args):
self._call_async_callback(self._event_queue, callback, *args)
def _call_async_callback(self, queue: asyncio.Queue, callback, *args):
future = asyncio.run_coroutine_threadsafe(
self._callback_queue.put((callback, *args)), self._get_event_loop()
queue.put((callback, *args)), self._get_event_loop()
)
future.result()
async def _callback_task_handler(self):
async def _callback_task_handler(self, queue: asyncio.Queue):
while True:
# Wait to process any callback until we are joined.
await self._joined_event.wait()
(callback, *args) = await self._callback_queue.get()
(callback, *args) = await queue.get()
await callback(*args)
def _get_event_loop(self) -> asyncio.AbstractEventLoop:
@@ -936,11 +944,12 @@ class DailyInputTransport(BaseInputTransport):
# Whether we have seen a StartFrame already.
self._initialized = False
# Task that gets audio data from a device or the network and queues it
# internally to be processed.
self._audio_in_task = None
# Whether we have started audio streaming.
self._streaming_started = False
self._resampler = create_default_resampler()
# Store the list of participants we should stream. This is necessary in
# case we don't start streaming right away.
self._capture_participant_audio = []
self._vad_analyzer: Optional[VADAnalyzer] = params.vad_analyzer
@@ -948,12 +957,17 @@ class DailyInputTransport(BaseInputTransport):
def vad_analyzer(self) -> Optional[VADAnalyzer]:
return self._vad_analyzer
def start_audio_in_streaming(self):
# Create audio task. It reads audio frames from Daily and push them
# internally for VAD processing.
if not self._audio_in_task and self._params.audio_in_enabled:
logger.debug(f"Start receiving audio")
self._audio_in_task = self.create_task(self._audio_in_task_handler())
async def start_audio_in_streaming(self):
if not self._params.audio_in_enabled:
return
logger.debug(f"Start receiving audio")
for participant_id, audio_source, sample_rate in self._capture_participant_audio:
await self._client.capture_participant_audio(
participant_id, self._on_participant_audio_data, audio_source, sample_rate
)
self._streaming_started = True
async def setup(self, setup: FrameProcessorSetup):
await super().setup(setup)
@@ -983,27 +997,19 @@ class DailyInputTransport(BaseInputTransport):
await self.set_transport_ready(frame)
if self._params.audio_in_stream_on_start:
self.start_audio_in_streaming()
await self.start_audio_in_streaming()
async def stop(self, frame: EndFrame):
# Parent stop.
await super().stop(frame)
# Leave the room.
await self._client.leave()
# Stop audio thread.
if self._audio_in_task and self._params.audio_in_enabled:
await self.cancel_task(self._audio_in_task)
self._audio_in_task = None
async def cancel(self, frame: CancelFrame):
# Parent stop.
await super().cancel(frame)
# Leave the room.
await self._client.leave()
# Stop audio thread.
if self._audio_in_task and self._params.audio_in_enabled:
await self.cancel_task(self._audio_in_task)
self._audio_in_task = None
#
# FrameProcessor
@@ -1034,32 +1040,26 @@ class DailyInputTransport(BaseInputTransport):
self,
participant_id: str,
audio_source: str = "microphone",
sample_rate: int = 16000,
):
await self._client.capture_participant_audio(
participant_id, self._on_participant_audio_data, audio_source
)
if self._streaming_started:
await self._client.capture_participant_audio(
participant_id, self._on_participant_audio_data, audio_source, sample_rate
)
else:
self._capture_participant_audio.append((participant_id, audio_source, sample_rate))
async def _on_participant_audio_data(
self, participant_id: str, audio: AudioData, audio_source: str
):
resampled = await self._resampler.resample(
audio.audio_frames, audio.sample_rate, self._client.out_sample_rate
)
frame = UserAudioRawFrame(
user_id=participant_id,
audio=resampled,
sample_rate=self._client.out_sample_rate,
audio=audio.audio_frames,
sample_rate=audio.sample_rate,
num_channels=audio.num_channels,
)
frame.transport_source = audio_source
await self.push_frame(frame)
async def _audio_in_task_handler(self):
while True:
frame = await self._client.read_next_audio_frame()
if frame:
await self.push_audio_frame(frame)
await self.push_audio_frame(frame)
#
# Camera in
@@ -1376,9 +1376,10 @@ class DailyTransport(BaseTransport):
self,
participant_id: str,
audio_source: str = "microphone",
sample_rate: int = 16000,
):
if self._input:
await self._input.capture_participant_audio(participant_id, audio_source)
await self._input.capture_participant_audio(participant_id, audio_source, sample_rate)
async def capture_participant_video(
self,
@@ -1509,6 +1510,11 @@ class DailyTransport(BaseTransport):
id = participant["id"]
logger.info(f"Participant joined {id}")
if self._input and self._params.audio_in_enabled:
await self._input.capture_participant_audio(
id, "microphone", self._client.in_sample_rate
)
if not self._other_participant_has_joined:
self._other_participant_has_joined = True
await self._call_event_handler("on_first_participant_joined", participant)

7
src/pipecat/transports/services/livekit.py
View File

@@ -17,13 +17,13 @@ from pipecat.frames.frames import (
AudioRawFrame,
CancelFrame,
EndFrame,
InputAudioRawFrame,
OutputAudioRawFrame,
StartFrame,
TransportMessageFrame,
TransportMessageUrgentFrame,
UserAudioRawFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
from pipecat.processors.frame_processor import FrameDirection, FrameProcessorSetup
from pipecat.transports.base_input import BaseInputTransport
from pipecat.transports.base_output import BaseOutputTransport
from pipecat.transports.base_transport import BaseTransport, TransportParams
@@ -411,7 +411,8 @@ class LiveKitInputTransport(BaseInputTransport):
pipecat_audio_frame = await self._convert_livekit_audio_to_pipecat(
audio_frame_event
)
input_audio_frame = InputAudioRawFrame(
input_audio_frame = UserAudioRawFrame(
user_id=participant_id,
audio=pipecat_audio_frame.audio,
sample_rate=pipecat_audio_frame.sample_rate,
num_channels=pipecat_audio_frame.num_channels,

532
src/pipecat/transports/services/tavus.py Normal file
View File

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