From 083b32887e70a0eb8b4dd9367a755fd08c24dd8b Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 24 Jul 2025 01:03:44 -0400 Subject: [PATCH] NeuphonicHttpTTSService: Refactor to use POST API --- CHANGELOG.md | 4 + .../07v-interruptible-neuphonic-http.py | 94 ++++++------- pyproject.toml | 2 +- src/pipecat/services/neuphonic/tts.py | 123 ++++++++++++++---- 4 files changed, 155 insertions(+), 68 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e5c6dd18b..c27d7a596 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- Changed `NeuphonicHttpTTSService` to use a POST based request instead of the + `pyneuphonic` package. This removes a package requirement, allowing Neuphonic + to work with more services. + - Updated the `deepgram` optional dependency to 4.7.0, which downgrades the `tasks cancelled error` to a debug log. This removes the log from appearing in Pipecat logs upon leaving. diff --git a/examples/foundational/07v-interruptible-neuphonic-http.py b/examples/foundational/07v-interruptible-neuphonic-http.py index fda109c8f..7c6389a13 100644 --- a/examples/foundational/07v-interruptible-neuphonic-http.py +++ b/examples/foundational/07v-interruptible-neuphonic-http.py @@ -7,6 +7,7 @@ import argparse import os +import aiohttp from dotenv import load_dotenv from loguru import logger @@ -50,60 +51,63 @@ transport_params = { async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_sigint: bool): logger.info(f"Starting bot") - stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + # Create an HTTP session + async with aiohttp.ClientSession() as session: + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) - tts = NeuphonicHttpTTSService( - api_key=os.getenv("NEUPHONIC_API_KEY"), - voice_id="fc854436-2dac-4d21-aa69-ae17b54e98eb", # Emily - ) + tts = NeuphonicHttpTTSService( + api_key=os.getenv("NEUPHONIC_API_KEY"), + voice_id="fc854436-2dac-4d21-aa69-ae17b54e98eb", # Emily + aiohttp_session=session, + ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) + 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 + 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.", + }, ] - ) - task = PipelineTask( - pipeline, - params=PipelineParams( - enable_metrics=True, - enable_usage_metrics=True, - ), - ) + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) - @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()]) + 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 + ] + ) - @transport.event_handler("on_client_disconnected") - async def on_client_disconnected(transport, client): - logger.info(f"Client disconnected") - await task.cancel() + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_metrics=True, + enable_usage_metrics=True, + ), + ) - runner = PipelineRunner(handle_sigint=handle_sigint) + @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()]) - await runner.run(task) + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + await task.cancel() + + runner = PipelineRunner(handle_sigint=handle_sigint) + + await runner.run(task) if __name__ == "__main__": diff --git a/pyproject.toml b/pyproject.toml index c9e353e77..3c8c9b704 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -73,7 +73,7 @@ mem0 = [ "mem0ai~=0.1.94" ] mlx-whisper = [ "mlx-whisper~=0.4.2" ] moondream = [ "einops~=0.8.0", "timm~=1.0.13", "transformers>=4.48.0" ] nim = [] -neuphonic = [ "pyneuphonic~=1.5.13", "websockets>=13.1,<15.0" ] +neuphonic = [ "websockets>=13.1,<15.0" ] noisereduce = [ "noisereduce~=3.0.3" ] openai = [ "websockets>=13.1,<15.0" ] openpipe = [ "openpipe~=4.50.0" ] diff --git a/src/pipecat/services/neuphonic/tts.py b/src/pipecat/services/neuphonic/tts.py index 9472a4197..7cf186571 100644 --- a/src/pipecat/services/neuphonic/tts.py +++ b/src/pipecat/services/neuphonic/tts.py @@ -15,6 +15,7 @@ import base64 import json from typing import Any, AsyncGenerator, Mapping, Optional +import aiohttp from loguru import logger from pydantic import BaseModel @@ -39,7 +40,6 @@ from pipecat.utils.asyncio.watchdog_async_iterator import WatchdogAsyncIterator from pipecat.utils.tracing.service_decorators import traced_tts try: - from pyneuphonic import Neuphonic, TTSConfig from websockets.asyncio.client import connect as websocket_connect from websockets.protocol import State except ModuleNotFoundError as e: @@ -407,9 +407,10 @@ class NeuphonicHttpTTSService(TTSService): *, api_key: str, voice_id: Optional[str] = None, + aiohttp_session: aiohttp.ClientSession, url: str = "https://api.neuphonic.com", sample_rate: Optional[int] = 22050, - encoding: str = "pcm_linear", + encoding: Optional[str] = "pcm_linear", params: Optional[InputParams] = None, **kwargs, ): @@ -418,6 +419,7 @@ class NeuphonicHttpTTSService(TTSService): Args: api_key: Neuphonic API key for authentication. voice_id: ID of the voice to use for synthesis. + aiohttp_session: Shared aiohttp session for HTTP requests. url: Base URL for the Neuphonic HTTP API. sample_rate: Audio sample rate in Hz. Defaults to 22050. encoding: Audio encoding format. Defaults to "pcm_linear". @@ -429,13 +431,11 @@ class NeuphonicHttpTTSService(TTSService): params = params or NeuphonicHttpTTSService.InputParams() self._api_key = api_key - self._url = url - self._settings = { - "lang_code": self.language_to_service_language(params.language), - "speed": params.speed, - "encoding": encoding, - "sampling_rate": sample_rate, - } + self._session = aiohttp_session + self._base_url = url.rstrip("/") + self._lang_code = self.language_to_service_language(params.language) or "en" + self._speed = params.speed + self._encoding = encoding self.set_voice(voice_id) def can_generate_metrics(self) -> bool: @@ -473,6 +473,40 @@ class NeuphonicHttpTTSService(TTSService): """ pass + def _parse_sse_message(self, message: str) -> dict | None: + """Parse a Server-Sent Event message. + + Args: + message: The SSE message to parse. + + Returns: + Parsed message dictionary or None if not a data message. + """ + message = message.strip() + + if not message or "data" not in message: + return None + + try: + # Split on ": " and take the part after "data: " + _, data_content = message.split(": ", 1) + + if not data_content or data_content == "[DONE]": + return None + + message_dict = json.loads(data_content) + + # Check for errors in the response + if message_dict.get("errors") is not None: + raise Exception( + f"Neuphonic API error {message_dict.get('status_code', 'unknown')}: {message_dict['errors']}" + ) + + return message_dict + except (ValueError, json.JSONDecodeError) as e: + logger.warning(f"Failed to parse SSE message: {e}") + return None + @traced_tts async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Neuphonic streaming API. @@ -485,26 +519,71 @@ class NeuphonicHttpTTSService(TTSService): """ logger.debug(f"Generating TTS: [{text}]") - client = Neuphonic(api_key=self._api_key, base_url=self._url.replace("https://", "")) + url = f"{self._base_url}/sse/speak/{self._lang_code}" - sse = client.tts.AsyncSSEClient() + headers = { + "X-API-KEY": self._api_key, + "Content-Type": "application/json", + } + + payload = { + "text": text, + "lang_code": self._lang_code, + "encoding": self._encoding, + "sampling_rate": self.sample_rate, + "speed": self._speed, + } + + if self._voice_id: + payload["voice_id"] = self._voice_id try: await self.start_ttfb_metrics() - response = sse.send(text, TTSConfig(**self._settings, voice_id=self._voice_id)) - await self.start_tts_usage_metrics(text) - yield TTSStartedFrame() + async with self._session.post(url, json=payload, headers=headers) as response: + if response.status != 200: + error_text = await response.text() + error_message = f"Neuphonic API error: HTTP {response.status} - {error_text}" + logger.error(error_message) + yield ErrorFrame(error=error_message) + return - async for message in response: - if message.status_code != 200: - logger.error(f"{self} error: {message.errors}") - yield ErrorFrame(error=f"Neuphonic API error: {message.errors}") + await self.start_tts_usage_metrics(text) + yield TTSStartedFrame() - await self.stop_ttfb_metrics() - yield TTSAudioRawFrame(message.data.audio, self.sample_rate, 1) + # Process SSE stream line by line + async for line in response.content: + if not line: + continue + + message = line.decode("utf-8", errors="ignore") + if not message.strip(): + continue + + try: + parsed_message = self._parse_sse_message(message) + + if ( + parsed_message is not None + and parsed_message.get("data", {}).get("audio") is not None + ): + audio_b64 = parsed_message["data"]["audio"] + audio_bytes = base64.b64decode(audio_b64) + + await self.stop_ttfb_metrics() + yield TTSAudioRawFrame(audio_bytes, self.sample_rate, 1) + + except Exception as e: + logger.error(f"Error processing SSE message: {e}") + # Don't yield error frame for individual message failures + continue + + except asyncio.CancelledError: + logger.debug("TTS generation cancelled") + raise except Exception as e: - logger.error(f"Error in run_tts: {e}") - yield ErrorFrame(error=str(e)) + logger.exception(f"Error in run_tts: {e}") + yield ErrorFrame(error=f"Neuphonic TTS error: {str(e)}") finally: + await self.stop_ttfb_metrics() yield TTSStoppedFrame()