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

Sending audio faster than realtime.

Browse Source
This commit is contained in:
filipi87
2026-05-20 12:03:24 -03:00
parent aef6226a1c
commit 996aa461ac
2 changed files with 167 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
scripts/daily/test_tavus_transport.py
View File

@@ -1,6 +1,7 @@
import asyncio
import os
import signal
import time
from daily import (
AudioData,
@@ -15,6 +16,14 @@ from loguru import logger
load_dotenv(override=True)
# Pipecat sends audio at this true content rate but declares it as
# DECLARED_SAMPLE_RATE to write_frames(), which makes delivery faster than
# real-time. We receive at the declared rate (no resampling) and play back at
# the true rate so the avatar consumes audio at normal speed.
TRUE_SAMPLE_RATE = 24000
DECLARED_SAMPLE_RATE = 48000
SPEEDUP = DECLARED_SAMPLE_RATE // TRUE_SAMPLE_RATE
def completion_callback(future):
def _callback(*args):
@@ -37,19 +46,21 @@ class DailyProxyApp(EventHandler):
def __new__(cls, *args, **kwargs):
return super().__new__(cls)
def __init__(self, sample_rate: int):
def __init__(self):
super().__init__()
self._sample_rate = sample_rate
self._loop = asyncio.new_event_loop()
self._audio_queue: asyncio.Queue = asyncio.Queue()
# Raw PCM buffer — filled at DECLARED_SAMPLE_RATE speed, drained at TRUE_SAMPLE_RATE speed.
self._buffer = bytearray()
self._audio_task: asyncio.Task | None = None
self._receive_start_time: float | None = None
self._client: CallClient = CallClient(event_handler=self)
self._client.update_subscription_profiles(
{"base": {"camera": "unsubscribed", "microphone": "subscribed"}}
)
self._audio_source = CustomAudioSource(self._sample_rate, 1)
# Playback source declared at TRUE_SAMPLE_RATE — consumes audio at real-time speed.
self._audio_source = CustomAudioSource(TRUE_SAMPLE_RATE, 1)
self._audio_track = CustomAudioTrack(self._audio_source)
def on_joined(self, data, error):
@@ -113,7 +124,6 @@ class DailyProxyApp(EventHandler):
if self._audio_task:
self._audio_task.cancel()
try:
# Waits for it to finish
await self._audio_task
except asyncio.CancelledError:
pass
@@ -121,36 +131,60 @@ class DailyProxyApp(EventHandler):
async def capture_participant_audio(self, participant_id: str):
logger.info(f"Capturing participant audio: {participant_id}")
# Receiving from this custom track
# audio_source: str = "microphone"
audio_source: str = "stream"
media = {"media": {"customAudio": {audio_source: "subscribed"}}}
await self.update_subscriptions(participant_settings={participant_id: media})
# Must match the declared rate Pipecat used so WebRTC skips resampling —
# every original byte arrives intact.
self._client.set_audio_renderer(
participant_id,
self._audio_data_received,
audio_source=audio_source,
sample_rate=self._sample_rate,
sample_rate=DECLARED_SAMPLE_RATE,
callback_interval_ms=20,
)
logger.info(
f"Receiving at declared_rate={DECLARED_SAMPLE_RATE} Hz "
f"(true content: {TRUE_SAMPLE_RATE} Hz, ~{SPEEDUP}x faster than real-time)"
)
async def send_audio(self, audio: AudioData):
future = asyncio.get_running_loop().create_future()
self._audio_source.write_frames(audio.audio_frames, completion=completion_callback(future))
await future
async def _buffer_audio(self, audio_data: AudioData):
"""Append received bytes to the buffer and log the fill rate."""
new_bytes = audio_data.audio_frames
if self._receive_start_time is None:
self._receive_start_time = time.monotonic()
async def queue_audio(self, audio: AudioData):
await self._audio_queue.put(audio)
self._buffer.extend(new_bytes)
def _audio_data_received(self, participant_id: str, audio_data: AudioData, audio_source: str):
# logger.info(f"Received audio data for {participant_id}, audio_source: {audio_source}")
asyncio.run_coroutine_threadsafe(self.queue_audio(audio_data), self._loop)
asyncio.run_coroutine_threadsafe(self._buffer_audio(audio_data), self._loop)
async def _audio_task_handler(self):
"""Drain the buffer at TRUE_SAMPLE_RATE speed (real-time playback)."""
chunk_frames = int(TRUE_SAMPLE_RATE * 20 / 1000) # 20 ms chunks
chunk_bytes = chunk_frames * 2 # 16-bit mono
last_log_time = self._loop.time()
while True:
audio = await self._audio_queue.get()
await self.send_audio(audio)
if len(self._buffer) >= chunk_bytes:
chunk = bytes(self._buffer[:chunk_bytes])
del self._buffer[:chunk_bytes]
future = asyncio.get_running_loop().create_future()
self._audio_source.write_frames(chunk, completion=completion_callback(future))
await future
else:
await asyncio.sleep(0.001)
now = self._loop.time()
if now - last_log_time >= 1.0:
buffer_seconds = len(self._buffer) / (TRUE_SAMPLE_RATE * 2)
if buffer_seconds > 0:
logger.info(
f"Buffer status: {len(self._buffer)}B ({buffer_seconds:.3f}s buffered)"
)
last_log_time = now
#
# Daily (EventHandler)
@@ -160,7 +194,7 @@ class DailyProxyApp(EventHandler):
participant_name = participant["info"]["userName"]
logger.info(f"Participant {participant_name} joined")
if participant_name != "Pipecat":
# We are only subscribing for audios from Pipecat.
# We are only subscribing for audio from Pipecat.
return
asyncio.run_coroutine_threadsafe(
self.capture_participant_audio(participant_id=participant["id"]), self._loop
@@ -173,7 +207,7 @@ class DailyProxyApp(EventHandler):
def main():
Daily.init()
room_url = os.environ["TAVUS_SAMPLE_ROOM_URL"]
app = DailyProxyApp(sample_rate=24000)
app = DailyProxyApp()
app.run(room_url)

176
src/pipecat/transports/tavus/transport.py
View File

@@ -40,10 +40,16 @@ from pipecat.transports.base_output import BaseOutputTransport
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import (
DailyCallbacks,
DailyCustomAudioTrackParams,
DailyParams,
DailyTransportClient,
)
# Opus codec maximum. When the Tavus server supports fast audio delivery it
# returns this as stream_declared_sample_rate so that write_frames() blocks for
# n/48000 s instead of n/true_rate s, delivering audio faster than real-time.
_STREAM_DECLARED_SAMPLE_RATE = 48000
class TavusApi:
"""Helper class for interacting with the Tavus API (v2).
@@ -69,20 +75,30 @@ class TavusApi:
# Only for development
self._dev_room_url = os.getenv("TAVUS_SAMPLE_ROOM_URL")
async def create_conversation(self, replica_id: str, persona_id: str) -> dict:
async def create_conversation(
self, replica_id: str, persona_id: str, sample_rate: int
) -> dict:
"""Create a new conversation with the specified replica and persona.
Args:
replica_id: ID of the replica to use in the conversation.
persona_id: ID of the persona to use in the conversation.
sample_rate: True audio sample rate of the pipeline's output. Sent
to Tavus so the server can negotiate fast audio delivery. When
the server supports it, the response includes
``stream_declared_sample_rate`` — the rate Pipecat should
declare to the ``CustomAudioSource`` for faster-than-realtime
delivery.
Returns:
Dictionary containing conversation_id and conversation_url.
Dictionary containing conversation_id, conversation_url, and
optionally stream_declared_sample_rate.
"""
if self._dev_room_url:
return {
"conversation_id": self.MOCK_CONVERSATION_ID,
"conversation_url": self._dev_room_url,
"stream_declared_sample_rate": _STREAM_DECLARED_SAMPLE_RATE,
}
logger.debug(f"Creating Tavus conversation: replica={replica_id}, persona={persona_id}")
@@ -90,6 +106,8 @@ class TavusApi:
payload = {
"replica_id": replica_id,
"persona_id": persona_id,
# TODO: start to send it when Tavus start to support it.
# "sample_rate": sample_rate,
}
async with self._session.post(url, headers=self._headers, json=payload) as r:
r.raise_for_status()
@@ -202,76 +220,64 @@ class TavusTransportClient:
self._client: DailyTransportClient | None = None
self._callbacks = callbacks
self._params = params
self._setup: FrameProcessorSetup | None = None
self._initialized = False
async def _initialize(self) -> str:
"""Initialize the conversation and return the room URL."""
response = await self._api.create_conversation(self._replica_id, self._persona_id)
self._conversation_id = response["conversation_id"]
return response["conversation_url"]
def _build_daily_callbacks(self) -> DailyCallbacks:
"""Build the DailyCallbacks object."""
return 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_before_leave=partial(self._on_handle_callback, "on_before_leave"),
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_dtmf_event=partial(self._on_handle_callback, "on_dtmf_event"),
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"),
on_transcription_stopped=partial(
self._on_handle_callback, "on_transcription_stopped"
),
on_transcription_error=partial(self._on_handle_callback, "on_transcription_error"),
)
async def setup(self, setup: FrameProcessorSetup):
"""Setup the client and initialize the conversation.
"""Save setup context for later use in start().
Args:
setup: The frame processor setup configuration.
"""
if self._conversation_id is not None:
logger.debug(f"Conversation ID already defined: {self._conversation_id}")
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_before_leave=partial(self._on_handle_callback, "on_before_leave"),
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_dtmf_event=partial(self._on_handle_callback, "on_dtmf_event"),
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"),
on_transcription_stopped=partial(
self._on_handle_callback, "on_transcription_stopped"
),
on_transcription_error=partial(self._on_handle_callback, "on_transcription_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)
self._conversation_id = None
self._setup = setup
async def cleanup(self):
"""Cleanup client resources."""
try:
await self._client.cleanup()
except Exception as e:
logger.error(f"Exception during cleanup: {e}")
if self._client:
try:
await self._client.cleanup()
except Exception as e:
logger.error(f"Exception during cleanup: {e}")
async def _on_joined(self, data):
"""Handle joined event."""
@@ -295,12 +301,56 @@ class TavusTransportClient:
return await self._api.get_persona_name(self._persona_id)
async def start(self, frame: StartFrame):
"""Start the client and join the room.
"""Create the conversation, build the Daily client, and join the room.
Args:
frame: The start frame containing initialization parameters.
"""
if self._initialized:
return
self._initialized = True
logger.debug("TavusTransportClient start invoked!")
try:
sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate
response = await self._api.create_conversation(
self._replica_id, self._persona_id, sample_rate
)
self._conversation_id = response["conversation_id"]
room_url = response["conversation_url"]
params = self._params
stream_declared_sample_rate = response.get("stream_declared_sample_rate")
if stream_declared_sample_rate:
# Tavus supports fast audio delivery: we write true-rate PCM bytes into a
# CustomAudioSource declared at stream_declared_sample_rate (e.g. 48 kHz).
# write_frames() blocks for n/declared_rate seconds instead of n/true_rate
# seconds, so audio is delivered faster than real-time. The receiver must
# also request the same declared rate so WebRTC skips resampling and every
# original byte arrives intact.
# We always override sample_rate here even if the user already provided
# "stream" params, because the declared rate must match what the server
# negotiated — other fields (channels, send_settings) are preserved.
logger.debug(
f"Tavus fast audio: true_rate={sample_rate} declared_rate={stream_declared_sample_rate}"
)
existing = dict(params.custom_audio_track_params or {})
existing["stream"] = (
existing.get("stream") or DailyCustomAudioTrackParams()
).model_copy(update={"sample_rate": stream_declared_sample_rate})
params = params.model_copy(update={"custom_audio_track_params": existing})
self._client = DailyTransportClient(
room_url, None, "Pipecat", params, self._build_daily_callbacks(), self._bot_name
)
await self._client.setup(self._setup)
except Exception as e:
logger.error(f"Failed to start TavusTransportClient: {e}")
await self._api.end_conversation(self._conversation_id)
self._conversation_id = None
self._initialized = False
return
await self._client.start(frame)
await self._client.join()