AI キャラクターに合う声をローカルの音声合成モデルで作る

mkdir qwen3-tts && cd qwen3-tts
python3.12 -m venv .venv && source .venv/bin/activate
pip install -U qwen-tts soundfile

import sys, time
import torch
import soundfile as sf
from qwen_tts import Qwen3TTSModel

DEVICE = "mps"
DTYPE = torch.float16  # エラーや雑音が出たら torch.float32 にする

INSTRUCT = "明るく高めの声の、アニメに出てくる元気な女の子。語尾がはずみ、親しみやすい話し方。"
REF_TEXT = "こんにちは！今日はどんなお手伝いをしましょうか？"
LINES = [
    "いらっしゃいませ！ご予約のお名前を教えてください。",
    "えっと、その件でしたら、明日の午後にはお返事できると思います。",
    "わあ、ありがとうございます！とってもうれしいです。",
]


def load(model_id):
    t = time.perf_counter()
    m = Qwen3TTSModel.from_pretrained(model_id, device_map=DEVICE, dtype=DTYPE)
    print(f"load {model_id}: {time.perf_counter() - t:.1f}s")
    return m


def report(name, wav, sr, sec):
    dur = len(wav) / sr
    print(f"{name}: 合成 {sec:.2f}s / 音声 {dur:.2f}s / 実時間比 {sec / dur:.2f}")


if sys.argv[1] == "design":
    m = load("Qwen/Qwen3-TTS-12Hz-1.7B-VoiceDesign")
    for i in range(1, 4):
        t = time.perf_counter()
        wavs, sr = m.generate_voice_design(text=REF_TEXT, language="Japanese", instruct=INSTRUCT)
        report(f"design_{i}.wav", wavs[0], sr, time.perf_counter() - t)
        sf.write(f"design_{i}.wav", wavs[0], sr)
else:
    ref = sys.argv[2]
    m = load("Qwen/Qwen3-TTS-12Hz-1.7B-Base")
    prompt = m.create_voice_clone_prompt(ref_audio=ref, ref_text=REF_TEXT)
    for i, line in enumerate(LINES, 1):
        t = time.perf_counter()
        wavs, sr = m.generate_voice_clone(text=line, language="Japanese", voice_clone_prompt=prompt)
        report(f"clone_{i}.wav", wavs[0], sr, time.perf_counter() - t)
        sf.write(f"clone_{i}.wav", wavs[0], sr)

python try_tts.py design

python try_tts.py clone design_2.wav

"""Qwen3-TTS の streaming の試し。

qwen-tts は生成が終わってから音声を返す。ここでは、言語モデルが音のトークンを
1 ステップ出すたびに横取りし、CHUNK_FRAMES 個たまったら音声に戻す。
最初の音声ができるまでの時間と、再生が途切れるかを測る。

  python stream_tts.py design_2.wav

stream_N.wav は音声をつないだもの、live_N.wav は途切れを無音で再現したもの。
"""
import functools, sys, time
import numpy as np
import torch
import soundfile as sf
from qwen_tts import Qwen3TTSModel

DEVICE = "mps"
DTYPE = torch.float16
REF_TEXT = "こんにちは！今日はどんなお手伝いをしましょうか？"
LINES = [
    "いらっしゃいませ！ご予約のお名前を教えてください。",
    "えっと、その件でしたら、明日の午後にはお返事できると思います。",
    "わあ、ありがとうございます！とってもうれしいです。",
]
CHUNK_FRAMES = 4    # 何フレームごとに音声に戻すか (12.5 フレーム = 約 1 秒)
LEFT_CONTEXT = 25   # 音声に戻すときに前に付けるフレーム数 (つなぎ目の雑音を防ぐ)


