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

Merge branch 'main' into m-ods/assemblyai-universal-streaming

Browse Source
This commit is contained in:
Martin Schweiger
2025-05-30 10:11:02 +08:00
parent db838634e7 366add2536
commit f23572b318
34 changed files with 2141 additions and 298 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()

252
src/pipecat/serializers/plivo.py Normal file
View File

@@ -0,0 +1,252 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import base64
import json
from typing import Optional
from loguru import logger
from pydantic import BaseModel
from pipecat.audio.utils import create_default_resampler, pcm_to_ulaw, ulaw_to_pcm
from pipecat.frames.frames import (
AudioRawFrame,
CancelFrame,
EndFrame,
Frame,
InputAudioRawFrame,
InputDTMFFrame,
KeypadEntry,
StartFrame,
StartInterruptionFrame,
TransportMessageFrame,
TransportMessageUrgentFrame,
)
from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType
class PlivoFrameSerializer(FrameSerializer):
"""Serializer for Plivo Audio Streaming WebSocket protocol.
This serializer handles converting between Pipecat frames and Plivo's WebSocket
audio streaming protocol. It supports audio conversion, DTMF events, and automatic
call termination.
When auto_hang_up is enabled (default), the serializer will automatically terminate
the Plivo call when an EndFrame or CancelFrame is processed, but requires Plivo
credentials to be provided.
Attributes:
_stream_id: The Plivo Stream ID.
_call_id: The associated Plivo Call ID.
_auth_id: Plivo auth ID for API access.
_auth_token: Plivo authentication token for API access.
_params: Configuration parameters.
_plivo_sample_rate: Sample rate used by Plivo (typically 8kHz).
_sample_rate: Input sample rate for the pipeline.
_resampler: Audio resampler for format conversion.
"""
class InputParams(BaseModel):
"""Configuration parameters for PlivoFrameSerializer.
Attributes:
plivo_sample_rate: Sample rate used by Plivo, defaults to 8000 Hz.
sample_rate: Optional override for pipeline input sample rate.
auto_hang_up: Whether to automatically terminate call on EndFrame.
"""
plivo_sample_rate: int = 8000
sample_rate: Optional[int] = None
auto_hang_up: bool = True
def __init__(
self,
stream_id: str,
call_id: Optional[str] = None,
auth_id: Optional[str] = None,
auth_token: Optional[str] = None,
params: Optional[InputParams] = None,
):
"""Initialize the PlivoFrameSerializer.
Args:
stream_id: The Plivo Stream ID.
call_id: The associated Plivo Call ID (optional, but required for auto hang-up).
auth_id: Plivo auth ID (required for auto hang-up).
auth_token: Plivo auth token (required for auto hang-up).
params: Configuration parameters.
"""
self._stream_id = stream_id
self._call_id = call_id
self._auth_id = auth_id
self._auth_token = auth_token
self._params = params or PlivoFrameSerializer.InputParams()
self._plivo_sample_rate = self._params.plivo_sample_rate
self._sample_rate = 0 # Pipeline input rate
self._resampler = create_default_resampler()
self._hangup_attempted = False
@property
def type(self) -> FrameSerializerType:
"""Gets the serializer type.
Returns:
The serializer type, either TEXT or BINARY.
"""
return FrameSerializerType.TEXT
async def setup(self, frame: StartFrame):
"""Sets up the serializer with pipeline configuration.
Args:
frame: The StartFrame containing pipeline configuration.
"""
self._sample_rate = self._params.sample_rate or frame.audio_in_sample_rate
async def serialize(self, frame: Frame) -> str | bytes | None:
"""Serializes a Pipecat frame to Plivo WebSocket format.
Handles conversion of various frame types to Plivo WebSocket messages.
For EndFrames, initiates call termination if auto_hang_up is enabled.
Args:
frame: The Pipecat frame to serialize.
Returns:
Serialized data as string or bytes, or None if the frame isn't handled.
"""
if (
self._params.auto_hang_up
and not self._hangup_attempted
and isinstance(frame, (EndFrame, CancelFrame))
):
self._hangup_attempted = True
await self._hang_up_call()
return None
elif isinstance(frame, StartInterruptionFrame):
answer = {"event": "clearAudio", "streamId": self._stream_id}
return json.dumps(answer)
elif isinstance(frame, AudioRawFrame):
data = frame.audio
# Output: Convert PCM at frame's rate to 8kHz μ-law for Plivo
serialized_data = await pcm_to_ulaw(
data, frame.sample_rate, self._plivo_sample_rate, self._resampler
)
payload = base64.b64encode(serialized_data).decode("utf-8")
answer = {
"event": "playAudio",
"media": {
"contentType": "audio/x-mulaw",
"sampleRate": self._plivo_sample_rate,
"payload": payload,
},
"streamId": self._stream_id,
}
return json.dumps(answer)
elif isinstance(frame, (TransportMessageFrame, TransportMessageUrgentFrame)):
return json.dumps(frame.message)
# Return None for unhandled frames
return None
async def _hang_up_call(self):
"""Hang up the Plivo call using Plivo's REST API."""
try:
import aiohttp
auth_id = self._auth_id
auth_token = self._auth_token
call_id = self._call_id
if not call_id or not auth_id or not auth_token:
missing = []
if not call_id:
missing.append("call_id")
if not auth_id:
missing.append("auth_id")
if not auth_token:
missing.append("auth_token")
logger.warning(
f"Cannot hang up Plivo call: missing required parameters: {', '.join(missing)}"
)
return
# Plivo API endpoint for hanging up calls
endpoint = f"https://api.plivo.com/v1/Account/{auth_id}/Call/{call_id}/"
# Create basic auth from auth_id and auth_token
auth = aiohttp.BasicAuth(auth_id, auth_token)
# Make the DELETE request to hang up the call
async with aiohttp.ClientSession() as session:
async with session.delete(endpoint, auth=auth) as response:
if response.status == 204: # Plivo returns 204 for successful hangup
logger.debug(f"Successfully terminated Plivo call {call_id}")
elif response.status == 404: # Call already ended
logger.debug(f"Plivo call {call_id} already terminated")
else:
# Get the error details for better debugging
error_text = await response.text()
logger.error(
f"Failed to terminate Plivo call {call_id}: "
f"Status {response.status}, Response: {error_text}"
)
except Exception as e:
logger.exception(f"Failed to hang up Plivo call: {e}")
async def deserialize(self, data: str | bytes) -> Frame | None:
"""Deserializes Plivo WebSocket data to Pipecat frames.
Handles conversion of Plivo media events to appropriate Pipecat frames.
Args:
data: The raw WebSocket data from Plivo.
Returns:
A Pipecat frame corresponding to the Plivo event, or None if unhandled.
"""
try:
message = json.loads(data)
except json.JSONDecodeError:
logger.warning(f"Failed to parse JSON message: {data}")
return None
if message.get("event") == "media":
media = message.get("media", {})
payload_base64 = media.get("payload")
if not payload_base64:
return None
payload = base64.b64decode(payload_base64)
# Input: Convert Plivo's 8kHz μ-law to PCM at pipeline input rate
deserialized_data = await ulaw_to_pcm(
payload, self._plivo_sample_rate, self._sample_rate, self._resampler
)
audio_frame = InputAudioRawFrame(
audio=deserialized_data, num_channels=1, sample_rate=self._sample_rate
)
return audio_frame
elif message.get("event") == "dtmf":
dtmf_data = message.get("dtmf", {})
digit = dtmf_data.get("digit")
if digit:
try:
return InputDTMFFrame(KeypadEntry(digit))
except ValueError:
# Handle case where string doesn't match any enum value
logger.warning(f"Invalid DTMF digit received: {digit}")
return None
else:
return None

