diff --git a/CHANGELOG.md b/CHANGELOG.md index 8208329a0..099783d7d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Added `GoogleHttpTTSService` which uses Google's HTTP TTS API. +- Added `TavusTransport`, a new transport implementation compatible with any + Pipecat pipeline. When using the `TavusTransport`the Pipecat bot will + connect in the same room as the Tavus Avatar and the user. + +- Added `UserBotLatencyLogObserver`. This is an observer that logs the latency + between when the user stops speaking and when the bot starts speaking. This + gives you an initial idea on how quickly the AI services respond. + +- Added `SarvamTTSService`, which implements Sarvam AI's TTS API: + https://docs.sarvam.ai/api-reference-docs/text-to-speech/convert. + - Added `PipelineTask.add_observer()` and `PipelineTask.remove_observer()` to allow mangaging observers at runtime. This is useful for cases where the task is passed around to other code components that might want to observe the @@ -77,6 +88,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Updated `GoogleTTSService` to use Google's streaming TTS API. The default voice also updated to `en-US-Chirp3-HD-Charon`. +- ⚠️Refactored the `TavusVideoService`, so it acts like a proxy, sending audio to + Tavus and receiving both audio and video. This will make `TavusVideoService` usable + with any Pipecat pipeline and with any transport. This is a **breaking change**, + check the `examples/foundational/21a-tavus-layer-small-webrtc.py` to see how to + use it. + +- `DailyTransport` now uses custom microphone audio tracks instead of virtual + microphones. Now, multiple Daily transports can be used in the same process. + +- `DailyTransport` now captures audio from individual participants instead of + the whole room. This allows identifying audio frames per participant. + +- Updated the default model for `AnthropicLLMService` to + `claude-sonnet-4-20250514`. + +- Updated the default model for `GeminiMultimodalLiveLLMService` to + `models/gemini-2.5-flash-preview-native-audio-dialog`. + - `BaseTextFilter` methods `filter()`, `update_settings()`, `handle_interruption()` and `reset_interruption()` are now async. @@ -112,6 +141,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed a `DailyTransport` issue that would cause images needing resize to block + the event loop. + +- Fixed an issue with `ElevenLabsTTSService` where changing the model or voice + while the service is running wasn't working. + - Fixed an issue that would cause multiple instances of the same class to behave incorrectly if any of the given constructor arguments defaulted to a mutable value (e.g. lists, dictionaries, objects). @@ -122,13 +157,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Performance +- `DailyTransport`: process audio, video and events in separate tasks. + - Don't create event handler tasks if no user event handlers have been registered. ### Other -- Added foundation example `07y-minimax-http.py` to show how to use the - `MiniMaxHttpTTSService`. +- Added foundation examples `07y-interruptible-minimax.py` and + `07z-interruptible-sarvam.py`to show how to use the `MiniMaxHttpTTSService` + and `SarvamTTSService`, respectively. - Added an `open-telemetry-tracing` example, showing how to setup tracing. The example also includes Jaeger as an open source OpenTelemetry client to review diff --git a/README.md b/README.md index deea0d58f..ae6c3cc53 100644 --- a/README.md +++ b/README.md @@ -8,6 +8,8 @@ **Pipecat** is an open-source Python framework for building real-time voice and multimodal conversational agents. Orchestrate audio and video, AI services, different transports, and conversation pipelines effortlesslyβ€”so you can focus on what makes your agent unique. +> Want to dive right in? [Install Pipecat](https://docs.pipecat.ai/getting-started/installation) then try the [quickstart](https://docs.pipecat.ai/getting-started/quickstart). + ## πŸš€ What You Can Build - **Voice Assistants** – natural, streaming conversations with AI @@ -49,18 +51,18 @@ You can connect to Pipecat from any platform using our official SDKs: ## 🧩 Available services -| Category | Services | -| ------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| Speech-to-Text | [AssemblyAI](https://docs.pipecat.ai/server/services/stt/assemblyai), [AWS](https://docs.pipecat.ai/server/services/stt/aws), [Azure](https://docs.pipecat.ai/server/services/stt/azure), [Deepgram](https://docs.pipecat.ai/server/services/stt/deepgram), [Fal Wizper](https://docs.pipecat.ai/server/services/stt/fal), [Gladia](https://docs.pipecat.ai/server/services/stt/gladia), [Google](https://docs.pipecat.ai/server/services/stt/google), [Groq (Whisper)](https://docs.pipecat.ai/server/services/stt/groq), [OpenAI (Whisper)](https://docs.pipecat.ai/server/services/stt/openai), [Parakeet (NVIDIA)](https://docs.pipecat.ai/server/services/stt/parakeet), [Ultravox](https://docs.pipecat.ai/server/services/stt/ultravox), [Whisper](https://docs.pipecat.ai/server/services/stt/whisper) | -| LLMs | [Anthropic](https://docs.pipecat.ai/server/services/llm/anthropic), [AWS](https://docs.pipecat.ai/server/services/llm/aws), [Azure](https://docs.pipecat.ai/server/services/llm/azure), [Cerebras](https://docs.pipecat.ai/server/services/llm/cerebras), [DeepSeek](https://docs.pipecat.ai/server/services/llm/deepseek), [Fireworks AI](https://docs.pipecat.ai/server/services/llm/fireworks), [Gemini](https://docs.pipecat.ai/server/services/llm/gemini), [Grok](https://docs.pipecat.ai/server/services/llm/grok), [Groq](https://docs.pipecat.ai/server/services/llm/groq), [NVIDIA NIM](https://docs.pipecat.ai/server/services/llm/nim), [Ollama](https://docs.pipecat.ai/server/services/llm/ollama), [OpenAI](https://docs.pipecat.ai/server/services/llm/openai), [OpenRouter](https://docs.pipecat.ai/server/services/llm/openrouter), [Perplexity](https://docs.pipecat.ai/server/services/llm/perplexity), [Qwen](https://docs.pipecat.ai/server/services/llm/qwen), [Together AI](https://docs.pipecat.ai/server/services/llm/together) | -| Text-to-Speech | [AWS](https://docs.pipecat.ai/server/services/tts/aws), [Azure](https://docs.pipecat.ai/server/services/tts/azure), [Cartesia](https://docs.pipecat.ai/server/services/tts/cartesia), [Deepgram](https://docs.pipecat.ai/server/services/tts/deepgram), [ElevenLabs](https://docs.pipecat.ai/server/services/tts/elevenlabs), [FastPitch (NVIDIA)](https://docs.pipecat.ai/server/services/tts/fastpitch), [Fish](https://docs.pipecat.ai/server/services/tts/fish), [Google](https://docs.pipecat.ai/server/services/tts/google), [LMNT](https://docs.pipecat.ai/server/services/tts/lmnt), [MiniMax](https://docs.pipecat.ai/server/services/tts/minimax), [Neuphonic](https://docs.pipecat.ai/server/services/tts/neuphonic), [OpenAI](https://docs.pipecat.ai/server/services/tts/openai), [Piper](https://docs.pipecat.ai/server/services/tts/piper), [PlayHT](https://docs.pipecat.ai/server/services/tts/playht), [Rime](https://docs.pipecat.ai/server/services/tts/rime), [XTTS](https://docs.pipecat.ai/server/services/tts/xtts) | -| Speech-to-Speech | [Gemini Multimodal Live](https://docs.pipecat.ai/server/services/s2s/gemini), [OpenAI Realtime](https://docs.pipecat.ai/server/services/s2s/openai) | -| Transport | [Daily (WebRTC)](https://docs.pipecat.ai/server/services/transport/daily), [FastAPI Websocket](https://docs.pipecat.ai/server/services/transport/fastapi-websocket), [SmallWebRTCTransport](https://docs.pipecat.ai/server/services/transport/small-webrtc), [WebSocket Server](https://docs.pipecat.ai/server/services/transport/websocket-server), Local | -| Video | [Tavus](https://docs.pipecat.ai/server/services/video/tavus), [Simli](https://docs.pipecat.ai/server/services/video/simli) | -| Memory | [mem0](https://docs.pipecat.ai/server/services/memory/mem0) | -| Vision & Image | [fal](https://docs.pipecat.ai/server/services/image-generation/fal), [Google Imagen](https://docs.pipecat.ai/server/services/image-generation/fal), [Moondream](https://docs.pipecat.ai/server/services/vision/moondream) | -| Audio Processing | [Silero VAD](https://docs.pipecat.ai/server/utilities/audio/silero-vad-analyzer), [Krisp](https://docs.pipecat.ai/server/utilities/audio/krisp-filter), [Koala](https://docs.pipecat.ai/server/utilities/audio/koala-filter), [Noisereduce](https://docs.pipecat.ai/server/utilities/audio/noisereduce-filter) | -| Analytics & Metrics | [Sentry](https://docs.pipecat.ai/server/services/analytics/sentry) | +| Category | Services | +| ------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Speech-to-Text | [AssemblyAI](https://docs.pipecat.ai/server/services/stt/assemblyai), [AWS](https://docs.pipecat.ai/server/services/stt/aws), [Azure](https://docs.pipecat.ai/server/services/stt/azure), [Deepgram](https://docs.pipecat.ai/server/services/stt/deepgram), [Fal Wizper](https://docs.pipecat.ai/server/services/stt/fal), [Gladia](https://docs.pipecat.ai/server/services/stt/gladia), [Google](https://docs.pipecat.ai/server/services/stt/google), [Groq (Whisper)](https://docs.pipecat.ai/server/services/stt/groq), [OpenAI (Whisper)](https://docs.pipecat.ai/server/services/stt/openai), [Parakeet (NVIDIA)](https://docs.pipecat.ai/server/services/stt/parakeet), [Ultravox](https://docs.pipecat.ai/server/services/stt/ultravox), [Whisper](https://docs.pipecat.ai/server/services/stt/whisper) | +| LLMs | [Anthropic](https://docs.pipecat.ai/server/services/llm/anthropic), [AWS](https://docs.pipecat.ai/server/services/llm/aws), [Azure](https://docs.pipecat.ai/server/services/llm/azure), [Cerebras](https://docs.pipecat.ai/server/services/llm/cerebras), [DeepSeek](https://docs.pipecat.ai/server/services/llm/deepseek), [Fireworks AI](https://docs.pipecat.ai/server/services/llm/fireworks), [Gemini](https://docs.pipecat.ai/server/services/llm/gemini), [Grok](https://docs.pipecat.ai/server/services/llm/grok), [Groq](https://docs.pipecat.ai/server/services/llm/groq), [NVIDIA NIM](https://docs.pipecat.ai/server/services/llm/nim), [Ollama](https://docs.pipecat.ai/server/services/llm/ollama), [OpenAI](https://docs.pipecat.ai/server/services/llm/openai), [OpenRouter](https://docs.pipecat.ai/server/services/llm/openrouter), [Perplexity](https://docs.pipecat.ai/server/services/llm/perplexity), [Qwen](https://docs.pipecat.ai/server/services/llm/qwen), [Together AI](https://docs.pipecat.ai/server/services/llm/together) | +| Text-to-Speech | [AWS](https://docs.pipecat.ai/server/services/tts/aws), [Azure](https://docs.pipecat.ai/server/services/tts/azure), [Cartesia](https://docs.pipecat.ai/server/services/tts/cartesia), [Deepgram](https://docs.pipecat.ai/server/services/tts/deepgram), [ElevenLabs](https://docs.pipecat.ai/server/services/tts/elevenlabs), [FastPitch (NVIDIA)](https://docs.pipecat.ai/server/services/tts/fastpitch), [Fish](https://docs.pipecat.ai/server/services/tts/fish), [Google](https://docs.pipecat.ai/server/services/tts/google), [LMNT](https://docs.pipecat.ai/server/services/tts/lmnt), [MiniMax](https://docs.pipecat.ai/server/services/tts/minimax), [Neuphonic](https://docs.pipecat.ai/server/services/tts/neuphonic), [OpenAI](https://docs.pipecat.ai/server/services/tts/openai), [Piper](https://docs.pipecat.ai/server/services/tts/piper), [PlayHT](https://docs.pipecat.ai/server/services/tts/playht), [Rime](https://docs.pipecat.ai/server/services/tts/rime), [Sarvam](https://docs.pipecat.ai/server/services/tts/sarvam), [XTTS](https://docs.pipecat.ai/server/services/tts/xtts) | +| Speech-to-Speech | [AWS Nova Sonic](https://docs.pipecat.ai/server/services/s2s/aws), [Gemini Multimodal Live](https://docs.pipecat.ai/server/services/s2s/gemini), [OpenAI Realtime](https://docs.pipecat.ai/server/services/s2s/openai) | +| Transport | [Daily (WebRTC)](https://docs.pipecat.ai/server/services/transport/daily), [FastAPI Websocket](https://docs.pipecat.ai/server/services/transport/fastapi-websocket), [SmallWebRTCTransport](https://docs.pipecat.ai/server/services/transport/small-webrtc), [WebSocket Server](https://docs.pipecat.ai/server/services/transport/websocket-server), Local | +| Video | [Tavus](https://docs.pipecat.ai/server/services/video/tavus), [Simli](https://docs.pipecat.ai/server/services/video/simli) | +| Memory | [mem0](https://docs.pipecat.ai/server/services/memory/mem0) | +| Vision & Image | [fal](https://docs.pipecat.ai/server/services/image-generation/fal), [Google Imagen](https://docs.pipecat.ai/server/services/image-generation/fal), [Moondream](https://docs.pipecat.ai/server/services/vision/moondream) | +| Audio Processing | [Silero VAD](https://docs.pipecat.ai/server/utilities/audio/silero-vad-analyzer), [Krisp](https://docs.pipecat.ai/server/utilities/audio/krisp-filter), [Koala](https://docs.pipecat.ai/server/utilities/audio/koala-filter), [Noisereduce](https://docs.pipecat.ai/server/utilities/audio/noisereduce-filter) | +| Analytics & Metrics | [Sentry](https://docs.pipecat.ai/server/services/analytics/sentry) | πŸ“š [View full services documentation β†’](https://docs.pipecat.ai/server/services/supported-services) diff --git a/dot-env.template b/dot-env.template index aa8068451..20d73b3ad 100644 --- a/dot-env.template +++ b/dot-env.template @@ -105,3 +105,6 @@ TWILIO_AUTH_TOKEN=... # MiniMax MINIMAX_API_KEY=... MINIMAX_GROUP_ID=... + +# Sarvam AI +SARVAM_API_KEY=... \ No newline at end of file diff --git a/examples/chatbot-audio-recording/bot.py b/examples/chatbot-audio-recording/bot.py index 5428a0d2f..128c97c7e 100644 --- a/examples/chatbot-audio-recording/bot.py +++ b/examples/chatbot-audio-recording/bot.py @@ -128,7 +128,14 @@ async def main(): ] ) - task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True)) + task = PipelineTask( + pipeline, + params=PipelineParams( + audio_in_sample_rate=16000, + audio_out_sample_rate=16000, + allow_interruptions=True, + ), + ) @audiobuffer.event_handler("on_audio_data") async def on_audio_data(buffer, audio, sample_rate, num_channels): diff --git a/examples/foundational/04a-transports-daily.py b/examples/foundational/04a-transports-daily.py index b8125c5da..98060dbab 100644 --- a/examples/foundational/04a-transports-daily.py +++ b/examples/foundational/04a-transports-daily.py @@ -37,9 +37,9 @@ async def main(): token, "Respond bot", DailyParams( + audio_in_enabled=True, audio_out_enabled=True, transcription_enabled=True, - vad_enabled=True, vad_analyzer=SileroVADAnalyzer(), ), ) diff --git a/examples/foundational/07z-interruptible-sarvam.py b/examples/foundational/07z-interruptible-sarvam.py new file mode 100644 index 000000000..fafee5e93 --- /dev/null +++ b/examples/foundational/07z-interruptible-sarvam.py @@ -0,0 +1,109 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import argparse +import os + +import aiohttp +from dotenv import load_dotenv +from loguru import logger + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.openai.llm import OpenAILLMService +from pipecat.services.sarvam.tts import SarvamTTSService +from pipecat.transcriptions.language import Language +from pipecat.transports.base_transport import TransportParams +from pipecat.transports.network.small_webrtc import SmallWebRTCTransport +from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection + +load_dotenv(override=True) + + +async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace): + logger.info(f"Starting bot") + + transport = SmallWebRTCTransport( + webrtc_connection=webrtc_connection, + params=TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + ), + ) + # Create an HTTP session + async with aiohttp.ClientSession() as session: + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = SarvamTTSService( + api_key=os.getenv("SARVAM_API_KEY"), + aiohttp_session=session, + params=SarvamTTSService.InputParams(language=Language.EN), + ) + + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) + + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, + ] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, + context_aggregator.user(), # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + report_only_initial_ttfb=True, + ), + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + # Kick off the conversation. + messages.append({"role": "system", "content": "Please introduce yourself to the user."}) + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + + @transport.event_handler("on_client_closed") + async def on_client_closed(transport, client): + logger.info(f"Client closed connection") + await task.cancel() + + runner = PipelineRunner(handle_sigint=False) + + await runner.run(task) + + +if __name__ == "__main__": + from run import main + + main() diff --git a/examples/foundational/21-tavus-layer-tavus-transport.py b/examples/foundational/21-tavus-layer-tavus-transport.py new file mode 100644 index 000000000..c9bcd2501 --- /dev/null +++ b/examples/foundational/21-tavus-layer-tavus-transport.py @@ -0,0 +1,112 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import os +import sys + +import aiohttp +from dotenv import load_dotenv +from loguru import logger + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.services.cartesia.tts import CartesiaTTSService +from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.google.llm import GoogleLLMService +from pipecat.transports.services.tavus import TavusParams, TavusTransport + +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + + +async def main(): + async with aiohttp.ClientSession() as session: + transport = TavusTransport( + bot_name="Pipecat bot", + api_key=os.getenv("TAVUS_API_KEY"), + replica_id=os.getenv("TAVUS_REPLICA_ID"), + session=session, + params=TavusParams( + audio_in_enabled=True, + audio_out_enabled=True, + microphone_out_enabled=False, + vad_analyzer=SileroVADAnalyzer(), + ), + ) + + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="a167e0f3-df7e-4d52-a9c3-f949145efdab", + ) + + llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY")) + + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, + ] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, # STT + context_aggregator.user(), # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + audio_in_sample_rate=16000, + audio_out_sample_rate=24000, + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + report_only_initial_ttfb=True, + ), + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, participant): + logger.info(f"Client connected") + # Kick off the conversation. + messages.append( + { + "role": "system", + "content": "Start by greeting the user and ask how you can help.", + } + ) + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, participant): + logger.info(f"Client disconnected") + await task.cancel() + + runner = PipelineRunner(handle_sigint=False) + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/foundational/21a-tavus-layer-small-webrtc.py b/examples/foundational/21a-tavus-layer-small-webrtc.py new file mode 100644 index 000000000..2f557b5cd --- /dev/null +++ b/examples/foundational/21a-tavus-layer-small-webrtc.py @@ -0,0 +1,125 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import argparse +import os + +import aiohttp +from dotenv import load_dotenv +from loguru import logger + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.services.cartesia.tts import CartesiaTTSService +from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.google.llm import GoogleLLMService +from pipecat.services.tavus.video import TavusVideoService +from pipecat.transports.base_transport import TransportParams +from pipecat.transports.network.small_webrtc import SmallWebRTCTransport +from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection + +load_dotenv(override=True) + + +async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace): + logger.info(f"Starting bot") + async with aiohttp.ClientSession() as session: + transport = SmallWebRTCTransport( + webrtc_connection=webrtc_connection, + params=TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + video_out_enabled=True, + video_out_is_live=True, + vad_analyzer=SileroVADAnalyzer(), + video_out_width=1280, + video_out_height=720, + ), + ) + + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="a167e0f3-df7e-4d52-a9c3-f949145efdab", + ) + + llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY")) + + tavus = TavusVideoService( + api_key=os.getenv("TAVUS_API_KEY"), + replica_id=os.getenv("TAVUS_REPLICA_ID"), + session=session, + ) + + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, + ] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, # STT + context_aggregator.user(), # User responses + llm, # LLM + tts, # TTS + tavus, # Tavus output layer + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + audio_in_sample_rate=16000, + audio_out_sample_rate=24000, + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + report_only_initial_ttfb=True, + ), + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + # Kick off the conversation. + messages.append( + { + "role": "system", + "content": "Start by greeting the user and ask how you can help.", + } + ) + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + + @transport.event_handler("on_client_closed") + async def on_client_closed(transport, client): + logger.info(f"Client closed connection") + await task.cancel() + + runner = PipelineRunner(handle_sigint=False) + + await runner.run(task) + + +if __name__ == "__main__": + from run import main + + main() diff --git a/examples/foundational/21-tavus-layer.py b/examples/foundational/21b-tavus-layer-daily-transport.py similarity index 64% rename from examples/foundational/21-tavus-layer.py rename to examples/foundational/21b-tavus-layer-daily-transport.py index ffe95e074..564828136 100644 --- a/examples/foundational/21-tavus-layer.py +++ b/examples/foundational/21b-tavus-layer-daily-transport.py @@ -7,9 +7,9 @@ import asyncio import os import sys -from typing import Any, Mapping import aiohttp +from daily_runner import configure from dotenv import load_dotenv from loguru import logger @@ -20,7 +20,7 @@ from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.deepgram.stt import DeepgramSTTService -from pipecat.services.openai.llm import OpenAILLMService +from pipecat.services.google.llm import GoogleLLMService from pipecat.services.tavus.video import TavusVideoService from pipecat.transports.services.daily import DailyParams, DailyTransport @@ -32,23 +32,20 @@ logger.add(sys.stderr, level="DEBUG") async def main(): async with aiohttp.ClientSession() as session: - tavus = TavusVideoService( - api_key=os.getenv("TAVUS_API_KEY"), - replica_id=os.getenv("TAVUS_REPLICA_ID"), - session=session, - ) - - # get persona, look up persona_name, set this as the bot name to ignore - persona_name = await tavus.get_persona_name() - room_url = await tavus.initialize() + (room_url, token) = await configure(session) transport = DailyTransport( - room_url=room_url, - token=None, - bot_name="Pipecat bot", - params=DailyParams( + room_url, + token, + "Pipecat bot", + DailyParams( audio_in_enabled=True, + audio_out_enabled=True, + video_out_enabled=True, + video_out_is_live=True, vad_analyzer=SileroVADAnalyzer(), + video_out_width=1280, + video_out_height=720, ), ) @@ -59,7 +56,13 @@ async def main(): voice_id="a167e0f3-df7e-4d52-a9c3-f949145efdab", ) - llm = OpenAILLMService(model="gpt-4o-mini") + llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY")) + + tavus = TavusVideoService( + api_key=os.getenv("TAVUS_API_KEY"), + replica_id=os.getenv("TAVUS_REPLICA_ID"), + session=session, + ) messages = [ { @@ -87,10 +90,8 @@ async def main(): task = PipelineTask( pipeline, params=PipelineParams( - # We just use 16000 because that's what Tavus is expecting and - # we avoid resampling. audio_in_sample_rate=16000, - audio_out_sample_rate=16000, + audio_out_sample_rate=24000, allow_interruptions=True, enable_metrics=True, enable_usage_metrics=True, @@ -98,33 +99,22 @@ async def main(): ), ) - @transport.event_handler("on_participant_joined") - async def on_participant_joined( - transport: DailyTransport, participant: Mapping[str, Any] - ) -> None: - # Ignore the Tavus replica's microphone - if participant.get("info", {}).get("userName", "") == persona_name: - logger.debug(f"Ignoring {participant['id']}'s microphone") - await transport.update_subscriptions( - participant_settings={ - participant["id"]: { - "media": {"microphone": "unsubscribed"}, - } - } - ) - - if participant.get("info", {}).get("userName", "") != persona_name: - # Kick off the conversation. - messages.append( - {"role": "system", "content": "Please introduce yourself to the user."} - ) - await task.queue_frames([context_aggregator.user().get_context_frame()]) + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + # Kick off the conversation. + messages.append( + { + "role": "system", + "content": "Start by greeting the user and ask how you can help.", + } + ) + await task.queue_frames([context_aggregator.user().get_context_frame()]) @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): await task.cancel() - runner = PipelineRunner() + runner = PipelineRunner(handle_sigint=False) await runner.run(task) diff --git a/examples/foundational/29-turn-tracking-observer.py b/examples/foundational/29-turn-tracking-observer.py index 0e9c7acec..08c39fe55 100644 --- a/examples/foundational/29-turn-tracking-observer.py +++ b/examples/foundational/29-turn-tracking-observer.py @@ -11,6 +11,7 @@ from dotenv import load_dotenv from loguru import logger from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.observers.loggers.user_bot_latency_log_observer import UserBotLatencyLogObserver from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -76,6 +77,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac enable_usage_metrics=True, report_only_initial_ttfb=True, ), + observers=[UserBotLatencyLogObserver()], ) turn_observer = task.turn_tracking_observer diff --git a/examples/foundational/README.md b/examples/foundational/README.md index 14323d189..3c77a7bef 100644 --- a/examples/foundational/README.md +++ b/examples/foundational/README.md @@ -95,7 +95,7 @@ Depending on what you're trying to build, these learning paths will guide you th - **[18-gstreamer-filesrc.py](./18-gstreamer-filesrc.py)**: GStreamer video streaming (Video processing) - **[19-openai-realtime-beta.py](./19-openai-realtime-beta.py)**: OpenAI Speech-to-Speech (Direct S2S, Function calls) -- **[21-tavus-layer.py](./21-tavus-layer.py)**: Tavus digital twin (Avatar integration) +- **[21-tavus-layer-tavus-transport.py](./21-tavus-layer-tavus-transport.py)**: Tavus digital twin (Avatar integration) - **[27-simli-layer.py](./27-simli-layer.py)**: Simli avatar integration (Video synchronization) ### Performance & Optimization diff --git a/examples/open-telemetry-tracing-langfuse/README.md b/examples/open-telemetry-tracing-langfuse/README.md deleted file mode 100644 index 266c62757..000000000 --- a/examples/open-telemetry-tracing-langfuse/README.md +++ /dev/null @@ -1,140 +0,0 @@ -# Langfuse Tracing for Pipecat via OpenTelemetry - -This demo showcases [Langfuse](https://langfuse.com) tracing integration for Pipecat services via OpenTelemetry, allowing you to visualize service calls, performance metrics, and dependencies. - -This is a fork of the [OpenTelemetry Tracing for Pipecat](../open-telemetry-tracing) demo, but uses Langfuse instead of Jaeger. In contrast to the original demo, this demo uses the `opentelemetry-exporter-otlp-proto-http` exporter as the `grpc` exporter is not supported by Langfuse. - -Pipecat trace in Langfuse: - -https://github.com/user-attachments/assets/13dd7431-bf5e-42e3-8d6d-2ed84c51195d - -## Features - -- **Hierarchical Tracing**: Track entire conversations, turns, and service calls -- **Service Tracing**: Detailed spans for TTS, STT, and LLM services with rich context -- **TTFB Metrics**: Capture Time To First Byte metrics for latency analysis -- **Usage Statistics**: Track character counts for TTS and token usage for LLMs - -## Trace Structure - -Traces are organized hierarchically: - -``` -Conversation (conversation-uuid) -β”œβ”€β”€ turn-1 -β”‚ β”œβ”€β”€ stt_deepgramsttservice -β”‚ β”œβ”€β”€ llm_openaillmservice -β”‚ └── tts_cartesiattsservice -└── turn-2 - β”œβ”€β”€ stt_deepgramsttservice - β”œβ”€β”€ llm_openaillmservice - └── tts_cartesiattsservice - turn-N - └── ... -``` - -This organization helps you track conversation-to-conversation and turn-to-turn. - -## Setup Instructions - -### 1. Create a Langfuse Project and get API keys - -[Self-host](https://langfuse.com/self-hosting) Langfuse or create a free [Langfuse Cloud](https://cloud.langfuse.com) account. -Create a new project and get the API keys. - -### 2. Environment Configuration - -Base64 encode your Langfuse public and secret key: - -```bash -echo -n "pk-lf-1234567890:sk-lf-1234567890" | base64 -``` - -Create a `.env` file with your API keys to enable tracing: - -``` -ENABLE_TRACING=true -# OTLP endpoint (defaults to localhost:4317 if not set) -OTEL_EXPORTER_OTLP_ENDPOINT=http://cloud.langfuse.com/api/public/otel -OTEL_EXPORTER_OTLP_HEADERS=Authorization=Basic%20 -# Set to any value to enable console output for debugging -# OTEL_CONSOLE_EXPORT=true -``` - -### 3. Configure Your Pipeline Task - -Enable tracing in your Pipecat application: - -```python -# Initialize OpenTelemetry with your chosen exporter -from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter - -# Configured automatically from .env -exporter = OTLPSpanExporter() - -setup_tracing( - service_name="pipecat-demo", - exporter=exporter, - console_export=os.getenv("OTEL_CONSOLE_EXPORT", "false").lower() == "true", -) - -# Enable tracing in your PipelineTask -task = PipelineTask( - pipeline, - params=PipelineParams( - allow_interruptions=True, - enable_metrics=True, # Required for some service metrics - ), - enable_tracing=True, # Enables both turn and conversation tracing - conversation_id="customer-123", # Optional - will auto-generate if not provided -) -``` - -### 4. Install Dependencies - -```bash -pip install -r requirements.txt -``` - -### 5. Run the Demo - -```bash -python bot.py -``` - -### 6. View Traces in Langfuse - -Open your browser to [https://cloud.langfuse.com](https://cloud.langfuse.com) to view traces. - -## Understanding the Traces - -- **Conversation Spans**: The top-level span representing an entire conversation -- **Turn Spans**: Child spans of conversations that represent each turn in the dialog -- **Service Spans**: Detailed service operations nested under turns -- **Service Attributes**: Each service includes rich context about its operation: - - **TTS**: Voice ID, character count, service type - - **STT**: Transcription text, language, model - - **LLM**: Messages, tokens used, model, service configuration -- **Metrics**: Performance data like `metrics.ttfb_ms` and processing durations - -## How It Works - -The tracing system consists of: - -1. **TurnTrackingObserver**: Detects conversation turns -2. **TurnTraceObserver**: Creates spans for turns and conversations -3. **Service Decorators**: `@traced_tts`, `@traced_stt`, `@traced_llm` for service-specific tracing -4. **Context Providers**: Share context between different parts of the pipeline - -## Troubleshooting - -- **No Traces in Langfuse**: Ensure that your credentials are correct and follow this [troubleshooting guide](https://langfuse.com/faq/all/missing-traces) -- **Debugging Traces**: Set `OTEL_CONSOLE_EXPORT=true` to print traces to the console for debugging -- **Missing Metrics**: Check that `enable_metrics=True` in PipelineParams -- **Connection Errors**: Verify network connectivity to Langfuse -- **Exporter Issues**: Try the Console exporter (`OTEL_CONSOLE_EXPORT=true`) to verify tracing works - -## References - -- [OpenTelemetry Python Documentation](https://opentelemetry-python.readthedocs.io/) -- [Langfuse OpenTelemetry Documentation](https://langfuse.com/docs/opentelemetry/get-started) diff --git a/examples/open-telemetry-tracing/README.md b/examples/open-telemetry-tracing/README.md deleted file mode 100644 index 8695a8751..000000000 --- a/examples/open-telemetry-tracing/README.md +++ /dev/null @@ -1,176 +0,0 @@ -# OpenTelemetry Tracing for Pipecat - -This demo showcases OpenTelemetry tracing integration for Pipecat services, allowing you to visualize service calls, performance metrics, and dependencies in a Jaeger dashboard. - -## Features - -- **Hierarchical Tracing**: Track entire conversations, turns, and service calls -- **Service Tracing**: Detailed spans for TTS, STT, and LLM services with rich context -- **TTFB Metrics**: Capture Time To First Byte metrics for latency analysis -- **Usage Statistics**: Track character counts for TTS and token usage for LLMs -- **Flexible Exporters**: Use Jaeger, Zipkin, or any OpenTelemetry-compatible backend - -## Trace Structure - -Traces are organized hierarchically: - -``` -Conversation (conversation-uuid) -β”œβ”€β”€ turn-1 -β”‚ β”œβ”€β”€ stt_deepgramsttservice -β”‚ β”œβ”€β”€ llm_openaillmservice -β”‚ └── tts_cartesiattsservice -└── turn-2 - β”œβ”€β”€ stt_deepgramsttservice - β”œβ”€β”€ llm_openaillmservice - └── tts_cartesiattsservice - turn-N - └── ... -``` - -This organization helps you track conversation-to-conversation and turn-to-turn. - -## Setup Instructions - -### 1. Start the Jaeger Container - -Run Jaeger in Docker to collect and visualize traces: - -```bash -docker run -d --name jaeger \ - -e COLLECTOR_ZIPKIN_HOST_PORT=:9411 \ - -p 16686:16686 \ - -p 4317:4317 \ - -p 4318:4318 \ - jaegertracing/all-in-one:latest -``` - -### 2. Environment Configuration - -Create a `.env` file with your API keys and enable tracing: - -``` -ENABLE_TRACING=true -OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 # Point to your preferred backend -# OTEL_CONSOLE_EXPORT=true # Set to any value for debug output to console - -# Service API keys -DEEPGRAM_API_KEY=your_key_here -CARTESIA_API_KEY=your_key_here -OPENAI_API_KEY=your_key_here -``` - -### 3. Configure Your Pipeline Task - -Enable tracing in your Pipecat application: - -```python -# Initialize OpenTelemetry with your chosen exporter -from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter - -exporter = OTLPSpanExporter( - endpoint="http://localhost:4317", # Jaeger OTLP endpoint - insecure=True, -) - -setup_tracing( - service_name="pipecat-demo", - exporter=exporter, - console_export=os.getenv("OTEL_CONSOLE_EXPORT", "false").lower() == "true", -) - -# Enable tracing in your PipelineTask -task = PipelineTask( - pipeline, - params=PipelineParams( - allow_interruptions=True, - enable_metrics=True, # Required for some service metrics - ), - enable_tracing=True, # Enables both turn and conversation tracing - conversation_id="customer-123", # Optional - will auto-generate if not provided -) -``` - -### 4. Exporter Options - -While this demo uses Jaeger, you can configure any OpenTelemetry-compatible exporter: - -#### Jaeger (Default for the demo) - -```python -from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter - -exporter = OTLPSpanExporter( - endpoint="http://localhost:4317", # Jaeger OTLP endpoint - insecure=True, -) -``` - -#### Cloud Providers - -Many cloud providers offer OpenTelemetry-compatible observability services: - -- AWS X-Ray -- Google Cloud Trace -- Azure Monitor -- Datadog APM - -See the OpenTelemetry documentation for specific exporter configurations: -https://opentelemetry.io/ecosystem/vendors/ - -#### LLM Tracing and Evaluation Providers - -Many LLM-focused tracing and evaluation projects support OpenTelemetry, for example: - -- Langfuse ([integration example](../open-telemetry-tracing-langfuse/)) -- Arize Phoenix - -### 5. Install Dependencies - -```bash -pip install -r requirements.txt -``` - -### 6. Run the Demo - -```bash -python bot.py -``` - -### 7. View Traces in Jaeger - -Open your browser to [http://localhost:16686](http://localhost:16686) and select the "pipecat-demo" service to view traces. - -## Understanding the Traces - -- **Conversation Spans**: The top-level span representing an entire conversation -- **Turn Spans**: Child spans of conversations that represent each turn in the dialog -- **Service Spans**: Detailed service operations nested under turns -- **Service Attributes**: Each service includes rich context about its operation: - - **TTS**: Voice ID, character count, service type - - **STT**: Transcription text, language, model - - **LLM**: Messages, tokens used, model, service configuration -- **Metrics**: Performance data like `metrics.ttfb_ms` and processing durations - -## How It Works - -The tracing system consists of: - -1. **TurnTrackingObserver**: Detects conversation turns -2. **TurnTraceObserver**: Creates spans for turns and conversations -3. **Service Decorators**: `@traced_tts`, `@traced_stt`, `@traced_llm` for service-specific tracing -4. **Context Providers**: Share context between different parts of the pipeline - -## Troubleshooting - -- **No Traces in Jaeger**: Ensure the Docker container is running and the OTLP endpoint is correct -- **Debugging Traces**: Set `OTEL_CONSOLE_EXPORT=true` to print traces to the console for debugging -- **Missing Metrics**: Check that `enable_metrics=True` in PipelineParams -- **Connection Errors**: Verify network connectivity to the Jaeger container -- **Exporter Issues**: Try the Console exporter (`OTEL_CONSOLE_EXPORT=true`) to verify tracing works -- **Other Backends**: If using a different backend, ensure you've configured the correct exporter and endpoint - -## References - -- [OpenTelemetry Python Documentation](https://opentelemetry-python.readthedocs.io/) -- [Jaeger Documentation](https://www.jaegertracing.io/docs/latest/) diff --git a/examples/open-telemetry-tracing/run.py b/examples/open-telemetry-tracing/run.py deleted file mode 100644 index e7012c9e9..000000000 --- a/examples/open-telemetry-tracing/run.py +++ /dev/null @@ -1,205 +0,0 @@ -# -# Copyright (c) 2024–2025, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -import argparse -import asyncio -import importlib.util -import os -import sys -from contextlib import asynccontextmanager -from inspect import iscoroutinefunction, signature -from typing import Any, Callable, Dict, Optional, Tuple - -import uvicorn -from dotenv import load_dotenv -from fastapi import BackgroundTasks, FastAPI -from fastapi.responses import RedirectResponse -from loguru import logger -from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI - -from pipecat.transports.network.webrtc_connection import IceServer, SmallWebRTCConnection - -# Load environment variables -load_dotenv(override=True) - -app = FastAPI() - -# Store connections by pc_id -pcs_map: Dict[str, SmallWebRTCConnection] = {} - -ice_servers = [ - IceServer( - urls="stun:stun.l.google.com:19302", - ) -] - -# Mount the frontend at / -app.mount("/client", SmallWebRTCPrebuiltUI) - -# Store program arguments -args: argparse.Namespace = argparse.Namespace() - -# Store the bot module and function info -bot_module: Any = None -run_bot_func: Optional[Callable] = None -is_webrtc_bot: bool = True - - -def import_bot_file(file_path: str) -> Tuple[Any, Callable, bool]: - """Dynamically import the bot file and determine how to run it. - - Returns: - tuple: (module, run_function, is_webrtc_bot) - - module: The imported module - - run_function: Either run_bot or main function - - is_webrtc_bot: True if run_bot function exists and accepts a WebRTC connection - """ - if not os.path.exists(file_path): - raise FileNotFoundError(f"Bot file not found: {file_path}") - - # Extract module name without extension - module_name = os.path.splitext(os.path.basename(file_path))[0] - - # Load the module - spec = importlib.util.spec_from_file_location(module_name, file_path) - if not spec or not spec.loader: - raise ImportError(f"Could not load spec for {file_path}") - - module = importlib.util.module_from_spec(spec) - sys.modules[module_name] = module - spec.loader.exec_module(module) - - # Check for run_bot function first - if hasattr(module, "run_bot"): - run_func = module.run_bot - # Check if the function accepts a WebRTC connection - sig = signature(run_func) - is_webrtc = len(sig.parameters) > 0 - return module, run_func, is_webrtc - - # Fall back to main function - if hasattr(module, "main") and iscoroutinefunction(module.main): - return module, module.main, False - - raise AttributeError(f"No run_bot or async main function found in {file_path}") - - -@app.get("/", include_in_schema=False) -async def root_redirect(): - return RedirectResponse(url="/client/") - - -@app.post("/api/offer") -async def offer(request: dict, background_tasks: BackgroundTasks): - global run_bot_func, is_webrtc_bot - - if not run_bot_func: - raise RuntimeError("No bot file has been loaded") - - if not is_webrtc_bot: - return { - "error": "This bot doesn't support WebRTC connections, it's running in standalone mode" - } - - pc_id = request.get("pc_id") - - if pc_id and pc_id in pcs_map: - pipecat_connection = pcs_map[pc_id] - logger.info(f"Reusing existing connection for pc_id: {pc_id}") - await pipecat_connection.renegotiate( - sdp=request["sdp"], type=request["type"], restart_pc=request.get("restart_pc", False) - ) - else: - pipecat_connection = SmallWebRTCConnection(ice_servers) - await pipecat_connection.initialize(sdp=request["sdp"], type=request["type"]) - - @pipecat_connection.event_handler("closed") - async def handle_disconnected(webrtc_connection: SmallWebRTCConnection): - logger.info(f"Discarding peer connection for pc_id: {webrtc_connection.pc_id}") - pcs_map.pop(webrtc_connection.pc_id, None) - - # We've already checked that run_bot_func exists - assert run_bot_func is not None - background_tasks.add_task(run_bot_func, pipecat_connection, args) - - answer = pipecat_connection.get_answer() - # Updating the peer connection inside the map - pcs_map[answer["pc_id"]] = pipecat_connection - - return answer - - -@asynccontextmanager -async def lifespan(app: FastAPI): - yield # Run app - coros = [pc.close() for pc in pcs_map.values()] - await asyncio.gather(*coros) - pcs_map.clear() - - -async def run_standalone_bot() -> None: - """Run a standalone bot that doesn't require WebRTC""" - global run_bot_func - if run_bot_func is not None: - await run_bot_func() - else: - raise RuntimeError("No bot function available to run") - - -def main(parser: Optional[argparse.ArgumentParser] = None): - global args - - if not parser: - parser = argparse.ArgumentParser(description="Pipecat Bot Runner") - parser.add_argument("bot_file", nargs="?", help="Path to the bot file", default=None) - parser.add_argument( - "--host", default="localhost", help="Host for HTTP server (default: localhost)" - ) - parser.add_argument( - "--port", type=int, default=7860, help="Port for HTTP server (default: 7860)" - ) - parser.add_argument("--verbose", "-v", action="count", default=0) - args = parser.parse_args() - - logger.remove(0) - if args.verbose: - logger.add(sys.stderr, level="TRACE") - else: - logger.add(sys.stderr, level="DEBUG") - - # Infer the bot file from the caller if not provided explicitly - bot_file = args.bot_file - if bot_file is None: - # Get the __file__ of the script that called main() - import inspect - - caller_frame = inspect.stack()[1] - caller_globals = caller_frame.frame.f_globals - bot_file = caller_globals.get("__file__") - - if not bot_file: - print("❌ Could not determine the bot file. Pass it explicitly to main().") - sys.exit(1) - - # Import the bot file - try: - global run_bot_func, bot_module, is_webrtc_bot - bot_module, run_bot_func, is_webrtc_bot = import_bot_file(bot_file) - logger.info(f"Successfully loaded bot from {bot_file}") - - if is_webrtc_bot: - logger.info("Detected WebRTC-compatible bot, starting web server...") - uvicorn.run(app, host=args.host, port=args.port) - else: - logger.info("Detected standalone bot, running directly...") - asyncio.run(run_standalone_bot()) - except Exception as e: - logger.error(f"Error loading bot file: {e}") - sys.exit(1) - - -if __name__ == "__main__": - main() diff --git a/examples/open-telemetry/README.md b/examples/open-telemetry/README.md new file mode 100644 index 000000000..1d3871e94 --- /dev/null +++ b/examples/open-telemetry/README.md @@ -0,0 +1,69 @@ +# OpenTelemetry Tracing with Pipecat + +This repository demonstrates OpenTelemetry tracing integration for Pipecat services, with examples for different backends. + +## Tracing Features in Pipecat + +- **Hierarchical Tracing**: Track entire conversations, turns, and service calls +- **Service Tracing**: Detailed spans for TTS, STT, and LLM services with rich context +- **TTFB Metrics**: Capture Time To First Byte metrics for latency analysis +- **Usage Statistics**: Track character counts for TTS and token usage for LLMs + +## Trace Structure + +Traces are organized hierarchically: + +``` +Conversation (conversation) +β”œβ”€β”€ turn +β”‚ β”œβ”€β”€ stt_deepgramsttservice +β”‚ β”œβ”€β”€ llm_openaillmservice +β”‚ └── tts_cartesiattsservice +└── turn + β”œβ”€β”€ stt_deepgramsttservice + β”œβ”€β”€ llm_openaillmservice + └── tts_cartesiattsservice + turn + └── ... +``` + +This organization helps you track conversation-to-conversation and turn-to-turn interactions. + +## Available Demos + +| Demo | Description | +| ------------------------------- | ------------------------------------------------------------------------- | +| [Jaeger Tracing](./jaeger/) | Tracing with Jaeger, an open-source end-to-end distributed tracing system | +| [Langfuse Tracing](./langfuse/) | Tracing with Langfuse, a specialized platform for LLM observability | + +## Common Requirements + +- Python 3.10+ +- Pipecat and its dependencies +- API keys for the services used (Deepgram, Cartesia, OpenAI) +- The appropriate OpenTelemetry exporters + +## How Tracing Works + +The tracing system consists of: + +1. **TurnTrackingObserver**: Detects conversation turns +2. **TurnTraceObserver**: Creates spans for turns and conversations +3. **Service Decorators**: `@traced_tts`, `@traced_stt`, `@traced_llm` for service-specific tracing +4. **Context Providers**: Share context between different parts of the pipeline + +## Getting Started + +1. Choose one of the demos from the table above +2. Follow the README instructions in the respective directory + +## Common Troubleshooting + +- **Debugging Traces**: Set `OTEL_CONSOLE_EXPORT=true` to print traces to the console for debugging +- **Missing Metrics**: Check that `enable_metrics=True` in PipelineParams +- **API Key Issues**: Verify your API keys are set correctly in the .env file + +## References + +- [OpenTelemetry Python Documentation](https://opentelemetry-python.readthedocs.io/) +- [Pipecat Documentation](https://docs.pipecat.ai/server/utilities/opentelemetry) diff --git a/examples/open-telemetry/jaeger/README.md b/examples/open-telemetry/jaeger/README.md new file mode 100644 index 000000000..a19d3ee9d --- /dev/null +++ b/examples/open-telemetry/jaeger/README.md @@ -0,0 +1,80 @@ +# Jaeger Tracing for Pipecat + +This demo showcases OpenTelemetry tracing integration for Pipecat services using Jaeger, allowing you to visualize service calls, performance metrics, and dependencies. + +## Setup Instructions + +### 1. Start the Jaeger Container + +Run Jaeger in Docker to collect and visualize traces: + +```bash +docker run -d --name jaeger \ + -e COLLECTOR_ZIPKIN_HOST_PORT=:9411 \ + -p 16686:16686 \ + -p 4317:4317 \ + -p 4318:4318 \ + jaegertracing/all-in-one:latest +``` + +### 2. Environment Configuration + +Create a `.env` file with your API keys and enable tracing: + +``` +ENABLE_TRACING=true +OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 # Point to your Jaeger backend +# OTEL_CONSOLE_EXPORT=true # Set to any value for debug output to console + +# Service API keys +DEEPGRAM_API_KEY=your_key_here +CARTESIA_API_KEY=your_key_here +OPENAI_API_KEY=your_key_here +``` + +### 3. Install Dependencies + +```bash +pip install -r requirements.txt +``` + +### 4. Run the Demo + +```bash +python bot.py +``` + +### 5. View Traces in Jaeger + +Open your browser to [http://localhost:16686](http://localhost:16686) and select the "pipecat-demo" service to view traces. + +## Jaeger-Specific Configuration + +In the `bot.py` file, note the GRPC exporter configuration: + +```python +from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter + +# Create the exporter +otlp_exporter = OTLPSpanExporter( + endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317"), + insecure=True, +) + +# Set up tracing with the exporter +setup_tracing( + service_name="pipecat-demo", + exporter=otlp_exporter, + console_export=bool(os.getenv("OTEL_CONSOLE_EXPORT")), +) +``` + +## Troubleshooting + +- **No Traces in Jaeger**: Ensure the Docker container is running and the OTLP endpoint is correct +- **Connection Errors**: Verify network connectivity to the Jaeger container +- **Exporter Issues**: Try the Console exporter (`OTEL_CONSOLE_EXPORT=true`) to verify tracing works + +## References + +- [Jaeger Documentation](https://www.jaegertracing.io/docs/latest/) diff --git a/examples/open-telemetry-tracing/bot.py b/examples/open-telemetry/jaeger/bot.py similarity index 98% rename from examples/open-telemetry-tracing/bot.py rename to examples/open-telemetry/jaeger/bot.py index 0b44c3865..18fe34ef4 100644 --- a/examples/open-telemetry-tracing/bot.py +++ b/examples/open-telemetry/jaeger/bot.py @@ -6,6 +6,7 @@ import argparse import os +import sys from dotenv import load_dotenv from loguru import logger @@ -154,6 +155,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac if __name__ == "__main__": + sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) from run import main main() diff --git a/examples/open-telemetry-tracing/env.example b/examples/open-telemetry/jaeger/env.example similarity index 100% rename from examples/open-telemetry-tracing/env.example rename to examples/open-telemetry/jaeger/env.example diff --git a/examples/open-telemetry-tracing/requirements.txt b/examples/open-telemetry/jaeger/requirements.txt similarity index 100% rename from examples/open-telemetry-tracing/requirements.txt rename to examples/open-telemetry/jaeger/requirements.txt diff --git a/examples/open-telemetry/langfuse/README.md b/examples/open-telemetry/langfuse/README.md new file mode 100644 index 000000000..002a3f64a --- /dev/null +++ b/examples/open-telemetry/langfuse/README.md @@ -0,0 +1,82 @@ +# Langfuse Tracing for Pipecat + +This demo showcases [Langfuse](https://langfuse.com) tracing integration for Pipecat services via OpenTelemetry, allowing you to visualize service calls, performance metrics, and dependencies with a focus on LLM observability. + +Pipecat trace in Langfuse: + +https://github.com/user-attachments/assets/13dd7431-bf5e-42e3-8d6d-2ed84c51195d + +## Setup Instructions + +### 1. Create a Langfuse Project and get API keys + +[Self-host](https://langfuse.com/self-hosting) Langfuse or create a free [Langfuse Cloud](https://cloud.langfuse.com) account. +Create a new project and get the API keys. + +### 2. Environment Configuration + +Base64 encode your Langfuse public and secret key: + +```bash +echo -n "pk-lf-1234567890:sk-lf-1234567890" | base64 +``` + +Create a `.env` file with your API keys to enable tracing: + +``` +ENABLE_TRACING=true +# OTLP endpoint for Langfuse +OTEL_EXPORTER_OTLP_ENDPOINT=http://cloud.langfuse.com/api/public/otel +OTEL_EXPORTER_OTLP_HEADERS=Authorization=Basic%20 +# Set to any value to enable console output for debugging +# OTEL_CONSOLE_EXPORT=true + +# Service API keys +DEEPGRAM_API_KEY=your_key_here +CARTESIA_API_KEY=your_key_here +OPENAI_API_KEY=your_key_here +``` + +### 3. Install Dependencies + +```bash +pip install -r requirements.txt +``` + +### 4. Run the Demo + +```bash +python bot.py +``` + +### 5. View Traces in Langfuse + +Open your browser to [https://cloud.langfuse.com](https://cloud.langfuse.com) to view traces. + +## Langfuse-Specific Configuration + +In the `bot.py` file, note the HTTP exporter configuration: + +```python +from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter + +# Create the exporter - configured from environment variables +otlp_exporter = OTLPSpanExporter() + +# Set up tracing with the exporter +setup_tracing( + service_name="pipecat-demo", + exporter=otlp_exporter, + console_export=bool(os.getenv("OTEL_CONSOLE_EXPORT")), +) +``` + +## Troubleshooting + +- **No Traces in Langfuse**: Ensure that your credentials are correct and follow this [troubleshooting guide](https://langfuse.com/faq/all/missing-traces) +- **Connection Errors**: Verify network connectivity to Langfuse +- **Authorization Issues**: Check that your base64 encoding is correct and the API keys are valid + +## References + +- [Langfuse OpenTelemetry Documentation](https://langfuse.com/docs/opentelemetry/get-started) diff --git a/examples/open-telemetry-tracing-langfuse/bot.py b/examples/open-telemetry/langfuse/bot.py similarity index 98% rename from examples/open-telemetry-tracing-langfuse/bot.py rename to examples/open-telemetry/langfuse/bot.py index f4d6d76ac..9f311970e 100644 --- a/examples/open-telemetry-tracing-langfuse/bot.py +++ b/examples/open-telemetry/langfuse/bot.py @@ -6,6 +6,7 @@ import argparse import os +import sys from dotenv import load_dotenv from loguru import logger @@ -151,6 +152,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac if __name__ == "__main__": + sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) from run import main main() diff --git a/examples/open-telemetry-tracing-langfuse/env.example b/examples/open-telemetry/langfuse/env.example similarity index 100% rename from examples/open-telemetry-tracing-langfuse/env.example rename to examples/open-telemetry/langfuse/env.example diff --git a/examples/open-telemetry-tracing-langfuse/requirements.txt b/examples/open-telemetry/langfuse/requirements.txt similarity index 100% rename from examples/open-telemetry-tracing-langfuse/requirements.txt rename to examples/open-telemetry/langfuse/requirements.txt diff --git a/examples/open-telemetry-tracing-langfuse/run.py b/examples/open-telemetry/run.py similarity index 100% rename from examples/open-telemetry-tracing-langfuse/run.py rename to examples/open-telemetry/run.py diff --git a/pyproject.toml b/pyproject.toml index b8db368f9..4572e3ba7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -47,7 +47,7 @@ azure = [ "azure-cognitiveservices-speech~=1.42.0"] cartesia = [ "cartesia~=2.0.3", "websockets~=13.1" ] cerebras = [] deepseek = [] -daily = [ "daily-python~=0.18.2" ] +daily = [ "daily-python~=0.19.0" ] deepgram = [ "deepgram-sdk~=3.8.0" ] elevenlabs = [ "websockets~=13.1" ] fal = [ "fal-client~=0.5.9" ] diff --git a/src/pipecat/audio/vad/silero.py b/src/pipecat/audio/vad/silero.py index 26d9368a8..cb1dc7631 100644 --- a/src/pipecat/audio/vad/silero.py +++ b/src/pipecat/audio/vad/silero.py @@ -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) diff --git a/src/pipecat/observers/loggers/user_bot_latency_log_observer.py b/src/pipecat/observers/loggers/user_bot_latency_log_observer.py new file mode 100644 index 000000000..f601a9e9e --- /dev/null +++ b/src/pipecat/observers/loggers/user_bot_latency_log_observer.py @@ -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}") diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 449bf3e33..7bd54237d 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -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() diff --git a/src/pipecat/services/anthropic/llm.py b/src/pipecat/services/anthropic/llm.py index 5e0ea0ad2..2c4e9078d 100644 --- a/src/pipecat/services/anthropic/llm.py +++ b/src/pipecat/services/anthropic/llm.py @@ -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, diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py index 6f6584894..aeab4e787 100644 --- a/src/pipecat/services/elevenlabs/tts.py +++ b/src/pipecat/services/elevenlabs/tts.py @@ -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) diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 10b284890..8bc6fea53 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -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, diff --git a/src/pipecat/services/sarvam/__init__.py b/src/pipecat/services/sarvam/__init__.py new file mode 100644 index 000000000..0d444e949 --- /dev/null +++ b/src/pipecat/services/sarvam/__init__.py @@ -0,0 +1,8 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + + +from .tts import * diff --git a/src/pipecat/services/sarvam/tts.py b/src/pipecat/services/sarvam/tts.py new file mode 100644 index 000000000..f9ce4e70f --- /dev/null +++ b/src/pipecat/services/sarvam/tts.py @@ -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() diff --git a/src/pipecat/services/tavus/video.py b/src/pipecat/services/tavus/video.py index 3699ba512..8fa30db8a 100644 --- a/src/pipecat/services/tavus/video.py +++ b/src/pipecat/services/tavus/video.py @@ -7,10 +7,11 @@ """This module implements Tavus as a sink transport layer""" import asyncio -import base64 +import time from typing import Optional import aiohttp +from daily.daily import AudioData, VideoFrame from loguru import logger from pipecat.audio.utils import create_default_resampler @@ -18,19 +19,38 @@ from pipecat.frames.frames import ( CancelFrame, EndFrame, Frame, + OutputAudioRawFrame, + OutputImageRawFrame, StartFrame, StartInterruptionFrame, - TransportMessageUrgentFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, ) -from pipecat.processors.frame_processor import FrameDirection +from pipecat.processors.frame_processor import FrameDirection, FrameProcessorSetup from pipecat.services.ai_service import AIService +from pipecat.transports.services.tavus import TavusCallbacks, TavusParams, TavusTransportClient class TavusVideoService(AIService): - """Class to send base64 encoded audio to Tavus""" + """ + Service class that proxies audio to Tavus and receives both audio and video in return. + + It uses the `TavusTransportClient` to manage the session and handle communication. When + audio is sent, Tavus responds with both audio and video streams, which are then routed + through Pipecat’s media pipeline. + + In use cases such as with `DailyTransport`, this results in two distinct virtual rooms: + - **Tavus room**: Contains the Tavus Avatar and the Pipecat Bot. + - **User room**: Contains the Pipecat Bot and the user. + + Args: + api_key (str): Tavus API key used for authentication. + replica_id (str): ID of the Tavus voice replica to use for speech synthesis. + persona_id (str): ID of the Tavus persona. Defaults to "pipecat0" to use the Pipecat TTS voice. + session (aiohttp.ClientSession): Async HTTP session used for communication with Tavus. + **kwargs: Additional arguments passed to the parent `AIService` class. + """ def __init__( self, @@ -39,54 +59,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) diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index f9a27a6d3..0dea8efdc 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -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 diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 81492b84d..f26602ed7 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -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) diff --git a/src/pipecat/transports/network/webrtc_connection.py b/src/pipecat/transports/network/webrtc_connection.py index 3981460e7..49aa2b1da 100644 --- a/src/pipecat/transports/network/webrtc_connection.py +++ b/src/pipecat/transports/network/webrtc_connection.py @@ -144,6 +144,7 @@ class SmallWebRTCConnection(BaseObject): self._renegotiation_in_progress = False self._last_received_time = None self._message_queue = [] + self._pending_app_messages = [] def _setup_listeners(self): @self._pc.on("datachannel") @@ -170,7 +171,11 @@ class SmallWebRTCConnection(BaseObject): if json_message["type"] == SIGNALLING_TYPE and json_message.get("message"): self._handle_signalling_message(json_message["message"]) else: - await self._call_event_handler("app-message", json_message) + if self.is_connected(): + await self._call_event_handler("app-message", json_message) + else: + logger.debug("Client not connected. Queuing app-message.") + self._pending_app_messages.append(json_message) except Exception as e: logger.exception(f"Error parsing JSON message {message}, {e}") @@ -225,6 +230,9 @@ class SmallWebRTCConnection(BaseObject): # If we already connected, trigger again the connected event if self.is_connected(): await self._call_event_handler("connected") + logger.debug("Flushing pending app-messages") + for message in self._pending_app_messages: + await self._call_event_handler("app-message", message) # We are renegotiating here, because likely we have loose the first video frames # and aiortc does not handle that pretty well. video_input_track = self.video_input_track() @@ -293,6 +301,7 @@ class SmallWebRTCConnection(BaseObject): if self._pc: await self._pc.close() self._message_queue.clear() + self._pending_app_messages.clear() self._track_map = {} def get_answer(self): diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 095084037..7fe3ed0dd 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -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) diff --git a/src/pipecat/transports/services/livekit.py b/src/pipecat/transports/services/livekit.py index 53a9b1789..143163ed3 100644 --- a/src/pipecat/transports/services/livekit.py +++ b/src/pipecat/transports/services/livekit.py @@ -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, diff --git a/src/pipecat/transports/services/tavus.py b/src/pipecat/transports/services/tavus.py new file mode 100644 index 000000000..8d704a242 --- /dev/null +++ b/src/pipecat/transports/services/tavus.py @@ -0,0 +1,532 @@ +import asyncio +import base64 +import time +from functools import partial +from typing import Any, Awaitable, Callable, Mapping, Optional + +import aiohttp +from daily.daily import AudioData +from loguru import logger +from pydantic import BaseModel + +from pipecat.audio.utils import create_default_resampler +from pipecat.frames.frames import ( + CancelFrame, + EndFrame, + Frame, + InputAudioRawFrame, + OutputImageRawFrame, + StartFrame, + StartInterruptionFrame, + TransportMessageFrame, + TransportMessageUrgentFrame, + TTSStartedFrame, + TTSStoppedFrame, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup +from pipecat.transports.base_input import BaseInputTransport +from pipecat.transports.base_output import BaseOutputTransport +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.services.daily import ( + DailyCallbacks, + DailyParams, + DailyTransportClient, +) + + +class TavusApi: + """ + A helper class for interacting with the Tavus API (v2). + """ + + BASE_URL = "https://tavusapi.com/v2" + + def __init__(self, api_key: str, session: aiohttp.ClientSession): + """ + Initialize the TavusApi client. + + Args: + api_key (str): Tavus API key. + session (aiohttp.ClientSession): An aiohttp session for making HTTP requests. + """ + self._api_key = api_key + self._session = session + self._headers = {"Content-Type": "application/json", "x-api-key": self._api_key} + + async def create_conversation(self, replica_id: str, persona_id: str) -> dict: + logger.debug(f"Creating Tavus conversation: replica={replica_id}, persona={persona_id}") + url = f"{self.BASE_URL}/conversations" + payload = { + "replica_id": replica_id, + "persona_id": persona_id, + } + async with self._session.post(url, headers=self._headers, json=payload) as r: + r.raise_for_status() + response = await r.json() + logger.debug(f"Created Tavus conversation: {response}") + return response + + async def end_conversation(self, conversation_id: str): + if conversation_id is None: + return + + url = f"{self.BASE_URL}/conversations/{conversation_id}/end" + async with self._session.post(url, headers=self._headers) as r: + r.raise_for_status() + logger.debug(f"Ended Tavus conversation {conversation_id}") + + async def get_persona_name(self, persona_id: str) -> str: + url = f"{self.BASE_URL}/personas/{persona_id}" + async with self._session.get(url, headers=self._headers) as r: + r.raise_for_status() + response = await r.json() + logger.debug(f"Fetched Tavus persona: {response}") + return response["persona_name"] + + +class TavusCallbacks(BaseModel): + """Callback handlers for the Tavus events. + + Attributes: + on_participant_joined: Called when a participant joins. + on_participant_left: Called when a participant leaves. + """ + + on_participant_joined: Callable[[Mapping[str, Any]], Awaitable[None]] + on_participant_left: Callable[[Mapping[str, Any], str], Awaitable[None]] + + +class TavusParams(DailyParams): + """Configuration parameters for the Tavus transport.""" + + audio_in_enabled: bool = True + audio_out_enabled: bool = True + microphone_out_enabled: bool = False + + +class TavusTransportClient: + """ + A transport client that integrates a Pipecat Bot with the Tavus platform by managing + conversation sessions using the Tavus API. + + This client uses `TavusApi` to interact with the Tavus backend services. When a conversation + is started via `TavusApi`, Tavus provides a `roomURL` that can be used to connect the Pipecat Bot + into the same virtual room where the TavusBot is operating. + + Args: + bot_name (str): The name of the Pipecat bot instance. + params (TavusParams): Optional parameters for Tavus operation. Defaults to `TavusParams()`. + callbacks (TavusCallbacks): Callback handlers for Tavus-related events. + api_key (str): API key for authenticating with Tavus API. + replica_id (str): ID of the replica to use in the Tavus conversation. + persona_id (str): ID of the Tavus persona. Defaults to "pipecat0", which signals Tavus to use + the TTS voice of the Pipecat bot instead of a Tavus persona voice. + session (aiohttp.ClientSession): The aiohttp session for making async HTTP requests. + sample_rate: Audio sample rate to be used by the client. + """ + + def __init__( + self, + *, + bot_name: str, + params: TavusParams = TavusParams(), + callbacks: TavusCallbacks, + api_key: str, + replica_id: str, + persona_id: str = "pipecat0", # Use `pipecat0` so that your TTS voice is used in place of the Tavus persona + session: aiohttp.ClientSession, + ) -> None: + self._bot_name = bot_name + self._api = TavusApi(api_key, session) + self._replica_id = replica_id + self._persona_id = persona_id + self._conversation_id: Optional[str] = None + self._other_participant_has_joined = False + self._client: Optional[DailyTransportClient] = None + self._callbacks = callbacks + self._params = params + + async def _initialize(self) -> str: + response = await self._api.create_conversation(self._replica_id, self._persona_id) + self._conversation_id = response["conversation_id"] + return response["conversation_url"] + + async def setup(self, setup: FrameProcessorSetup): + if self._conversation_id is not None: + return + try: + room_url = await self._initialize() + daily_callbacks = DailyCallbacks( + on_active_speaker_changed=partial( + self._on_handle_callback, "on_active_speaker_changed" + ), + on_joined=self._on_joined, + on_left=self._on_left, + on_error=partial(self._on_handle_callback, "on_error"), + on_app_message=partial(self._on_handle_callback, "on_app_message"), + on_call_state_updated=partial(self._on_handle_callback, "on_call_state_updated"), + on_client_connected=partial(self._on_handle_callback, "on_client_connected"), + on_client_disconnected=partial(self._on_handle_callback, "on_client_disconnected"), + on_dialin_connected=partial(self._on_handle_callback, "on_dialin_connected"), + on_dialin_ready=partial(self._on_handle_callback, "on_dialin_ready"), + on_dialin_stopped=partial(self._on_handle_callback, "on_dialin_stopped"), + on_dialin_error=partial(self._on_handle_callback, "on_dialin_error"), + on_dialin_warning=partial(self._on_handle_callback, "on_dialin_warning"), + on_dialout_answered=partial(self._on_handle_callback, "on_dialout_answered"), + on_dialout_connected=partial(self._on_handle_callback, "on_dialout_connected"), + on_dialout_stopped=partial(self._on_handle_callback, "on_dialout_stopped"), + on_dialout_error=partial(self._on_handle_callback, "on_dialout_error"), + on_dialout_warning=partial(self._on_handle_callback, "on_dialout_warning"), + on_participant_joined=self._callbacks.on_participant_joined, + on_participant_left=self._callbacks.on_participant_left, + on_participant_updated=partial(self._on_handle_callback, "on_participant_updated"), + on_transcription_message=partial( + self._on_handle_callback, "on_transcription_message" + ), + on_recording_started=partial(self._on_handle_callback, "on_recording_started"), + on_recording_stopped=partial(self._on_handle_callback, "on_recording_stopped"), + on_recording_error=partial(self._on_handle_callback, "on_recording_error"), + ) + self._client = DailyTransportClient( + room_url, None, "Pipecat", self._params, daily_callbacks, self._bot_name + ) + await self._client.setup(setup) + except Exception as e: + logger.error(f"Failed to setup TavusTransportClient: {e}") + await self._api.end_conversation(self._conversation_id) + + async def cleanup(self): + if self._client is None: + return + await self._client.cleanup() + self._client = None + + async def _on_joined(self, data): + logger.debug("TavusTransportClient joined!") + + async def _on_left(self): + logger.debug("TavusTransportClient left!") + + async def _on_handle_callback(self, event_name, *args, **kwargs): + logger.trace(f"[Callback] {event_name} called with args={args}, kwargs={kwargs}") + + async def get_persona_name(self) -> str: + return await self._api.get_persona_name(self._persona_id) + + async def start(self, frame: StartFrame): + logger.debug("TavusTransportClient start invoked!") + await self._client.start(frame) + await self._client.join() + + async def stop(self): + await self._client.leave() + await self._api.end_conversation(self._conversation_id) + + async def capture_participant_video( + self, + participant_id: str, + callback: Callable, + framerate: int = 30, + video_source: str = "camera", + color_format: str = "RGB", + ): + await self._client.capture_participant_video( + participant_id, callback, framerate, video_source, color_format + ) + + async def capture_participant_audio( + self, + participant_id: str, + callback: Callable, + audio_source: str = "microphone", + sample_rate: int = 16000, + callback_interval_ms: int = 20, + ): + await self._client.capture_participant_audio( + participant_id, callback, audio_source, sample_rate, callback_interval_ms + ) + + async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + await self._client.send_message(frame) + + @property + def out_sample_rate(self) -> int: + return self._client.out_sample_rate + + @property + def in_sample_rate(self) -> int: + return self._client.in_sample_rate + + async def encode_audio_and_send(self, audio: bytes, done: bool, inference_id: str): + """Encodes audio to base64 and sends it to Tavus""" + audio_base64 = base64.b64encode(audio).decode("utf-8") + await self._send_audio_message(audio_base64, done=done, inference_id=inference_id) + + async def send_interrupt_message(self) -> None: + transport_frame = TransportMessageUrgentFrame( + message={ + "message_type": "conversation", + "event_type": "conversation.interrupt", + "conversation_id": self._conversation_id, + } + ) + await self.send_message(transport_frame) + + async def _send_audio_message(self, audio_base64: str, done: bool, inference_id: str): + transport_frame = TransportMessageUrgentFrame( + message={ + "message_type": "conversation", + "event_type": "conversation.echo", + "conversation_id": self._conversation_id, + "properties": { + "modality": "audio", + "inference_id": inference_id, + "audio": audio_base64, + "done": done, + "sample_rate": self.out_sample_rate, + }, + } + ) + await self.send_message(transport_frame) + + async def update_subscriptions(self, participant_settings=None, profile_settings=None): + await self._client.update_subscriptions( + participant_settings=participant_settings, profile_settings=profile_settings + ) + + async def write_raw_audio_frames(self, frames: bytes, destination: Optional[str] = None): + await self._client.write_raw_audio_frames(frames, destination) + + +class TavusInputTransport(BaseInputTransport): + def __init__( + self, + client: TavusTransportClient, + params: TransportParams, + **kwargs, + ): + super().__init__(params, **kwargs) + self._client = client + self._params = params + self._resampler = create_default_resampler() + + async def setup(self, setup: FrameProcessorSetup): + await super().setup(setup) + await self._client.setup(setup) + + async def cleanup(self): + await super().cleanup() + await self._client.cleanup() + + async def start(self, frame: StartFrame): + await super().start(frame) + await self._client.start(frame) + await self.set_transport_ready(frame) + + async def stop(self, frame: EndFrame): + await super().stop(frame) + await self._client.stop() + + async def cancel(self, frame: CancelFrame): + await super().cancel(frame) + await self._client.stop() + + async def start_capturing_audio(self, participant): + if self._params.audio_in_enabled: + logger.info( + f"TavusTransportClient start capturing audio for participant {participant['id']}" + ) + await self._client.capture_participant_audio( + participant_id=participant["id"], + callback=self._on_participant_audio_data, + sample_rate=self._client.in_sample_rate, + ) + + async def _on_participant_audio_data( + self, participant_id: str, audio: AudioData, audio_source: str + ): + frame = InputAudioRawFrame( + audio=audio.audio_frames, + sample_rate=audio.audio_frames, + num_channels=audio.num_channels, + ) + frame.transport_source = audio_source + await self.push_audio_frame(frame) + + +class TavusOutputTransport(BaseOutputTransport): + def __init__( + self, + client: TavusTransportClient, + params: TransportParams, + **kwargs, + ): + super().__init__(params, **kwargs) + self._client = client + self._params = params + self._samples_sent = 0 + self._start_time = time.time() + + async def setup(self, setup: FrameProcessorSetup): + await super().setup(setup) + await self._client.setup(setup) + + async def cleanup(self): + await super().cleanup() + await self._client.cleanup() + + async def start(self, frame: StartFrame): + await super().start(frame) + self._samples_sent = 0 + self._start_time = time.time() + await self._client.start(frame) + await self.set_transport_ready(frame) + + async def stop(self, frame: EndFrame): + await super().stop(frame) + await self._client.stop() + + async def cancel(self, frame: CancelFrame): + await super().cancel(frame) + await self._client.stop() + + async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + logger.info(f"TavusOutputTransport sending message {frame}") + await self._client.send_message(frame) + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, StartInterruptionFrame): + await self._handle_interruptions() + elif isinstance(frame, TTSStartedFrame): + self._current_idx_str = str(frame.id) + elif isinstance(frame, TTSStoppedFrame): + logger.debug(f"TAVUS: {self}: stopped speaking") + await self._client.encode_audio_and_send(b"\x00\x00", True, self._current_idx_str) + + async def _handle_interruptions(self): + await self._client.send_interrupt_message() + + async def write_raw_audio_frames(self, frames: bytes, destination: Optional[str] = None): + # Compute wait time for synchronization + wait = self._start_time + (self._samples_sent / self._sample_rate) - time.time() + if wait > 0: + await asyncio.sleep(wait) + + await self._client.encode_audio_and_send(frames, False, self._current_idx_str) + + # Update timestamp based on number of samples sent + self._samples_sent += len(frames) // 2 # 2 bytes per sample (16-bit) + + async def write_raw_video_frame( + self, frame: OutputImageRawFrame, destination: Optional[str] = None + ): + pass + + +class TavusTransport(BaseTransport): + """ + Transport implementation for Tavus video calls. + + When used, the Pipecat bot joins the same virtual room as the Tavus Avatar and the user. + This is achieved by using `TavusTransportClient`, which initiates the conversation via + `TavusApi` and obtains a room URL that all participants connect to. + + Args: + bot_name (str): The name of the Pipecat bot. + session (aiohttp.ClientSession): aiohttp session used for async HTTP requests. + api_key (str): Tavus API key for authentication. + replica_id (str): ID of the replica model used for voice generation. + persona_id (str): ID of the Tavus persona. Defaults to "pipecat0" to use the Pipecat TTS voice. + params (TavusParams): Optional Tavus-specific configuration parameters. + input_name (Optional[str]): Optional name for the input transport. + output_name (Optional[str]): Optional name for the output transport. + """ + + def __init__( + self, + bot_name: str, + session: aiohttp.ClientSession, + api_key: str, + replica_id: str, + persona_id: str = "pipecat0", # Use `pipecat0` so that your TTS voice is used in place of the Tavus persona + params: TavusParams = TavusParams(), + input_name: Optional[str] = None, + output_name: Optional[str] = None, + ): + super().__init__(input_name=input_name, output_name=output_name) + self._params = params + + # TODO: Filipi - We can remove this if we stop sending the audio through app messages + # Limiting this so we don't go over 20 messages per second + # each message is going to have 50ms of audio + self._params.audio_out_10ms_chunks = 5 + + callbacks = TavusCallbacks( + on_participant_joined=self._on_participant_joined, + on_participant_left=self._on_participant_left, + ) + self._client = TavusTransportClient( + bot_name="Pipecat", + callbacks=callbacks, + api_key=api_key, + replica_id=replica_id, + persona_id=persona_id, + session=session, + params=params, + ) + self._input: Optional[TavusInputTransport] = None + self._output: Optional[TavusOutputTransport] = None + self._tavus_participant_id = None + + # Register supported handlers. The user will only be able to register + # these handlers. + self._register_event_handler("on_client_connected") + self._register_event_handler("on_client_disconnected") + + async def _on_participant_left(self, participant, reason): + persona_name = await self._client.get_persona_name() + if participant.get("info", {}).get("userName", "") != persona_name: + await self._on_client_disconnected(participant) + + async def _on_participant_joined(self, participant): + # get persona, look up persona_name, set this as the bot name to ignore + persona_name = await self._client.get_persona_name() + # Ignore the Tavus replica's microphone + if participant.get("info", {}).get("userName", "") == persona_name: + self._tavus_participant_id = participant["id"] + else: + await self._on_client_connected(participant) + if self._tavus_participant_id: + logger.debug(f"Ignoring {self._tavus_participant_id}'s microphone") + await self.update_subscriptions( + participant_settings={ + self._tavus_participant_id: { + "media": {"microphone": "unsubscribed"}, + } + } + ) + if self._input: + await self._input.start_capturing_audio(participant) + + async def update_subscriptions(self, participant_settings=None, profile_settings=None): + await self._client.update_subscriptions( + participant_settings=participant_settings, + profile_settings=profile_settings, + ) + + def input(self) -> FrameProcessor: + if not self._input: + self._input = TavusInputTransport(client=self._client, params=self._params) + return self._input + + def output(self) -> FrameProcessor: + if not self._output: + self._output = TavusOutputTransport(client=self._client, params=self._params) + return self._output + + async def _on_client_connected(self, participant: Any): + await self._call_event_handler("on_client_connected", participant) + + async def _on_client_disconnected(self, participant: Any): + await self._call_event_handler("on_client_disconnected", participant)