From 54c0bf0c705e6da01b361ae63877c822388b492b Mon Sep 17 00:00:00 2001 From: roey <159067767+roey-priel@users.noreply.github.com> Date: Fri, 25 Oct 2024 09:46:25 -0700 Subject: [PATCH] Adding `TavusVideoService` layer (#617) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: roey <159067767+roey-tavus@users.noreply.github.com> Co-authored-by: Mert Gerdan Co-authored-by: Aleix Conchillo Flaqué --- dot-env.template | 7 +- examples/foundational/21-tavus-layer.py | 132 +++++++++++++++++++++++ src/pipecat/services/fireworks.py | 3 +- src/pipecat/services/tavus.py | 137 ++++++++++++++++++++++++ 4 files changed, 277 insertions(+), 2 deletions(-) create mode 100644 examples/foundational/21-tavus-layer.py create mode 100644 src/pipecat/services/tavus.py diff --git a/dot-env.template b/dot-env.template index e940b1076..56a69fb78 100644 --- a/dot-env.template +++ b/dot-env.template @@ -46,5 +46,10 @@ PLAY_HT_API_KEY=... # OpenAI OPENAI_API_KEY=... -#OpenPipe +# OpenPipe OPENPIPE_API_KEY=... + +# Tavus +TAVUS_API_KEY=... +TAVUS_REPLICA_ID=... +TAVUS_PERSONA_ID=... \ No newline at end of file diff --git a/examples/foundational/21-tavus-layer.py b/examples/foundational/21-tavus-layer.py new file mode 100644 index 000000000..06cfe8dd8 --- /dev/null +++ b/examples/foundational/21-tavus-layer.py @@ -0,0 +1,132 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import aiohttp +from typing import Any +import os +import sys + +from pipecat.frames.frames import LLMMessagesFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.llm_response import ( + LLMAssistantResponseAggregator, + LLMUserResponseAggregator, +) +from pipecat.services.cartesia import CartesiaTTSService +from pipecat.services.openai import OpenAILLMService +from pipecat.services.deepgram import DeepgramSTTService +from pipecat.services.tavus import TavusVideoService +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.audio.vad.silero import SileroVADAnalyzer + +from loguru import logger + +from dotenv import load_dotenv + +load_dotenv(override=True) + +logger.remove(0) +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"), + persona_id=os.getenv("TAVUS_PERSONA_ID", "pipecat0"), + session=session, + ) + + # get persona, look up persona_name, set this as the bot name to ignore + persona_name = await tavus.get_persona_name() + room_url = await tavus.initialize() + + transport = DailyTransport( + room_url=room_url, + token=None, + bot_name="Pipecat bot", + params=DailyParams( + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + vad_audio_passthrough=True, + ), + ) + + 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 = OpenAILLMService(model="gpt-4o-mini") + + 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.", + }, + ] + + tma_in = LLMUserResponseAggregator(messages) + tma_out = LLMAssistantResponseAggregator(messages) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, # STT + tma_in, # User responses + llm, # LLM + tts, # TTS + tavus, # Tavus output layer + transport.output(), # Transport bot output + tma_out, # Assistant spoken responses + ] + ) + + task = PipelineTask( + pipeline, + PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + report_only_initial_ttfb=True, + ), + ) + + @transport.event_handler("on_participant_joined") + async def on_participant_joined( + transport: DailyTransport, participant: dict[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") + 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([LLMMessagesFrame(messages)]) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/pipecat/services/fireworks.py b/src/pipecat/services/fireworks.py index a6e826c12..632e7ad17 100644 --- a/src/pipecat/services/fireworks.py +++ b/src/pipecat/services/fireworks.py @@ -22,7 +22,8 @@ class FireworksLLMService(BaseOpenAILLMService): def __init__( self, *, + api_key: str, model: str = "accounts/fireworks/models/firefunction-v1", base_url: str = "https://api.fireworks.ai/inference/v1", ): - super().__init__(model=model, base_url=base_url) + super().__init__(api_key=api_key, model=model, base_url=base_url) diff --git a/src/pipecat/services/tavus.py b/src/pipecat/services/tavus.py new file mode 100644 index 000000000..ff2b7fb87 --- /dev/null +++ b/src/pipecat/services/tavus.py @@ -0,0 +1,137 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + + +"""This module implements Tavus as a sink transport layer""" + +import aiohttp +import base64 + +from pipecat.frames.frames import ( + Frame, + TTSAudioRawFrame, + TransportMessageUrgentFrame, + TTSStartedFrame, + TTSStoppedFrame, + StartInterruptionFrame, + EndFrame, + CancelFrame, +) +from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.ai_services import AIService +from pipecat.audio.utils import resample_audio + +from loguru import logger + + +class TavusVideoService(AIService): + """Class to send base64 encoded audio to Tavus""" + + def __init__( + self, + *, + api_key: str, + replica_id: str, + persona_id: str = "pipecat0", + session: aiohttp.ClientSession, + **kwargs, + ) -> None: + super().__init__(**kwargs) + self._api_key = api_key + self._replica_id = replica_id + self._persona_id = persona_id + self._session = session + + self._conversation_id: str + + 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() + + logger.debug(f"TavusVideoService joined {response_json['conversation_url']}") + self._conversation_id = response_json["conversation_id"] + return response_json["conversation_url"] + + 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"] + + async def _end_conversation(self) -> None: + 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() + + async def _encode_audio_and_send( + self, audio: bytes, original_sample_rate: int, done: bool + ) -> None: + """Encodes audio to base64 and sends it to Tavus""" + if not done: + audio = resample_audio(audio, original_sample_rate, 16000) + audio_base64 = base64.b64encode(audio).decode("utf-8") + logger.trace(f"TavusVideoService sending {len(audio)} bytes") + await self._send_audio_message(audio_base64, done=done) + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TTSStartedFrame): + await self.start_processing_metrics() + await self.start_ttfb_metrics() + self._current_idx_str = str(frame.id) + elif isinstance(frame, TTSAudioRawFrame): + await self._encode_audio_and_send(frame.audio, frame.sample_rate, done=False) + elif isinstance(frame, TTSStoppedFrame): + await self._encode_audio_and_send(b"\x00", 16000, done=True) + await self.stop_ttfb_metrics() + await self.stop_processing_metrics() + elif isinstance(frame, StartInterruptionFrame): + await self._send_interrupt_message() + elif isinstance(frame, (EndFrame, CancelFrame)): + await self._end_conversation() + else: + await self.push_frame(frame, direction) + + 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) + + async def _send_audio_message(self, audio_base64: str, done: bool) -> None: + 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, + }, + } + ) + await self.push_frame(transport_frame)