161
src/pipecat/services/google/tts.py
View File

@@ -202,7 +202,7 @@ def language_to_google_tts_language(language: Language) -> Optional[str]:
return language_map.get(language)
class GoogleTTSService(TTSService):
class GoogleHttpTTSService(TTSService):
class InputParams(BaseModel):
pitch: Optional[str] = None
rate: Optional[str] = None
@@ -217,14 +217,14 @@ class GoogleTTSService(TTSService):
*,
credentials: Optional[str] = None,
credentials_path: Optional[str] = None,
voice_id: str = "en-US-Neural2-A",
voice_id: str = "en-US-Chirp3-HD-Charon",
sample_rate: Optional[int] = None,
params: Optional[InputParams] = None,
**kwargs,
):
super().__init__(sample_rate=sample_rate, **kwargs)
params = params or GoogleTTSService.InputParams()
params = params or GoogleHttpTTSService.InputParams()
self._settings = {
"pitch": params.pitch,
@@ -371,7 +371,160 @@ class GoogleTTSService(TTSService):
await self.stop_ttfb_metrics()
frame = TTSAudioRawFrame(chunk, self.sample_rate, 1)
yield frame
await asyncio.sleep(0) # Allow other tasks to run
yield TTSStoppedFrame()
except Exception as e:
logger.exception(f"{self} error generating TTS: {e}")
error_message = f"TTS generation error: {str(e)}"
yield ErrorFrame(error=error_message)
class GoogleTTSService(TTSService):
"""Text-to-Speech service using Google Cloud Text-to-Speech API.
Converts text to speech using Google's TTS models with streaming synthesis
for low latency. Supports multiple languages and voices.
Args:
credentials: JSON string containing Google Cloud service account credentials.
credentials_path: Path to Google Cloud service account JSON file.
voice_id: Google TTS voice identifier (e.g., "en-US-Chirp3-HD-Charon").
sample_rate: Audio sample rate in Hz.
params: Language only.
Notes:
Requires Google Cloud credentials via service account JSON, file path, or
default application credentials (GOOGLE_APPLICATION_CREDENTIALS env var).
Only Chirp 3 HD and Journey voices are supported. Use GoogleHttpTTSService for other voices.
Example:
```python
tts = GoogleTTSService(
credentials_path="/path/to/service-account.json",
voice_id="en-US-Chirp3-HD-Charon",
params=GoogleTTSService.InputParams(
language=Language.EN_US,
)
)
```
"""
class InputParams(BaseModel):
language: Optional[Language] = Language.EN
def __init__(
self,
*,
credentials: Optional[str] = None,
credentials_path: Optional[str] = None,
voice_id: str = "en-US-Chirp3-HD-Charon",
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
**kwargs,
):
super().__init__(sample_rate=sample_rate, **kwargs)
params = params or GoogleTTSService.InputParams()
self._settings = {
"language": self.language_to_service_language(params.language)
if params.language
else "en-US",
}
self.set_voice(voice_id)
self._client: texttospeech_v1.TextToSpeechAsyncClient = self._create_client(
credentials, credentials_path
)
def _create_client(
self, credentials: Optional[str], credentials_path: Optional[str]
) -> texttospeech_v1.TextToSpeechAsyncClient:
creds: Optional[service_account.Credentials] = None
# Create a Google Cloud service account for the Cloud Text-to-Speech API
# Using either the provided credentials JSON string or the path to a service account JSON
# file, create a Google Cloud service account and use it to authenticate with the API.
if credentials:
# Use provided credentials JSON string
json_account_info = json.loads(credentials)
creds = service_account.Credentials.from_service_account_info(json_account_info)
elif credentials_path:
# Use service account JSON file if provided
creds = service_account.Credentials.from_service_account_file(credentials_path)
else:
try:
creds, project_id = default(
scopes=["https://www.googleapis.com/auth/cloud-platform"]
)
except GoogleAuthError:
pass
if not creds:
raise ValueError("No valid credentials provided.")
return texttospeech_v1.TextToSpeechAsyncClient(credentials=creds)
def can_generate_metrics(self) -> bool:
return True
def language_to_service_language(self, language: Language) -> Optional[str]:
return language_to_google_tts_language(language)
@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()
voice = texttospeech_v1.VoiceSelectionParams(
language_code=self._settings["language"], name=self._voice_id
)
streaming_config = texttospeech_v1.StreamingSynthesizeConfig(
voice=voice,
streaming_audio_config=texttospeech_v1.StreamingAudioConfig(
audio_encoding=texttospeech_v1.AudioEncoding.PCM,
sample_rate_hertz=self.sample_rate,
),
)
config_request = texttospeech_v1.StreamingSynthesizeRequest(
streaming_config=streaming_config
)
async def request_generator():
yield config_request
yield texttospeech_v1.StreamingSynthesizeRequest(
input=texttospeech_v1.StreamingSynthesisInput(text=text)
)
streaming_responses = await self._client.streaming_synthesize(request_generator())
await self.start_tts_usage_metrics(text)
yield TTSStartedFrame()
audio_buffer = b""
CHUNK_SIZE = 1024
first_chunk_for_ttfb = False
async for response in streaming_responses:
chunk = response.audio_content
if not chunk:
continue
if not first_chunk_for_ttfb:
await self.stop_ttfb_metrics()
first_chunk_for_ttfb = True
audio_buffer += chunk
while len(audio_buffer) >= CHUNK_SIZE:
piece = audio_buffer[:CHUNK_SIZE]
audio_buffer = audio_buffer[CHUNK_SIZE:]
yield TTSAudioRawFrame(piece, self.sample_rate, 1)
if audio_buffer:
yield TTSAudioRawFrame(audio_buffer, self.sample_rate, 1)
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)