class Streamer:
    """talker の forward を包み、1 ステップごとの音のトークンを受け取って音声に戻す。"""

    def __init__(self, model, ref_code):
        self.m = model
        self.tok = model.model.speech_tokenizer
        self.eos = model.model.config.talker_config.codec_eos_token_id
        self.ref_code = ref_code  # 参照音声のトークン。最初の chunk の前の文脈に使う

    def run(self, text, prompt):
        self.frames, self.chunks, self.sent = [], [], 0
        self.done = False
        talker = self.m.model.talker
        orig = talker.forward

        # transformers は forward の引数の名前で generate の引数を検査するので、
        # wraps で元の signature を引き継ぐ
        @functools.wraps(orig)
        def hooked(*args, **kwargs):
            out = orig(*args, **kwargs)
            codes = out.hidden_states[1]
            if codes is not None and not self.done:
                if int(codes[0, 0]) == self.eos:
                    self.done = True
                    self._flush(final=True)
                else:
                    self.frames.append(codes[0].detach())
                    self._flush(final=False)
            return out

        talker.forward = hooked
        self.t0 = time.perf_counter()
        try:
            self.m.generate_voice_clone(text=text, language="Japanese", voice_clone_prompt=prompt)
        finally:
            talker.forward = orig
        if not self.done:
            self._flush(final=True)
        return self.chunks

    def _flush(self, final):
        new = len(self.frames) - self.sent
        if new <= 0 or (new < CHUNK_FRAMES and not final):
            return
        codes = torch.stack(self.frames)
        if self.ref_code is not None:
            codes = torch.cat([self.ref_code.to(codes.device), codes])
            base = len(self.ref_code)
        else:
            base = 0
        end = base + len(self.frames)
        start = base + self.sent
        ctx_start = max(0, start - LEFT_CONTEXT)
        wavs, sr = self.tok.decode([{"audio_codes": codes[ctx_start:end]}])
        wav = wavs[0]
        per_frame = len(wav) / (end - ctx_start)
        wav = wav[int(round((start - ctx_start) * per_frame)):]
        self.sr = sr
        self.chunks.append((time.perf_counter() - self.t0, wav))
        self.sent = len(self.frames)


def report(name, chunks, sr):
    # 最初の chunk ができた時点で再生を始めたとして、各 chunk が再生に間に合ったかを数える。
    # 間に合わなかった分は無音を挟み、実際に聞こえる音声を live として返す
    first = chunks[0][0]
    play_at, stalls, stall_sec = first, 0, 0.0
    live = [np.zeros(int(first * sr), dtype=np.float32)]  # 合成を始めてから最初の音声までの無音
    for t, wav in chunks:
        if t > play_at:
            stalls += 1
            stall_sec += t - play_at
            live.append(np.zeros(int((t - play_at) * sr), dtype=np.float32))
            play_at = t
        live.append(wav.astype(np.float32))
        play_at += len(wav) / sr
    total = sum(len(w) for _, w in chunks) / sr
    gen = chunks[-1][0]
    print(f"{name}: 最初の音声 {first:.2f}s / 合成 {gen:.2f}s / 音声 {total:.2f}s / "
          f"実時間比 {gen / total:.2f} / 途切れ {stalls} 回 計 {stall_sec:.2f}s")
    return np.concatenate(live)


m = Qwen3TTSModel.from_pretrained("Qwen/Qwen3-TTS-12Hz-1.7B-Base", device_map=DEVICE, dtype=DTYPE)
prompt = m.create_voice_clone_prompt(ref_audio=sys.argv[1], ref_text=REF_TEXT)
s = Streamer(m, prompt[0].ref_code)

s.run(LINES[0], prompt)  # 1 回目は GPU の準備が入るので、計測から外す
for i, line in enumerate(LINES, 1):
    chunks = s.run(line, prompt)
    live = report(f"stream_{i}.wav", chunks, s.sr)
    sf.write(f"stream_{i}.wav", np.concatenate([w for _, w in chunks]), s.sr)
    sf.write(f"live_{i}.wav", live, s.sr)  # 途切れを無音で再現した、実際に聞こえる音声