Qwen3-TTS に声クローン機能を追加した話 — Gradio UI・自分の声で TTS・ハマり全記録

tts_model_type: custom_voice does not support generate_voice_clone,
Please check Model Card or Readme for more details.

import gradio as gr
from fastapi import FastAPI

app = FastAPI(lifespan=lifespan)  # 既存の FastAPI アプリ

with gr.Blocks(title="Qwen3-TTS") as gradio_app:
    # UI コンポーネントを定義
    pass

# /ui に統合（ポート追加不要）
gr.mount_gradio_app(app, gradio_app, path="/ui")

http://server:9001/v1/tts/synthesize  ← REST API（従来通り）
http://server:9001/ui                 ← Gradio WebUI（新規）

# ❌ 全パスに 64KB 制限をかけると音声アップロードが詰まる
@app.middleware("http")
async def limit_request_size(request: Request, call_next):
    content_length = request.headers.get("content-length")
    if content_length and int(content_length) > 64 * 1024:
        raise HTTPException(413, "リクエストサイズが上限を超えています")
    return await call_next(request)

# ✅ Gradio のパスは除外する
@app.middleware("http")
async def limit_request_size(request: Request, call_next):
    if request.url.path.startswith("/ui"):  # ← /ui 以下は通す
        return await call_next(request)
    content_length = request.headers.get("content-length")
    if content_length and int(content_length) > 64 * 1024:
        raise HTTPException(413, "リクエストサイズが上限を超えています")
    return await call_next(request)

_model_custom = None  # CustomVoice（9話者）
_model_base   = None  # Base（声クローン）

@asynccontextmanager
async def lifespan(app: FastAPI):
    global _model_custom, _model_base
    from qwen_tts import Qwen3TTSModel

    _model_custom = Qwen3TTSModel.from_pretrained(
        "Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice",
        device_map="cuda:0", dtype=torch.bfloat16, attn_implementation="sdpa",
    )
    _model_base = Qwen3TTSModel.from_pretrained(
        "Qwen/Qwen3-TTS-12Hz-1.7B-Base",
        device_map="cuda:0", dtype=torch.bfloat16, attn_implementation="sdpa",
    )
    yield

# Gradio から受け取った音声を正規化
sr_ref, audio_data = ref_audio_from_gradio

# int16 → float32（Gradio は int16 で返すことがある）
if audio_data.dtype == np.int16:
    audio_data = audio_data.astype(np.float32) / 32768.0
# ステレオ → モノラル
if audio_data.ndim == 2:
    audio_data = audio_data.mean(axis=1)

# 声クローン実行
wavs, sr = _model_base.generate_voice_clone(
    text="読み上げたいテキスト",
    language="Japanese",
    ref_audio=(audio_data, sr_ref),
    ref_text="参照音声で話した内容の書き起こし",  # ICL モード時は必須
    x_vector_only_mode=False,
)

# ❌ エラー: 'VoiceClonePromptItem' object is not subscriptable
wavs, sr = model.generate_voice_clone(
    voice_clone_prompt=profile_item  # 単体オブジェクトをそのまま渡すとエラー
)

# ✅ リストに包む
wavs, sr = model.generate_voice_clone(
    voice_clone_prompt=[profile_item]  # 必ずリスト
)

from pathlib import Path
import torch

PROFILES_DIR = Path("~/qwen3-tts/profiles")

def save_voice_profile(name: str, ref_audio, ref_text: str):
    """参照音声からプロファイルを作成して保存"""
    audio_data, sr_ref = ref_audio

    items = _model_base.create_voice_clone_prompt(
        ref_audio=(audio_data, sr_ref),
        ref_text=ref_text or None,
        x_vector_only_mode=False,
    )

    path = PROFILES_DIR / f"{name}.pt"
    torch.save(items[0], path)  # VoiceClonePromptItem を保存

def load_profiles() -> dict:
    """起動時に全プロファイルをメモリにロード"""
    profiles = {}
    for path in PROFILES_DIR.glob("*.pt"):
        profiles[path.stem] = torch.load(
            path, map_location="cpu", weights_only=False
        )
    return profiles

# 使用時
wavs, sr = _model_base.generate_voice_clone(
    text="テキスト",
    language="Japanese",
    voice_clone_prompt=[profiles["my_voice"]],  # ← リストに包む
)

# ❌ この時点では _voice_profiles はまだ空
with gr.Blocks() as gradio_app:
    speaker_dropdown = gr.Dropdown(
        choices=list(_voice_profiles.keys()),  # 空リスト
    )
# ↓ lifespan でプロファイルをロード（Gradio 構築より後）

with gr.Blocks() as gradio_app:
    speaker_dropdown = gr.Dropdown(choices=[])

    # ページ読み込み時に最新プロファイル一覧を取得
    gradio_app.load(
        fn=lambda: gr.update(choices=list(_voice_profiles.keys())),
        outputs=[speaker_dropdown],
    )

with gr.Blocks() as gradio_app:
    pass  # ← with ブロックを閉じた後に...

# ❌ AttributeError: Cannot call load outside of a gradio.Blocks context.
gradio_app.load(fn=..., outputs=[...])

with gr.Blocks() as gradio_app:
    # ✅ with ブロックの内側に書く
    gradio_app.load(fn=..., outputs=[...])

http://server:9001/
├── /v1/tts/synthesize  ← REST API（プリセット9話者）
├── /v1/tts/clone       ← REST API（声クローン）
├── /v1/voices          ← REST API（プロファイル管理）
├── /health             ← ヘルスチェック
├── /docs               ← Swagger UI
└── /ui                 ← Gradio WebUI
    ├── 話者選択タブ    ← 9話者 + 登録済みプロファイル
    └── 声クローンタブ  ← 参照音声アップロード + プロファイル保存