From d8cf15e7c3f5172111f8ff3ff0f362cc919b0669 Mon Sep 17 00:00:00 2001 From: octo-patch <266937838+octo-patch@users.noreply.github.com> Date: Wed, 12 Aug 2026 18:58:27 +0800 Subject: [PATCH] feat(tools): add MiniMax voice cloning --- raven/agent/loop/main.py | 2 + raven/agent/tools/media_gen.py | 205 ++++++++++++++++++++ raven/config/schema.py | 8 + tests/test_media_gen_voice_clone_minimax.py | 147 ++++++++++++++ 4 files changed, 362 insertions(+) create mode 100644 tests/test_media_gen_voice_clone_minimax.py diff --git a/raven/agent/loop/main.py b/raven/agent/loop/main.py index f6d76613..c554f1fd 100644 --- a/raven/agent/loop/main.py +++ b/raven/agent/loop/main.py @@ -35,6 +35,7 @@ ImageGenerateTool, SpeechGenerateTool, VideoGenerateTool, + VoiceCloneTool, ) from raven.agent.tools.message import MessageTool from raven.agent.tools.registry import ToolRegistry @@ -790,6 +791,7 @@ def _register_default_tools(self) -> None: media_tools = ( (ImageGenerateTool, media.image), (SpeechGenerateTool, media.speech), + (VoiceCloneTool, media.voice_clone), (VideoGenerateTool, media.video), ) for cls, tool_cfg in media_tools: diff --git a/raven/agent/tools/media_gen.py b/raven/agent/tools/media_gen.py index 6f64baeb..3fc71da6 100644 --- a/raven/agent/tools/media_gen.py +++ b/raven/agent/tools/media_gen.py @@ -54,6 +54,21 @@ from raven.config.schema import MediaToolConfig _DEFAULT_BASE = "https://openrouter.ai/api/v1" +_MINIMAX_VOICE_CLONE_ENDPOINTS = { + "global_en": "https://api.minimax.io/v1/voice_clone", + "cn_zh": "https://api.minimaxi.com/v1/voice_clone", +} +_MINIMAX_VOICE_CLONE_DEFAULT_ENDPOINT = _MINIMAX_VOICE_CLONE_ENDPOINTS["global_en"] +_MINIMAX_VOICE_CLONE_MODELS = frozenset( + { + "speech-2.8-hd", + "speech-2.6-hd", + "speech-02-hd", + "speech-01-hd", + } +) +_MINIMAX_VOICE_CLONE_DEFAULT_MODEL = "speech-2.8-hd" +_MINIMAX_VOICE_CLONE_AUDIO_FORMATS = frozenset({"mp3", "m4a", "wav"}) _EXT_MIME = { ".png": "image/png", @@ -422,6 +437,196 @@ async def _maybe_transcode(self, wav_path: Path, fmt: str) -> tuple[Path, str, s return out, fmt, "" +class VoiceCloneTool(_OpenRouterMediaTool): + """Create a reusable MiniMax voice from a local audio sample.""" + + name = "voice_clone" + default_model = _MINIMAX_VOICE_CLONE_DEFAULT_MODEL + description = ( + "Clone a voice from a local MP3, M4A, or WAV sample. Optionally upload a " + "second short prompt sample to improve similarity." + ) + parameters = { + "type": "object", + "properties": { + "audio_path": {"type": "string", "description": "Local path to the voice sample"}, + "voice_id": {"type": "string", "description": "Unique identifier to assign to the cloned voice"}, + "model": { + "type": "string", + "enum": sorted(_MINIMAX_VOICE_CLONE_MODELS), + "default": _MINIMAX_VOICE_CLONE_DEFAULT_MODEL, + "description": "Speech model used to validate the cloned voice", + }, + "prompt_audio_path": { + "type": "string", + "description": "Optional local path to a prompt sample shorter than eight seconds", + }, + "prompt_text": { + "type": "string", + "description": "Transcript of prompt_audio_path, including ending punctuation", + }, + "text": {"type": "string", "description": "Optional text for a preview audio sample"}, + "text_validation": { + "type": "string", + "description": "Optional expected transcript for validating the source audio", + }, + "accuracy": { + "type": "number", + "minimum": 0, + "maximum": 1, + "default": 0.7, + "description": "Minimum transcript similarity when text_validation is provided", + }, + "need_noise_reduction": {"type": "boolean", "default": False}, + "need_volume_normalization": {"type": "boolean", "default": False}, + "aigc_watermark": {"type": "boolean", "default": False}, + }, + "required": ["audio_path", "voice_id"], + } + + @property + def _minimax_api_key(self) -> str: + cfg_key = getattr(self._config, "api_key", "") if self._config else "" + return cfg_key or os.environ.get("MINIMAX_API_KEY", "") + + @property + def _clone_endpoint(self) -> str: + cfg_base = getattr(self._config, "api_base", "") if self._config else "" + if not cfg_base: + return _MINIMAX_VOICE_CLONE_DEFAULT_ENDPOINT + base = cfg_base.rstrip("/") + return base if base.endswith("/voice_clone") else f"{base}/voice_clone" + + @property + def _upload_endpoint(self) -> str: + return f"{self._clone_endpoint.rsplit('/voice_clone', 1)[0]}/files/upload" + + def _no_minimax_key_error(self) -> str: + return json.dumps( + { + "error": ( + "voice_clone: no API key configured. Set it under " + "tools.media.voiceClone.apiKey or providers.minimax.apiKey, " + "or export MINIMAX_API_KEY, then restart the gateway." + ) + }, + ensure_ascii=False, + ) + + @staticmethod + def _response_error(response: dict[str, Any]) -> str: + base_resp = response.get("base_resp") or {} + status_code = base_resp.get("status_code") + if status_code in (None, 0): + return "" + return str(base_resp.get("status_msg") or f"status_code={status_code}") + + async def _upload_audio( + self, + client: httpx.AsyncClient, + path: Path, + purpose: str, + headers: dict[str, str], + ) -> int | str: + suffix = path.suffix.lower().lstrip(".") + if suffix not in _MINIMAX_VOICE_CLONE_AUDIO_FORMATS: + supported = ", ".join(sorted(_MINIMAX_VOICE_CLONE_AUDIO_FORMATS)) + raise ValueError(f"unsupported audio format: {suffix or 'unknown'}; expected one of {supported}") + response = await client.post( + self._upload_endpoint, + headers=headers, + data={"purpose": purpose}, + files={"file": (path.name, path.read_bytes(), "application/octet-stream")}, + ) + response.raise_for_status() + payload = response.json() + if error := self._response_error(payload): + raise ValueError(error) + file_id = (payload.get("file") or {}).get("file_id") + if file_id in (None, ""): + raise ValueError("file upload did not return file.file_id") + return file_id + + async def execute( + self, + audio_path: str, + voice_id: str, + model: str | None = None, + prompt_audio_path: str | None = None, + prompt_text: str | None = None, + text: str | None = None, + text_validation: str | None = None, + accuracy: float = 0.7, + need_noise_reduction: bool = False, + need_volume_normalization: bool = False, + aigc_watermark: bool = False, + **kwargs: Any, + ) -> str: + if not self._minimax_api_key: + return self._no_minimax_key_error() + if bool(prompt_audio_path) != bool(prompt_text): + return json.dumps( + {"error": "prompt_audio_path and prompt_text must be provided together"}, + ensure_ascii=False, + ) + + model_id = self._model(model) + if model_id not in _MINIMAX_VOICE_CLONE_MODELS: + return json.dumps({"error": f"unsupported voice cloning model: {model_id}"}, ensure_ascii=False) + + source = Path(audio_path).expanduser() + prompt_source = Path(prompt_audio_path).expanduser() if prompt_audio_path else None + for path in (source, prompt_source): + if path is not None and not path.is_file(): + return json.dumps({"error": f"audio file not found: {path}"}, ensure_ascii=False) + + headers = {"Authorization": f"Bearer {self._minimax_api_key}"} + try: + async with httpx.AsyncClient(proxy=self._proxy, timeout=180.0) as client: + file_id = await self._upload_audio(client, source, "voice_clone", headers) + payload: dict[str, Any] = { + "file_id": file_id, + "voice_id": voice_id, + "model": model_id, + "accuracy": accuracy, + "need_noise_reduction": need_noise_reduction, + "need_volume_normalization": need_volume_normalization, + "aigc_watermark": aigc_watermark, + } + if prompt_source is not None: + prompt_file_id = await self._upload_audio(client, prompt_source, "prompt_audio", headers) + payload["clone_prompt"] = { + "prompt_audio": prompt_file_id, + "prompt_text": prompt_text, + } + if text is not None: + payload["text"] = text + if text_validation is not None: + payload["text_validation"] = text_validation + + response = await client.post( + self._clone_endpoint, + headers={**headers, "Content-Type": "application/json"}, + json=payload, + ) + response.raise_for_status() + result = response.json() + if error := self._response_error(result): + return json.dumps({"error": error, "voice_id": voice_id}, ensure_ascii=False) + except httpx.HTTPStatusError as e: + return self._format_http_error(e) + except Exception as e: + logger.error("voice_clone error: {}", e) + return json.dumps({"error": str(e)}, ensure_ascii=False) + + result_voice_id = result.get("voice_id") or voice_id + logger.info("voice_clone: created {} via {}", result_voice_id, model_id) + return json.dumps( + {"success": True, "voice_id": result_voice_id, "model": model_id}, + ensure_ascii=False, + ) + + def _find_video_url(obj: Any) -> str | None: """Recursively find the first downloadable video URL / data URI in a response. diff --git a/raven/config/schema.py b/raven/config/schema.py index 5c53414e..6cbbeea2 100644 --- a/raven/config/schema.py +++ b/raven/config/schema.py @@ -754,6 +754,7 @@ class MediaGenConfig(Base): image: MediaToolConfig = Field(default_factory=MediaToolConfig) speech: MediaToolConfig = Field(default_factory=MediaToolConfig) + voice_clone: MediaToolConfig = Field(default_factory=MediaToolConfig) video: MediaToolConfig = Field(default_factory=MediaToolConfig) proxy: str | None = None # HTTP/SOCKS proxy for media API calls output_subdir: str = "generated" # where generated files are written under workspace @@ -864,6 +865,13 @@ def effective_media_config(self) -> MediaGenConfig: configured = bool(tool.api_key or tool.model) if configured and or_key and not tool.api_key: tool.api_key = or_key + + voice_clone = media.voice_clone + voice_clone_configured = bool(voice_clone.api_key or voice_clone.api_base or voice_clone.model) + if voice_clone_configured and not voice_clone.api_key: + provider = self.providers.get("minimax") + if provider and provider.api_key: + voice_clone.api_key = provider.api_key return media def _match_provider(self, model: str | None = None) -> tuple["ProviderConfig | None", str | None]: diff --git a/tests/test_media_gen_voice_clone_minimax.py b/tests/test_media_gen_voice_clone_minimax.py new file mode 100644 index 00000000..921577c3 --- /dev/null +++ b/tests/test_media_gen_voice_clone_minimax.py @@ -0,0 +1,147 @@ +"""Unit tests for MiniMax voice cloning.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import httpx +import pytest + +from raven.agent.tools import media_gen as mg +from raven.agent.tools.media_gen import VoiceCloneTool +from raven.config.schema import Config, MediaToolConfig + + +def _patch_client(monkeypatch: pytest.MonkeyPatch, handler) -> None: + real_client = httpx.AsyncClient + + def factory(*args, **kwargs): + return real_client(transport=httpx.MockTransport(handler), **kwargs) + + monkeypatch.setattr(mg.httpx, "AsyncClient", factory) + + +async def test_voice_clone_uploads_source_and_creates_voice(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + source = tmp_path / "source.wav" + source.write_bytes(b"source audio") + requests: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + assert request.headers["Authorization"] == "Bearer unit-key" + if request.url.path == "/v1/files/upload": + assert request.url.host == "api.minimax.io" + assert b'form-data; name="purpose"' in request.content + assert b"voice_clone" in request.content + assert b"source audio" in request.content + return httpx.Response( + 200, + json={"file": {"file_id": 101, "purpose": "voice_clone"}, "base_resp": {"status_code": 0}}, + ) + + assert request.url == httpx.URL("https://api.minimax.io/v1/voice_clone") + assert json.loads(request.content) == { + "file_id": 101, + "voice_id": "RavenVoice001", + "model": "speech-2.8-hd", + "accuracy": 0.8, + "need_noise_reduction": True, + "need_volume_normalization": True, + "aigc_watermark": False, + "text": "Preview this voice.", + "text_validation": "Source transcript.", + } + return httpx.Response(200, json={"voice_id": "RavenVoice001", "base_resp": {"status_code": 0}}) + + _patch_client(monkeypatch, handler) + tool = VoiceCloneTool(MediaToolConfig(api_key="unit-key"), workspace=tmp_path) + + result = json.loads( + await tool.execute( + audio_path=str(source), + voice_id="RavenVoice001", + model="speech-2.8-hd", + accuracy=0.8, + need_noise_reduction=True, + need_volume_normalization=True, + text="Preview this voice.", + text_validation="Source transcript.", + ) + ) + + assert result == {"success": True, "voice_id": "RavenVoice001", "model": "speech-2.8-hd"} + assert len(requests) == 2 + + +async def test_voice_clone_uses_cn_endpoint_and_prompt_audio(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + source = tmp_path / "source.mp3" + prompt = tmp_path / "prompt.m4a" + source.write_bytes(b"source audio") + prompt.write_bytes(b"prompt audio") + purposes: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + assert request.url.host == "api.minimaxi.com" + if request.url.path == "/v1/files/upload": + purpose = "prompt_audio" if b"prompt_audio" in request.content else "voice_clone" + purposes.append(purpose) + file_id = 202 if purpose == "prompt_audio" else 101 + return httpx.Response( + 200, + json={"file": {"file_id": file_id, "purpose": purpose}, "base_resp": {"status_code": 0}}, + ) + + assert request.url.path == "/v1/voice_clone" + body = json.loads(request.content) + assert body["file_id"] == 101 + assert body["clone_prompt"] == {"prompt_audio": 202, "prompt_text": "A short prompt."} + return httpx.Response(200, json={"base_resp": {"status_code": 0}}) + + _patch_client(monkeypatch, handler) + tool = VoiceCloneTool( + MediaToolConfig(api_key="unit-key", api_base="https://api.minimaxi.com/v1"), + workspace=tmp_path, + ) + + result = json.loads( + await tool.execute( + audio_path=str(source), + voice_id="RavenVoice002", + model="speech-2.6-hd", + prompt_audio_path=str(prompt), + prompt_text="A short prompt.", + ) + ) + + assert result["success"] is True + assert result["voice_id"] == "RavenVoice002" + assert purposes == ["voice_clone", "prompt_audio"] + + +async def test_voice_clone_returns_api_error(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + source = tmp_path / "source.wav" + source.write_bytes(b"source audio") + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"base_resp": {"status_code": 1004, "status_msg": "Authentication failed"}}) + + _patch_client(monkeypatch, handler) + tool = VoiceCloneTool(MediaToolConfig(api_key="unit-key"), workspace=tmp_path) + + result = json.loads(await tool.execute(audio_path=str(source), voice_id="RavenVoice003")) + + assert result["error"] == "Authentication failed" + + +def test_voice_clone_inherits_minimax_provider_key() -> None: + config = Config.model_validate( + { + "providers": {"minimax": {"apiKey": "unit-key"}}, + "tools": {"media": {"voiceClone": {"model": "speech-2.8-hd"}}}, + } + ) + + resolved = config.effective_media_config().voice_clone + + assert resolved.api_key == "unit-key"