feat: add MiniMax voice cloning to TTS - #9482
Open
octo-patch wants to merge 1 commit into
Open
Conversation
dosubot
Bot
added
size:L
📄 Knowledge reviewDosu skipped reviewing this PR because your organization has used its
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - I've found 2 issues
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="tests/test_minimax_tts_api_source.py" line_range="153-105" />
<code_context>
+
+
+@pytest.mark.asyncio
+async def test_clone_voice_is_single_flight(monkeypatch, tmp_path):
+ audio_path = tmp_path / "reference.wav"
+ audio_path.write_bytes(b"wav data")
+ provider = _make_provider(audio_path)
+ session = _FakeSession([_FakeResponse({"file_id": "file-123"}), _FakeResponse({})])
+ monkeypatch.setattr(
+ "astrbot.core.provider.sources.minimax_tts_api_source.aiohttp.ClientSession",
+ lambda: session,
+ )
+ monkeypatch.setattr(
+ "astrbot.core.provider.sources.minimax_tts_api_source.aiohttp.FormData",
+ _FakeFormData,
+ )
+
+ assert await asyncio.gather(provider.clone_voice(), provider.clone_voice()) == [
+ "custom_voice",
+ "custom_voice",
+ ]
+ assert len(session.posts) == 2
+
+
</code_context>
<issue_to_address>
**suggestion (testing):** Consider adding a test for `clone_voice` when no audio path is configured
The existing test only covers concurrent cloning when an audio path is configured. There’s also the case where `minimax-voice-clone-audio` is unset and `clone_voice` is expected to raise `ValueError` (since `_ensure_voice_clone` no-ops). Please add a test that constructs a provider without `minimax-voice-clone-audio` and asserts that `clone_voice()` raises the correct `ValueError` and message.
Suggested implementation:
```python
assert [post[0] for post in session.posts] == [
f"{api_base}/files/upload",
f"{api_base}/voice_clone",
]
@pytest.mark.asyncio
async def test_clone_voice_without_audio_path_raises_value_error():
# Construct a provider without the `minimax-voice-clone-audio` setting.
# This relies on `_make_provider` not configuring the audio path when given `None`.
provider = _make_provider(audio_path=None)
with pytest.raises(ValueError) as exc_info:
await provider.clone_voice()
# The error message should make it clear that `minimax-voice-clone-audio`
# needs to be configured for `clone_voice` to work.
assert "minimax-voice-clone-audio" in str(exc_info.value)
@pytest.mark.asyncio
import pytest
from astrbot.core.provider.sources.minimax_tts_api_source import (
ProviderMiniMaxTTSAPI,
)
class _FakeResponse:
```
Depending on how `_make_provider` is currently implemented, you may need to:
1. Update `_make_provider` to treat `audio_path=None` as “no audio configured” and avoid setting the `minimax-voice-clone-audio` key in the provider configuration when `audio_path` is `None`.
2. If `_make_provider` cannot accept `None`, instead construct the provider explicitly in the new test (e.g. `ProviderMiniMaxTTSAPI(config=...)`) using the same required config keys as other tests, but omitting `minimax-voice-clone-audio`.
3. If the `ValueError` message is more specific (e.g. `"minimax-voice-clone-audio must be configured"`), you can tighten the assertion to match the exact message rather than just checking that it contains `minimax-voice-clone-audio`.
</issue_to_address>
### Comment 2
<location path="astrbot/core/provider/sources/minimax_tts_api_source.py" line_range="113" />
<code_context>
+ api_base = api_base.rsplit("/", 1)[0]
+ return f"{api_base.rstrip('/')}/{path.lstrip('/')}"
+
+ async def _ensure_voice_clone(self) -> None:
+ """Create the configured custom voice before the first synthesis request."""
+ if not self.voice_clone_audio:
</code_context>
<issue_to_address>
**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:
```python
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:
```python
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
```
```python
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
```
```python
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:
```python
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:
```python
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:
```python
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:
```python
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.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| ] | ||
| == "created-voice" | ||
| ) | ||
| assert len(session.posts) == 2 |
Contributor
There was a problem hiding this comment.
suggestion (testing): Consider adding a test for clone_voice when no audio path is configured
The existing test only covers concurrent cloning when an audio path is configured. There’s also the case where minimax-voice-clone-audio is unset and clone_voice is expected to raise ValueError (since _ensure_voice_clone no-ops). Please add a test that constructs a provider without minimax-voice-clone-audio and asserts that clone_voice() raises the correct ValueError and message.
Suggested implementation:
assert [post[0] for post in session.posts] == [
f"{api_base}/files/upload",
f"{api_base}/voice_clone",
]
@pytest.mark.asyncio
async def test_clone_voice_without_audio_path_raises_value_error():
# Construct a provider without the `minimax-voice-clone-audio` setting.
# This relies on `_make_provider` not configuring the audio path when given `None`.
provider = _make_provider(audio_path=None)
with pytest.raises(ValueError) as exc_info:
await provider.clone_voice()
# The error message should make it clear that `minimax-voice-clone-audio`
# needs to be configured for `clone_voice` to work.
assert "minimax-voice-clone-audio" in str(exc_info.value)
@pytest.mark.asyncio
import pytest
from astrbot.core.provider.sources.minimax_tts_api_source import (
ProviderMiniMaxTTSAPI,
)
class _FakeResponse:Depending on how _make_provider is currently implemented, you may need to:
- Update
_make_providerto treataudio_path=Noneas “no audio configured” and avoid setting theminimax-voice-clone-audiokey in the provider configuration whenaudio_pathisNone. - If
_make_providercannot acceptNone, instead construct the provider explicitly in the new test (e.g.ProviderMiniMaxTTSAPI(config=...)) using the same required config keys as other tests, but omittingminimax-voice-clone-audio. - If the
ValueErrormessage is more specific (e.g."minimax-voice-clone-audio must be configured"), you can tighten the assertion to match the exact message rather than just checking that it containsminimax-voice-clone-audio.
| api_base = api_base.rsplit("/", 1)[0] | ||
| return f"{api_base.rstrip('/')}/{path.lstrip('/')}" | ||
|
|
||
| async def _ensure_voice_clone(self) -> None: |
Contributor
There was a problem hiding this comment.
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 = TrueThen 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_typeasync 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_idasync 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.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Reason: MiniMax TTS lacks the configured voice cloning workflow for creating custom voice IDs from reference audio.
This change:
voice_clonepurpose;Checks:
uv run pytest -q tests/test_minimax_tts_api_source.py(6 passed)uv run ruff check .uv run ruff format --check .git diff --checkSummary by Sourcery
Integrate configurable MiniMax voice cloning into the TTS provider and ensure it runs once before synthesis using the cloned voice ID.
New Features:
Enhancements:
Documentation:
Tests: