Merge pull request #1839 from pipecat-ai/mb/cartesia-speed

Add support for Cartesia's speed parameter, update clients and APIs, deprecate emotion

CHANGELOG.md

@@ -64,12 +64,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- The API version for `CartesiaTTSService` and `CartesiaHttpTTSService` has
+  been updated. Also, the `cartesia` dependency has been updated to 2.x.
+
+- `CartesiaTTSService` and `CartesiaHttpTTSService` now support Cartesia's new
+  `speed` parameter which accepts values of `slow`, `normal`, and `fast`.
+
 - `GeminiMultimodalLiveLLMService` now uses the user transcription and usage
   metrics provided by Gemini Live.
 
 - `GoogleLLMService` has been updated to use `google-genai` instead of the
   deprecated `google-generativeai`.
 
+### Deprecated
+
+- In `CartesiaTTSService` and `CartesiaHttpTTSService`, `emotion` has been
+  deprecated by Cartesia. Pipecat is following suit and deprecating `emotion`
+  as well.
+
 ### Removed
 
 - Since `GeminiMultimodalLiveLLMService` now transcribes it's own audio, the

examples/foundational/07-interruptible-cartesia-http.py

@@ -0,0 +1,105 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import argparse
+import os
+
+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 CartesiaHttpTTSService
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.openai.llm import OpenAILLMService
+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(),
+        ),
+    )
+
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    tts = CartesiaHttpTTSService(
+        api_key=os.getenv("CARTESIA_API_KEY"),
+        voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+    )
+
+    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()

pyproject.toml

@@ -44,7 +44,7 @@ assemblyai = [ "assemblyai~=0.37.0" ]
 aws = [ "boto3~=1.37.16", "websockets~=13.1" ]
 aws-nova-sonic = [ "aws_sdk_bedrock_runtime~=0.0.2" ]
 azure = [ "azure-cognitiveservices-speech~=1.42.0"]
-cartesia = [ "cartesia~=1.4.0", "websockets~=13.1" ]
+cartesia = [ "cartesia~=2.0.3", "websockets~=13.1" ]
 cerebras = []
 deepseek = []
 daily = [ "daily-python~=0.18.2" ]

src/pipecat/services/cartesia/tts.py

@@ -7,6 +7,7 @@
 import base64
 import json
 import uuid
+import warnings
 from typing import AsyncGenerator, List, Optional, Union
 
 from loguru import logger
@@ -83,7 +84,7 @@ class CartesiaTTSService(AudioContextWordTTSService):
         *,
         api_key: str,
         voice_id: str,
-        cartesia_version: str = "2024-06-10",
+        cartesia_version: str = "2025-04-16",
         url: str = "wss://api.cartesia.ai/tts/websocket",
         model: str = "sonic-2",
         sample_rate: Optional[int] = None,
@@ -151,10 +152,13 @@ class CartesiaTTSService(AudioContextWordTTSService):
         voice_config["mode"] = "id"
         voice_config["id"] = self._voice_id
 
-        if self._settings["speed"] or self._settings["emotion"]:
+        if self._settings["emotion"]:
+            warnings.warn(
+                "The 'emotion' parameter in __experimental_controls is deprecated and will be removed in a future version.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
             voice_config["__experimental_controls"] = {}
-            if self._settings["speed"]:
-                voice_config["__experimental_controls"]["speed"] = self._settings["speed"]
             if self._settings["emotion"]:
                 voice_config["__experimental_controls"]["emotion"] = self._settings["emotion"]
 
@@ -169,6 +173,10 @@ class CartesiaTTSService(AudioContextWordTTSService):
             "add_timestamps": add_timestamps,
             "use_original_timestamps": False if self.model_name == "sonic" else True,
         }
+
+        if self._settings["speed"]:
+            msg["speed"] = self._settings["speed"]
+
         return json.dumps(msg)
 
     async def start(self, frame: StartFrame):
@@ -318,6 +326,7 @@ class CartesiaHttpTTSService(TTSService):
         voice_id: str,
         model: str = "sonic-2",
         base_url: str = "https://api.cartesia.ai",
+        cartesia_version: str = "2024-11-13",
         sample_rate: Optional[int] = None,
         encoding: str = "pcm_s16le",
         container: str = "raw",
@@ -327,6 +336,8 @@ class CartesiaHttpTTSService(TTSService):
         super().__init__(sample_rate=sample_rate, **kwargs)
 
         self._api_key = api_key
+        self._base_url = base_url
+        self._cartesia_version = cartesia_version
         self._settings = {
             "output_format": {
                 "container": container,
@@ -342,7 +353,10 @@ class CartesiaHttpTTSService(TTSService):
         self.set_voice(voice_id)
         self.set_model_name(model)
 
-        self._client = AsyncCartesia(api_key=api_key, base_url=base_url)
+        self._client = AsyncCartesia(
+            api_key=api_key,
+            base_url=base_url,
+        )
 
     def can_generate_metrics(self) -> bool:
         return True
@@ -367,37 +381,63 @@ class CartesiaHttpTTSService(TTSService):
         logger.debug(f"{self}: Generating TTS [{text}]")
 
         try:
-            voice_controls = None
+            voice_config = {"mode": "id", "id": self._voice_id}
-            if self._settings["speed"] or self._settings["emotion"]:
+
-                voice_controls = {}
+            if self._settings["emotion"]:
-                if self._settings["speed"]:
+                warnings.warn(
-                    voice_controls["speed"] = self._settings["speed"]
+                    "The 'emotion' parameter in voice.__experimental_controls is deprecated and will be removed in a future version.",
-                if self._settings["emotion"]:
+                    DeprecationWarning,
-                    voice_controls["emotion"] = self._settings["emotion"]
+                    stacklevel=2,
+                )
+                voice_config["__experimental_controls"] = {"emotion": self._settings["emotion"]}
 
             await self.start_ttfb_metrics()
 
-            output = await self._client.tts.sse(
+            payload = {
-                model_id=self._model_name,
+                "model_id": self._model_name,
-                transcript=text,
+                "transcript": text,
-                voice_id=self._voice_id,
+                "voice": voice_config,
-                output_format=self._settings["output_format"],
+                "output_format": self._settings["output_format"],
-                language=self._settings["language"],
+                "language": self._settings["language"],
-                stream=False,
+            }
-                _experimental_voice_controls=voice_controls,
-            )
 
-            await self.start_tts_usage_metrics(text)
+            if self._settings["speed"]:
+                payload["speed"] = self._settings["speed"]
 
             yield TTSStartedFrame()
 
+            session = await self._client._get_session()
+
+            headers = {
+                "Cartesia-Version": self._cartesia_version,
+                "X-API-Key": self._api_key,
+                "Content-Type": "application/json",
+            }
+
+            url = f"{self._base_url}/tts/bytes"
+
+            async with session.post(url, json=payload, headers=headers) as response:
+                if response.status != 200:
+                    error_text = await response.text()
+                    logger.error(f"Cartesia API error: {error_text}")
+                    await self.push_error(ErrorFrame(f"Cartesia API error: {error_text}"))
+                    raise Exception(f"Cartesia API returned status {response.status}: {error_text}")
+
+                audio_data = await response.read()
+
+            await self.start_tts_usage_metrics(text)
+
             frame = TTSAudioRawFrame(
-                audio=output["audio"], sample_rate=self.sample_rate, num_channels=1
+                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()