diff --git a/CHANGELOG.md b/CHANGELOG.md index 91b70e487..d2a0d6c39 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,13 @@ All notable changes to **Pipecat** will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## Unreleased + +### Added + +- Added `RimeHttpTTSService` and the `07q-interruptible-rime.py` foundational + example. + ## [0.0.48] - 2024-11-10 "Antonio release" ### Added diff --git a/examples/foundational/07q-interruptible-rime.py b/examples/foundational/07q-interruptible-rime.py new file mode 100644 index 000000000..2e13e2a8e --- /dev/null +++ b/examples/foundational/07q-interruptible-rime.py @@ -0,0 +1,100 @@ +# +# Copyright (c) 2024, 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 runner import configure + +from pipecat.audio.vad.silero import SileroVADAnalyzer +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.openai_llm_context import OpenAILLMContext +from pipecat.services.openai import OpenAILLMService +from pipecat.services.rime import RimeHttpTTSService +from pipecat.transports.services.daily import DailyParams, DailyTransport + +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + + +async def main(): + async with aiohttp.ClientSession() as session: + (room_url, token) = await configure(session) + + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_out_enabled=True, + transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + ), + ) + + tts = RimeHttpTTSService( + api_key=os.getenv("RIME_API_KEY", ""), + voice_id="rex", + params=RimeHttpTTSService.InputParams(reduce_latency=True), + ) + + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + + 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 + context_aggregator.user(), # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # 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_first_participant_joined") + async def on_first_participant_joined(transport, participant): + await transport.capture_participant_transcription(participant["id"]) + # 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/rime.py b/src/pipecat/services/rime.py new file mode 100644 index 000000000..6cf2c8129 --- /dev/null +++ b/src/pipecat/services/rime.py @@ -0,0 +1,101 @@ +from typing import AsyncGenerator, Optional + +import aiohttp +from loguru import logger +from pydantic import BaseModel + +from pipecat.frames.frames import ( + ErrorFrame, + Frame, + TTSAudioRawFrame, + TTSStartedFrame, + TTSStoppedFrame, +) +from pipecat.services.ai_services import TTSService + + +class RimeHttpTTSService(TTSService): + class InputParams(BaseModel): + pause_between_brackets: Optional[bool] = False + phonemize_between_brackets: Optional[bool] = False + inline_speed_alpha: Optional[str] = None + speed_alpha: Optional[float] = 1.0 + reduce_latency: Optional[bool] = False + + def __init__( + self, + *, + api_key: str, + voice_id: str = "eva", + model: str = "mist", + sample_rate: int = 24000, + params: InputParams = InputParams(), + **kwargs, + ): + super().__init__(sample_rate=sample_rate, **kwargs) + + self._api_key = api_key + self._base_url = "https://users.rime.ai/v1/rime-tts" + self._settings = { + "speaker": voice_id, + "modelId": model, + "samplingRate": sample_rate, + "speedAlpha": params.speed_alpha, + "reduceLatency": params.reduce_latency, + "pauseBetweenBrackets": params.pause_between_brackets, + "phonemizeBetweenBrackets": params.phonemize_between_brackets, + } + + if params.inline_speed_alpha: + self._settings["inlineSpeedAlpha"] = params.inline_speed_alpha + + def can_generate_metrics(self) -> bool: + return True + + async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + logger.debug(f"Generating TTS: [{text}]") + + headers = { + "Accept": "audio/pcm", + "Authorization": f"Bearer {self._api_key}", + "Content-Type": "application/json", + } + + payload = self._settings.copy() + payload["text"] = text + + try: + await self.start_ttfb_metrics() + await self.start_tts_usage_metrics(text) + + yield TTSStartedFrame() + + async with aiohttp.ClientSession() as session: + async with session.post(self._base_url, json=payload, headers=headers) as response: + if response.status != 200: + error_message = f"Rime TTS error: HTTP {response.status}" + logger.error(error_message) + yield ErrorFrame(error=error_message) + return + + # Process the streaming response + chunk_size = 8192 + first_chunk = True + + async for chunk in response.content.iter_chunked(chunk_size): + if first_chunk: + await self.stop_ttfb_metrics() + first_chunk = False + + if chunk: + frame = TTSAudioRawFrame(chunk, self._settings["samplingRate"], 1) + yield frame + + yield TTSStoppedFrame() + + except Exception as e: + logger.exception(f"Error generating TTS: {e}") + yield ErrorFrame(error=f"Rime TTS error: {str(e)}") + + finally: + yield TTSStoppedFrame()