From b6f09defc999cc2b1089fff6584e951172ad9aaa Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 15 May 2025 11:12:01 -0400 Subject: [PATCH] Add MiniMax TTS --- CHANGELOG.md | 12 +- README.md | 4 +- dot-env.template | 10 +- .../foundational/07y-interruptible-minimax.py | 111 +++++++ src/pipecat/services/hailuo.py | 195 ------------ src/pipecat/services/minimax/__init__.py | 8 + src/pipecat/services/minimax/tts.py | 298 ++++++++++++++++++ 7 files changed, 435 insertions(+), 203 deletions(-) create mode 100644 examples/foundational/07y-interruptible-minimax.py delete mode 100644 src/pipecat/services/hailuo.py create mode 100644 src/pipecat/services/minimax/__init__.py create mode 100644 src/pipecat/services/minimax/tts.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 60081503c..cbbeab96a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,11 +9,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `MiniMaxHttpTTSService`, which implements MiniMax's T2A API for TTS. + Learn more: https://www.minimax.io/platform_overview + - A new function `FrameProcessor.setup()` has been added to allow setting up frame processors before receiving a `StartFrame`. This is what's happening internally: `FrameProcessor.setup()` is called, `StartFrame` is pushed from - the beginning of the pipeline, your regular pipeline operations, `EndFrame` or - `CancelFrame` are pushed from the beginning of the pipeline and finally + the beginning of the pipeline, your regular pipeline operations, `EndFrame` + or `CancelFrame` are pushed from the beginning of the pipeline and finally `FrameProcessor.cleanup()` is called. - Added support for OpenTelemetry tracing in Pipecat. This initial @@ -55,12 +58,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Other +- Added foundation example `07y-minimax-http.py` to show how to use the + `MiniMaxHttpTTSService`. + - Added an `open-telemetry-tracing` example, showing how to setup tracing. The example also includes Jaeger as an open source OpenTelemetry client to review traces from the example runs. - Added foundational example `29-turn-tracking-observer.py` to show how to use - the `TurnTrackingObserver. + the `TurnTrackingObserver`. ## [0.0.67] - 2025-05-07 diff --git a/README.md b/README.md index ec3b0a791..deea0d58f 100644 --- a/README.md +++ b/README.md @@ -50,10 +50,10 @@ You can connect to Pipecat from any platform using our official SDKs: ## 🧩 Available services | Category | Services | -|---------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| ------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | Speech-to-Text | [AssemblyAI](https://docs.pipecat.ai/server/services/stt/assemblyai), [AWS](https://docs.pipecat.ai/server/services/stt/aws), [Azure](https://docs.pipecat.ai/server/services/stt/azure), [Deepgram](https://docs.pipecat.ai/server/services/stt/deepgram), [Fal Wizper](https://docs.pipecat.ai/server/services/stt/fal), [Gladia](https://docs.pipecat.ai/server/services/stt/gladia), [Google](https://docs.pipecat.ai/server/services/stt/google), [Groq (Whisper)](https://docs.pipecat.ai/server/services/stt/groq), [OpenAI (Whisper)](https://docs.pipecat.ai/server/services/stt/openai), [Parakeet (NVIDIA)](https://docs.pipecat.ai/server/services/stt/parakeet), [Ultravox](https://docs.pipecat.ai/server/services/stt/ultravox), [Whisper](https://docs.pipecat.ai/server/services/stt/whisper) | | LLMs | [Anthropic](https://docs.pipecat.ai/server/services/llm/anthropic), [AWS](https://docs.pipecat.ai/server/services/llm/aws), [Azure](https://docs.pipecat.ai/server/services/llm/azure), [Cerebras](https://docs.pipecat.ai/server/services/llm/cerebras), [DeepSeek](https://docs.pipecat.ai/server/services/llm/deepseek), [Fireworks AI](https://docs.pipecat.ai/server/services/llm/fireworks), [Gemini](https://docs.pipecat.ai/server/services/llm/gemini), [Grok](https://docs.pipecat.ai/server/services/llm/grok), [Groq](https://docs.pipecat.ai/server/services/llm/groq), [NVIDIA NIM](https://docs.pipecat.ai/server/services/llm/nim), [Ollama](https://docs.pipecat.ai/server/services/llm/ollama), [OpenAI](https://docs.pipecat.ai/server/services/llm/openai), [OpenRouter](https://docs.pipecat.ai/server/services/llm/openrouter), [Perplexity](https://docs.pipecat.ai/server/services/llm/perplexity), [Qwen](https://docs.pipecat.ai/server/services/llm/qwen), [Together AI](https://docs.pipecat.ai/server/services/llm/together) | -| Text-to-Speech | [AWS](https://docs.pipecat.ai/server/services/tts/aws), [Azure](https://docs.pipecat.ai/server/services/tts/azure), [Cartesia](https://docs.pipecat.ai/server/services/tts/cartesia), [Deepgram](https://docs.pipecat.ai/server/services/tts/deepgram), [ElevenLabs](https://docs.pipecat.ai/server/services/tts/elevenlabs), [FastPitch (NVIDIA)](https://docs.pipecat.ai/server/services/tts/fastpitch), [Fish](https://docs.pipecat.ai/server/services/tts/fish), [Google](https://docs.pipecat.ai/server/services/tts/google), [LMNT](https://docs.pipecat.ai/server/services/tts/lmnt), [Neuphonic](https://docs.pipecat.ai/server/services/tts/neuphonic), [OpenAI](https://docs.pipecat.ai/server/services/tts/openai), [Piper](https://docs.pipecat.ai/server/services/tts/piper), [PlayHT](https://docs.pipecat.ai/server/services/tts/playht), [Rime](https://docs.pipecat.ai/server/services/tts/rime), [XTTS](https://docs.pipecat.ai/server/services/tts/xtts) | +| Text-to-Speech | [AWS](https://docs.pipecat.ai/server/services/tts/aws), [Azure](https://docs.pipecat.ai/server/services/tts/azure), [Cartesia](https://docs.pipecat.ai/server/services/tts/cartesia), [Deepgram](https://docs.pipecat.ai/server/services/tts/deepgram), [ElevenLabs](https://docs.pipecat.ai/server/services/tts/elevenlabs), [FastPitch (NVIDIA)](https://docs.pipecat.ai/server/services/tts/fastpitch), [Fish](https://docs.pipecat.ai/server/services/tts/fish), [Google](https://docs.pipecat.ai/server/services/tts/google), [LMNT](https://docs.pipecat.ai/server/services/tts/lmnt), [MiniMax](https://docs.pipecat.ai/server/services/tts/minimax), [Neuphonic](https://docs.pipecat.ai/server/services/tts/neuphonic), [OpenAI](https://docs.pipecat.ai/server/services/tts/openai), [Piper](https://docs.pipecat.ai/server/services/tts/piper), [PlayHT](https://docs.pipecat.ai/server/services/tts/playht), [Rime](https://docs.pipecat.ai/server/services/tts/rime), [XTTS](https://docs.pipecat.ai/server/services/tts/xtts) | | Speech-to-Speech | [Gemini Multimodal Live](https://docs.pipecat.ai/server/services/s2s/gemini), [OpenAI Realtime](https://docs.pipecat.ai/server/services/s2s/openai) | | Transport | [Daily (WebRTC)](https://docs.pipecat.ai/server/services/transport/daily), [FastAPI Websocket](https://docs.pipecat.ai/server/services/transport/fastapi-websocket), [SmallWebRTCTransport](https://docs.pipecat.ai/server/services/transport/small-webrtc), [WebSocket Server](https://docs.pipecat.ai/server/services/transport/websocket-server), Local | | Video | [Tavus](https://docs.pipecat.ai/server/services/video/tavus), [Simli](https://docs.pipecat.ai/server/services/video/simli) | diff --git a/dot-env.template b/dot-env.template index 8033bdfba..aa8068451 100644 --- a/dot-env.template +++ b/dot-env.template @@ -95,9 +95,13 @@ OPENROUTER_API_KEY=... PIPER_BASE_URL=... # Smart turn -LOCAL_SMART_TURN_MODEL_PATH= +LOCAL_SMART_TURN_MODEL_PATH=... FAL_SMART_TURN_API_KEY=... # Twilio -TWILIO_ACCOUNT_SID= -TWILIO_AUTH_TOKEN= \ No newline at end of file +TWILIO_ACCOUNT_SID=... +TWILIO_AUTH_TOKEN=... + +# MiniMax +MINIMAX_API_KEY=... +MINIMAX_GROUP_ID=... diff --git a/examples/foundational/07y-interruptible-minimax.py b/examples/foundational/07y-interruptible-minimax.py new file mode 100644 index 000000000..5add114bc --- /dev/null +++ b/examples/foundational/07y-interruptible-minimax.py @@ -0,0 +1,111 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import argparse +import os + +import aiohttp +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.deepgram.stt import DeepgramSTTService +from pipecat.services.minimax.tts import MiniMaxHttpTTSService +from pipecat.services.openai.llm import OpenAILLMService +from pipecat.transcriptions.language import Language +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") + + # Create an HTTP session + async with aiohttp.ClientSession() as session: + 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 = MiniMaxHttpTTSService( + api_key=os.getenv("MINIMAX_API_KEY", ""), + group_id=os.getenv("MINIMAX_GROUP_ID", ""), + aiohttp_session=session, + params=MiniMaxHttpTTSService.InputParams(language=Language.EN), + ) + + 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() diff --git a/src/pipecat/services/hailuo.py b/src/pipecat/services/hailuo.py deleted file mode 100644 index 2e4da256e..000000000 --- a/src/pipecat/services/hailuo.py +++ /dev/null @@ -1,195 +0,0 @@ -from typing import AsyncGenerator, Optional -import aiohttp -import json -from loguru import logger -from pydantic import BaseModel -import time - -from pipecat.frames.frames import ( - ErrorFrame, - Frame, - TTSAudioRawFrame, - TTSStartedFrame, - TTSStoppedFrame, -) -from pipecat.services.ai_services import TTSService -from pipecat.transcriptions.language import Language - -class HailuoHttpTTSService(TTSService): - class InputParams(BaseModel): - speed: Optional[float] = 1.0 - volume: Optional[float] = 1.0 - pitch: Optional[float] = 0 - - def __init__( - self, - *, - api_key: str, - group_id: str, - model: str = "speech-01-turbo", - voice_id: str = "Santa_Claus", - sample_rate: int = 24000, - params: InputParams = InputParams(), - **kwargs, - ): - super().__init__(sample_rate=sample_rate, **kwargs) - - self._api_key = api_key - self._group_id = group_id - self._base_url = f"https://api.minimaxi.chat/v1/t2a_v2?GroupId={group_id}" - self._session = None - - self._settings = { - "model": model, - "stream": True, - "voice_setting": { - "voice_id": voice_id, - "speed": params.speed, - "vol": params.volume, - "pitch": params.pitch - }, - "audio_setting": { - "sample_rate": sample_rate, - "bitrate": 128000, - "format": "pcm", - "channel": 1 - } - } - - def can_generate_metrics(self) -> bool: - return True - - async def _init_session(self): - if self._session is None: - self._session = aiohttp.ClientSession() - - async def _close_session(self): - if self._session: - await self._session.close() - self._session = None - - async def start(self, frame: Frame): - await super().start(frame) - await self._init_session() - - async def stop(self, frame: Frame): - await super().stop(frame) - await self._close_session() - - async def cancel(self, frame: Frame): - await super().cancel(frame) - await self._close_session() - - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - text = text.strip() - if not text or text in ['"', "'", ']', '[']: - logger.debug(f"Skipping invalid text for TTS: [{text}]") - return - - logger.debug(f"Generating TTS: [{text}]") - start_time = time.time() - - try: - await self._init_session() - await self.start_ttfb_metrics() - yield TTSStartedFrame() - - headers = { - 'accept': 'application/json, text/plain, */*', - 'Content-Type': 'application/json', - 'Authorization': f'Bearer {self._api_key}' - } - - payload = { - "model": self._settings["model"], - "text": text, - "stream": True, - "voice_setting": self._settings["voice_setting"], - "audio_setting": self._settings["audio_setting"] - } - - async with self._session.post( - self._base_url, - headers=headers, - json=payload - ) as response: - if response.status != 200: - error_text = await response.text() - raise Exception(f"TTS API error: {response.status} - {error_text}") - - await self.start_tts_usage_metrics(text) - await self.stop_ttfb_metrics() - - buffer = bytearray() - # Control initial read size - async for chunk in response.content.iter_chunked(4096): - if not chunk: - continue - - buffer.extend(chunk) - - # Find complete data blocks - while b'data:' in buffer: - start = buffer.find(b'data:') - next_start = buffer.find(b'data:', start + 5) - - if next_start == -1: - # No next data block found, keep current data for next iteration - if start > 0: - buffer = buffer[start:] - break - - # Extract a complete data block - data_block = buffer[start:next_start] - buffer = buffer[next_start:] - - try: - data = json.loads(data_block[5:].decode('utf-8')) - # Skip data blocks containing extra_info - if "extra_info" in data: - logger.debug("Received final chunk with extra info") - break - - chunk_data = data.get("data", {}) - if not chunk_data: - continue - - audio_data = chunk_data.get("audio") - if not audio_data: - continue - - # Process audio data in chunks - CHUNK_SIZE = 4096 # 4KB per chunk - for i in range(0, len(audio_data), CHUNK_SIZE * 2): # *2 for hex string - # Split hex string - hex_chunk = audio_data[i:i + CHUNK_SIZE * 2] - if not hex_chunk: - continue - - try: - # Convert this chunk of data - audio_chunk = bytes.fromhex(hex_chunk) - if audio_chunk: - yield TTSAudioRawFrame( - audio=audio_chunk, - sample_rate=self._settings["audio_setting"]["sample_rate"], - num_channels=self._settings["audio_setting"]["channel"] - ) - except ValueError as e: - logger.error(f"Error converting hex to binary: {e}") - continue - - except json.JSONDecodeError as e: - logger.error(f"Error decoding JSON: {e}, data: {data_block[:100]}") - continue - - yield TTSStoppedFrame() - - total_time = time.time() - start_time - logger.debug(f"Total TTS processing time: {total_time:.4f}s for {len(text)} chars") - - except Exception as e: - logger.error(f"{self} error generating TTS: {e}") - yield ErrorFrame(f"{self} error: {str(e)}") - finally: - await self.stop_all_metrics() diff --git a/src/pipecat/services/minimax/__init__.py b/src/pipecat/services/minimax/__init__.py new file mode 100644 index 000000000..0d444e949 --- /dev/null +++ b/src/pipecat/services/minimax/__init__.py @@ -0,0 +1,8 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + + +from .tts import * diff --git a/src/pipecat/services/minimax/tts.py b/src/pipecat/services/minimax/tts.py new file mode 100644 index 000000000..8dcbab66a --- /dev/null +++ b/src/pipecat/services/minimax/tts.py @@ -0,0 +1,298 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import json +from typing import AsyncGenerator, Optional + +import aiohttp +from loguru import logger +from pydantic import BaseModel + +from pipecat.frames.frames import ( + ErrorFrame, + Frame, + StartFrame, + TTSAudioRawFrame, + TTSStartedFrame, + TTSStoppedFrame, +) +from pipecat.services.tts_service import TTSService +from pipecat.transcriptions.language import Language +from pipecat.utils.tracing.service_decorators import traced_tts + + +def language_to_minimax_language(language: Language) -> Optional[str]: + BASE_LANGUAGES = { + Language.AR: "Arabic", + Language.CS: "Czech", + Language.DE: "German", + Language.EL: "Greek", + Language.EN: "English", + Language.ES: "Spanish", + Language.FI: "Finnish", + Language.FR: "French", + Language.HI: "Hindi", + Language.ID: "Indonesian", + Language.IT: "Italian", + Language.JA: "Japanese", + Language.KO: "Korean", + Language.NL: "Dutch", + Language.PL: "Polish", + Language.PT: "Portuguese", + Language.RO: "Romanian", + Language.RU: "Russian", + Language.TH: "Thai", + Language.TR: "Turkish", + Language.UK: "Ukrainian", + Language.VI: "Vietnamese", + Language.YUE: "Chinese,Yue", + Language.ZH: "Chinese", + } + + result = BASE_LANGUAGES.get(language) + + # If not found in base languages, try to find the base language from a variant + if not result: + # Convert enum value to string and get the base language part (e.g. es-ES -> es) + lang_str = str(language.value) + base_code = lang_str.split("-")[0].lower() + # Find matching language + for code, name in BASE_LANGUAGES.items(): + if str(code.value).lower().startswith(base_code): + result = name + break + + return result + + +class MiniMaxHttpTTSService(TTSService): + """Text-to-speech service using MiniMax's T2A (Text-to-Audio) API. + + Platform documentation: + https://www.minimax.io/platform/document/T2A%20V2?key=66719005a427f0c8a5701643 + + Args: + api_key: MiniMax API key for authentication. + group_id: MiniMax Group ID to identify project. + model: TTS model name (default: "speech-02-turbo"). Options include + "speech-02-hd", "speech-02-turbo", "speech-01-hd", "speech-01-turbo". + voice_id: Voice identifier (default: "Calm_Woman"). + aiohttp_session: aiohttp.ClientSession for API communication. + sample_rate: Output audio sample rate in Hz (default: None, set from pipeline). + params: Additional configuration parameters. + """ + + class InputParams(BaseModel): + """Configuration parameters for MiniMax TTS. + + Attributes: + language: Language for TTS generation. + speed: Speech speed (range: 0.5 to 2.0). + volume: Speech volume (range: 0 to 10). + pitch: Pitch adjustment (range: -12 to 12). + emotion: Emotional tone (options: "happy", "sad", "angry", "fearful", + "disgusted", "surprised", "neutral"). + english_normalization: Whether to apply English text normalization. + """ + + language: Optional[Language] = Language.EN + speed: Optional[float] = 1.0 + volume: Optional[float] = 1.0 + pitch: Optional[float] = 0 + emotion: Optional[str] = None + english_normalization: Optional[bool] = None + + def __init__( + self, + *, + api_key: str, + group_id: str, + model: str = "speech-02-turbo", + voice_id: str = "Calm_Woman", + aiohttp_session: aiohttp.ClientSession, + sample_rate: Optional[int] = None, + params: InputParams = InputParams(), + **kwargs, + ): + super().__init__(sample_rate=sample_rate, **kwargs) + + self._api_key = api_key + self._group_id = group_id + self._base_url = f"https://api.minimaxi.chat/v1/t2a_v2?GroupId={group_id}" + self._session = aiohttp_session + self._model_name = model + self._voice_id = voice_id + + # Create voice settings + self._settings = { + "stream": True, + "voice_setting": { + "speed": params.speed, + "vol": params.volume, + "pitch": params.pitch, + }, + "audio_setting": { + "bitrate": 128000, + "format": "pcm", + "channel": 1, + }, + } + + # Set voice and model + self.set_voice(voice_id) + self.set_model_name(model) + + # Add language boost if provided + if params.language: + service_lang = self.language_to_service_language(params.language) + if service_lang: + self._settings["language_boost"] = service_lang + + # Add optional emotion if provided + if params.emotion: + # Validate emotion is in the supported list + supported_emotions = [ + "happy", + "sad", + "angry", + "fearful", + "disgusted", + "surprised", + "neutral", + ] + if params.emotion in supported_emotions: + self._settings["voice_setting"]["emotion"] = params.emotion + else: + logger.warning(f"Unsupported emotion: {params.emotion}. Using default.") + + # Add english_normalization if provided + if params.english_normalization is not None: + self._settings["english_normalization"] = params.english_normalization + + def can_generate_metrics(self) -> bool: + return True + + def language_to_service_language(self, language: Language) -> Optional[str]: + return language_to_minimax_language(language) + + def set_model_name(self, model: str): + """Set the TTS model to use""" + self._model_name = model + + def set_voice(self, voice: str): + """Set the voice to use""" + self._voice_id = voice + if "voice_setting" in self._settings: + self._settings["voice_setting"]["voice_id"] = voice + + async def start(self, frame: StartFrame): + await super().start(frame) + self._settings["audio_setting"]["sample_rate"] = self.sample_rate + logger.debug(f"MiniMax TTS initialized with sample rate: {self.sample_rate}") + + @traced_tts + async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + logger.debug(f"{self}: Generating TTS [{text}]") + + headers = { + "accept": "application/json, text/plain, */*", + "Content-Type": "application/json", + "Authorization": f"Bearer {self._api_key}", + } + + # Create payload from settings + payload = self._settings.copy() + payload["model"] = self._model_name + payload["text"] = text + + try: + await self.start_ttfb_metrics() + + async with self._session.post( + self._base_url, headers=headers, json=payload + ) as response: + if response.status != 200: + error_message = f"MiniMax TTS error: HTTP {response.status}" + logger.error(error_message) + yield ErrorFrame(error=error_message) + return + + await self.start_tts_usage_metrics(text) + yield TTSStartedFrame() + + # Process the streaming response + buffer = bytearray() + CHUNK_SIZE = 1024 + + async for chunk in response.content.iter_chunked(CHUNK_SIZE): + if not chunk: + continue + + buffer.extend(chunk) + + # Find complete data blocks + while b"data:" in buffer: + start = buffer.find(b"data:") + next_start = buffer.find(b"data:", start + 5) + + if next_start == -1: + # No next data block found, keep current data for next iteration + if start > 0: + buffer = buffer[start:] + break + + # Extract a complete data block + data_block = buffer[start:next_start] + buffer = buffer[next_start:] + + try: + data = json.loads(data_block[5:].decode("utf-8")) + # Skip data blocks containing extra_info + if "extra_info" in data: + logger.debug("Received final chunk with extra info") + continue + + chunk_data = data.get("data", {}) + if not chunk_data: + continue + + audio_data = chunk_data.get("audio") + if not audio_data: + continue + + # Process audio data in chunks + for i in range(0, len(audio_data), CHUNK_SIZE * 2): # *2 for hex string + # Split hex string + hex_chunk = audio_data[i : i + CHUNK_SIZE * 2] + if not hex_chunk: + continue + + try: + # Convert this chunk of data + audio_chunk = bytes.fromhex(hex_chunk) + if audio_chunk: + await self.stop_ttfb_metrics() + yield TTSAudioRawFrame( + audio=audio_chunk, + sample_rate=self._settings["audio_setting"][ + "sample_rate" + ], + num_channels=self._settings["audio_setting"]["channel"], + ) + except ValueError as e: + logger.error(f"Error converting hex to binary: {e}") + continue + + except json.JSONDecodeError as e: + logger.error(f"Error decoding JSON: {e}, data: {data_block[:100]}") + continue + + except Exception as e: + logger.exception(f"Error generating TTS: {e}") + yield ErrorFrame(error=f"MiniMax TTS error: {str(e)}") + finally: + await self.stop_ttfb_metrics() + yield TTSStoppedFrame()