Merge pull request #708 from pipecat-ai/mb/add-rime-ai

Add RimeTTSService

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

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())

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()