Skip to content

feat: add MiniMax voice cloning to TTS #9482

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
octo-patch wants to merge 1 commit into
base: master
Choose a base branch
from
+453 −0
Open

feat: add MiniMax voice cloning to TTS #9482

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 34 additions & 0 deletions astrbot/core/config/default.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1770,6 +1770,10 @@
"api_base": "https://api.minimax.chat/v1/t2a_v2",
"minimax-group-id": "",
"model": "speech-02-turbo",
"minimax-voice-clone-audio": "",
"minimax-voice-clone-id": "",
"minimax-voice-clone-model": "speech-2.8-hd",
"minimax-voice-clone-api-base": "https://api.minimaxi.com/v1",
"minimax-langboost": "auto",
"minimax-voice-speed": 1.0,
"minimax-voice-vol": 1.0,
Expand Down Expand Up @@ -2566,6 +2570,36 @@
"description": "单一音色",
"hint": "单一音色编号, 详见官网文档",
},
"minimax-voice-clone-audio": {
"type": "string",
"description": "Voice clone audio path",
"hint": "Local .mp3, .m4a, or .wav reference audio. Leave empty to disable automatic voice cloning.",
},
"minimax-voice-clone-id": {
"type": "string",
"description": "Voice clone ID",
"hint": "Custom voice ID to create before synthesis. Required when voice clone audio is configured.",
},
"minimax-voice-clone-model": {
"type": "string",
"description": "Voice clone model",
"options": [
"speech-2.8-hd",
"speech-2.6-hd",
"speech-02-hd",
"speech-01-hd",
],
"hint": "HD speech model used to create the custom voice.",
},
"minimax-voice-clone-api-base": {
"type": "string",
"description": "Voice clone API base",
"options": [
"https://api.minimax.io/v1",
"https://api.minimaxi.com/v1",
],
"hint": "Regional MiniMax API base for file upload and voice cloning.",
},
"minimax-voice-emotion": {
"type": "string",
"description": "情绪",
Expand Down
166 changes: 166 additions & 0 deletions astrbot/core/provider/sources/minimax_tts_api_source.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
import asyncio
import json
import os
import uuid
from collections.abc import AsyncIterator
from pathlib import Path

import aiohttp

Expand Down Expand Up @@ -32,6 +34,20 @@ def __init__(
)
self.group_id: str = provider_config.get("minimax-group-id", "")
self.set_model(provider_config.get("model", ""))
self.voice_clone_audio: str = str(
provider_config.get("minimax-voice-clone-audio") or "",
).strip()
self.voice_clone_id: str = str(
provider_config.get("minimax-voice-clone-id") or "",
).strip()
self.voice_clone_model: str = str(
provider_config.get("minimax-voice-clone-model") or "",
).strip()
self.voice_clone_api_base: str = str(
provider_config.get("minimax-voice-clone-api-base") or "",
).strip()
self._voice_clone_ready = False
self._voice_clone_lock = asyncio.Lock()
self.lang_boost: str = provider_config.get("minimax-langboost", "auto")
self.is_timber_weight: bool = provider_config.get(
"minimax-is-timber-weight",
Expand Down Expand Up @@ -84,6 +100,155 @@ def __init__(
"content-type": "application/json",
}

def _voice_clone_url(self, path: str) -> str:
"""Build a MiniMax voice-cloning endpoint URL."""
if self.voice_clone_api_base:
api_base = self.voice_clone_api_base
else:
api_base = self.api_base.rstrip("/")
if not api_base.endswith("/v1"):
api_base = api_base.rsplit("/", 1)[0]
return f"{api_base.rstrip('/')}/{path.lstrip('/')}"

async def _ensure_voice_clone(self) -> None:

@sourcery-ai sourcery-ai Bot Jul 31, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

issue (complexity): Consider refactoring the new voice-cloning logic into smaller helper methods and precomputed URL state to simplify _ensure_voice_clone and related flows without changing behavior.

You can reduce the complexity notably without changing behavior by extracting a few focused helpers and moving some normalization to initialization. For example:

1. Split _ensure_voice_clone responsibilities

Right now _ensure_voice_clone mixes configuration validation, filesystem, HTTP, response parsing, and locking. You can keep locking + state in _ensure_voice_clone and move the rest into helpers:

async def _ensure_voice_clone(self) -> None:
    if not self.voice_clone_audio or self._voice_clone_ready:
        return

    async with self._voice_clone_lock:
        if self._voice_clone_ready:
            return

        self._validate_voice_clone_config()
        audio_path, content_type = self._prepare_audio_file()

        try:
            async with aiohttp.ClientSession() as session:
                file_id = await self._upload_voice_clone_audio(session, audio_path, content_type)
                voice_id = await self._create_voice_clone(session, file_id)
        except aiohttp.ClientError as exc:
            raise RuntimeError(f"MiniMax voice cloning request failed: {exc!s}") from exc

        self.voice_setting["voice_id"] = voice_id
        self._voice_clone_ready = True

Then implement small helpers with the current logic moved out:

def _validate_voice_clone_config(self) -> None:
    if not self.voice_clone_id:
        raise ValueError("MiniMax voice cloning requires 'minimax-voice-clone-id'.")
    if not self.voice_clone_model:
        raise ValueError("MiniMax voice cloning requires 'minimax-voice-clone-model'.")
    if not self.voice_clone_model.endswith("-hd"):
        raise ValueError("MiniMax voice cloning requires an HD speech model.")
    if self.is_timber_weight:
        raise ValueError("MiniMax voice cloning cannot be combined with mixed voices.")

def _prepare_audio_file(self) -> tuple[Path, str]:
    audio_path = Path(self.voice_clone_audio).expanduser()
    if not audio_path.is_file():
        raise FileNotFoundError(
            f"MiniMax voice-clone audio file does not exist: {audio_path}",
        )

    content_type = {
        ".m4a": "audio/mp4",
        ".mp3": "audio/mpeg",
        ".wav": "audio/wav",
    }.get(audio_path.suffix.lower())

    if content_type is None:
        raise ValueError(
            "MiniMax voice cloning supports only .mp3, .m4a, and .wav files.",
        )

    return audio_path, content_type
async def _upload_voice_clone_audio(
    self,
    session: aiohttp.ClientSession,
    audio_path: Path,
    content_type: str,
) -> str:
    form = aiohttp.FormData()
    with audio_path.open("rb") as audio_file:
        form.add_field(
            "file",
            audio_file,
            filename=audio_path.name,
            content_type=content_type,
        )
        form.add_field("purpose", "voice_clone")

        async with session.post(
            self._voice_clone_url("files/upload"),
            headers={"Authorization": self.headers["Authorization"]},
            data=form,
            timeout=aiohttp.ClientTimeout(total=60),
        ) as response:
            await self._ensure_ok_response(response, "MiniMax voice-clone audio upload")
            upload_data = await response.json(content_type=None)

    self._ensure_base_resp_ok(upload_data, "MiniMax voice-clone audio upload")

    file_id = self._extract_file_id(upload_data)
    if not file_id:
        raise RuntimeError(
            "MiniMax voice-clone audio upload returned no file_id.",
        )
    return file_id
async def _create_voice_clone(
    self,
    session: aiohttp.ClientSession,
    file_id: str,
) -> str:
    async with session.post(
        self._voice_clone_url("voice_clone"),
        headers=self.headers,
        json={
            "file_id": file_id,
            "voice_id": self.voice_clone_id,
            "model": self.voice_clone_model,
        },
        timeout=aiohttp.ClientTimeout(total=60),
    ) as response:
        await self._ensure_ok_response(response, "MiniMax voice cloning")
        clone_data = await response.json(content_type=None)

    self._ensure_base_resp_ok(clone_data, "MiniMax voice cloning")
    return self._extract_voice_id(clone_data)

2. Centralize response status / base_resp handling

You repeat HTTP status checks and base_resp parsing. Centralizing them keeps the main flow linear:

async def _ensure_ok_response(
    self,
    response: aiohttp.ClientResponse,
    context: str,
) -> None:
    if response.status >= 400:
        error_text = (await response.text())[:1024]
        raise RuntimeError(
            f"{context} failed: HTTP {response.status}: {error_text}",
        )

def _ensure_base_resp_ok(self, data: dict, context: str) -> None:
    base_resp = data.get("base_resp", {}) or {}
    status_code = base_resp.get("status_code")
    if status_code not in (None, 0, "0"):
        status_msg = base_resp.get("status_msg", "unknown error")
        raise RuntimeError(f"{context} failed: {status_msg}")

And small extractors for the different response shapes:

def _extract_file_id(self, data: dict) -> str | None:
    file_data = data.get("file") or {}
    nested_data = data.get("data") or {}
    return (
        data.get("file_id")
        or file_data.get("file_id")
        or nested_data.get("file_id")
    )

def _extract_voice_id(self, data: dict) -> str:
    clone_result = data.get("data") or {}
    return (
        data.get("voice_id")
        or clone_result.get("voice_id")
        or self.voice_clone_id
    )

3. Normalize URL roots in __init__ instead of per call

You can precompute the voice-clone root, which makes _voice_clone_url trivial and easier to reason about:

def __init__(...):
    ...
    api_base_raw: str = provider_config.get(
        "api_base",
        "https://api.minimax.chat/v1/t2a_v2",
    )
    self.api_base = api_base_raw

    voice_clone_api_base_raw = str(
        provider_config.get("minimax-voice-clone-api-base") or "",
    ).strip()

    root = voice_clone_api_base_raw or api_base_raw
    root = root.rstrip("/")

    # Strip t2a path if present; keep `/v1` root.
    if root.endswith("/t2a_v2"):
        root = root.rsplit("/", 1)[0]
    self._voice_clone_root = root if root.endswith("/v1") else root.rsplit("/", 1)[0]
    ...

Then:

def _voice_clone_url(self, path: str) -> str:
    return f"{self._voice_clone_root.rstrip('/')}/{path.lstrip('/')}"

These changes keep all functionality intact but make the voice-cloning flow much easier to follow, test, and maintain by separating concerns and removing nested control flow from the “hot” methods.

"""Create the configured custom voice before the first synthesis request."""
if not self.voice_clone_audio:
return
if self._voice_clone_ready:
return

async with self._voice_clone_lock:
if self._voice_clone_ready:
return
if not self.voice_clone_id:
raise ValueError(
"MiniMax voice cloning requires 'minimax-voice-clone-id'.",
)
if not self.voice_clone_model:
raise ValueError(
"MiniMax voice cloning requires 'minimax-voice-clone-model'.",
)
if not self.voice_clone_model.endswith("-hd"):
raise ValueError(
"MiniMax voice cloning requires an HD speech model.",
)
if self.is_timber_weight:
raise ValueError(
"MiniMax voice cloning cannot be combined with mixed voices.",
)

audio_path = Path(self.voice_clone_audio).expanduser()
if not audio_path.is_file():
raise FileNotFoundError(
f"MiniMax voice-clone audio file does not exist: {audio_path}",
)

content_type = {
".m4a": "audio/mp4",
".mp3": "audio/mpeg",
".wav": "audio/wav",
}.get(audio_path.suffix.lower())
if content_type is None:
raise ValueError(
"MiniMax voice cloning supports only .mp3, .m4a, and .wav files.",
)

try:
async with aiohttp.ClientSession() as session:
form = aiohttp.FormData()
with audio_path.open("rb") as audio_file:
form.add_field(
"file",
audio_file,
filename=audio_path.name,
content_type=content_type,
)
form.add_field("purpose", "voice_clone")
async with session.post(
self._voice_clone_url("files/upload"),
headers={"Authorization": self.headers["Authorization"]},
data=form,
timeout=aiohttp.ClientTimeout(total=60),
) as response:
if response.status >= 400:
error_text = (await response.text())[:1024]
raise RuntimeError(
"MiniMax voice-clone audio upload failed: "
f"HTTP {response.status}: {error_text}",
)
upload_data = await response.json(content_type=None)

upload_base_resp = upload_data.get("base_resp", {})
upload_status_code = upload_base_resp.get("status_code")
if upload_status_code not in (None, 0, "0"):
status_msg = upload_base_resp.get(
"status_msg",
"unknown error",
)
raise RuntimeError(
f"MiniMax voice-clone audio upload failed: {status_msg}",
)

file_data = upload_data.get("file") or {}
nested_data = upload_data.get("data") or {}
file_id = (
upload_data.get("file_id")
or file_data.get("file_id")
or nested_data.get("file_id")
)
if not file_id:
raise RuntimeError(
"MiniMax voice-clone audio upload returned no file_id.",
)

async with session.post(
self._voice_clone_url("voice_clone"),
headers=self.headers,
json={
"file_id": file_id,
"voice_id": self.voice_clone_id,
"model": self.voice_clone_model,
},
timeout=aiohttp.ClientTimeout(total=60),
) as response:
if response.status >= 400:
error_text = (await response.text())[:1024]
raise RuntimeError(
"MiniMax voice cloning failed: "
f"HTTP {response.status}: {error_text}",
)
clone_data = await response.json(content_type=None)

base_resp = clone_data.get("base_resp", {})
status_code = base_resp.get("status_code")
if status_code not in (None, 0, "0"):
status_msg = base_resp.get("status_msg", "unknown error")
raise RuntimeError(
f"MiniMax voice cloning failed: {status_msg}",
)

clone_result = clone_data.get("data") or {}
voice_id = (
clone_data.get("voice_id")
or clone_result.get("voice_id")
or self.voice_clone_id
)
self.voice_setting["voice_id"] = voice_id
self._voice_clone_ready = True
except aiohttp.ClientError as exc:
raise RuntimeError(
f"MiniMax voice cloning request failed: {exc!s}",
) from exc

async def clone_voice(self) -> str:
"""Create and return the configured MiniMax custom voice ID."""
await self._ensure_voice_clone()
if not self.voice_clone_audio:
raise ValueError(
"MiniMax voice cloning requires 'minimax-voice-clone-audio'.",
)
return self.voice_setting["voice_id"]

def _build_tts_stream_body(self, text: str):
"""构建流式请求体"""
dict_body: dict[str, object] = {
Expand Down Expand Up @@ -154,6 +319,7 @@ async def _audio_play(self, audio_stream: AsyncIterator[str]) -> bytes:
return b"".join(chunks)

async def get_audio(self, text: str) -> str:
await self._ensure_voice_clone()
temp_dir = get_astrbot_temp_path()
os.makedirs(temp_dir, exist_ok=True)
path = os.path.join(temp_dir, f"minimax_tts_api_{uuid.uuid4()}.wav")
Expand Down
16 changes: 16 additions & 0 deletions dashboard/src/i18n/locales/en-US/features/config-metadata.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1564,6 +1564,22 @@
"description": "Single voice",
"hint": "Single voice ID; see the official documentation."
},
"minimax-voice-clone-audio": {
"description": "Voice clone audio path",
"hint": "Local .mp3, .m4a, or .wav reference audio. Leave empty to disable automatic voice cloning."
},
"minimax-voice-clone-id": {
"description": "Voice clone ID",
"hint": "Custom voice ID to create before synthesis. Required when voice clone audio is configured."
},
"minimax-voice-clone-model": {
"description": "Voice clone model",
"hint": "HD speech model used to create the custom voice."
},
"minimax-voice-clone-api-base": {
"description": "Voice clone API base",
"hint": "Regional MiniMax API base for file upload and voice cloning."
},
"minimax-voice-emotion": {
"description": "Emotion",
"hint": "Controls emotion of synthesized speech. When set to auto, it selects emotion based on text."
Expand Down
16 changes: 16 additions & 0 deletions dashboard/src/i18n/locales/ru-RU/features/config-metadata.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1561,6 +1561,22 @@
"description": "Одиночный голос",
"hint": "ID одиночного голоса; см. официальную документацию."
},
"minimax-voice-clone-audio": {
"description": "Voice clone audio path",
"hint": "Local .mp3, .m4a, or .wav reference audio. Leave empty to disable automatic voice cloning."
},
"minimax-voice-clone-id": {
"description": "Voice clone ID",
"hint": "Custom voice ID to create before synthesis. Required when voice clone audio is configured."
},
"minimax-voice-clone-model": {
"description": "Voice clone model",
"hint": "HD speech model used to create the custom voice."
},
"minimax-voice-clone-api-base": {
"description": "Voice clone API base",
"hint": "Regional MiniMax API base for file upload and voice cloning."
},
"minimax-voice-emotion": {
"description": "Эмоция",
"hint": "Эмоция речи. 'auto' выбирает эмоцию на основе текста."
Expand Down
16 changes: 16 additions & 0 deletions dashboard/src/i18n/locales/zh-CN/features/config-metadata.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1566,6 +1566,22 @@
"description": "单一音色",
"hint": "单一音色编号, 详见官网文档"
},
"minimax-voice-clone-audio": {
"description": "Voice clone audio path",
"hint": "Local .mp3, .m4a, or .wav reference audio. Leave empty to disable automatic voice cloning."
},
"minimax-voice-clone-id": {
"description": "Voice clone ID",
"hint": "Custom voice ID to create before synthesis. Required when voice clone audio is configured."
},
"minimax-voice-clone-model": {
"description": "Voice clone model",
"hint": "HD speech model used to create the custom voice."
},
"minimax-voice-clone-api-base": {
"description": "Voice clone API base",
"hint": "Regional MiniMax API base for file upload and voice cloning."
},
"minimax-voice-emotion": {
"description": "情绪",
"hint": "控制合成语音的情绪。当为 auto 时，将根据文本内容自动选择情绪。"
Expand Down
Loading