Written with AI assistance from the source video. The video is the authority.
7:58Better Stack
Log in to leave a comment
No posts yet
月50万ウォン未満の予算で個人SaaSを運営する場合、有料TTS APIの請求書は毎月の負担となります。ネットワーク遅延時間も問題です。画面のボタンを押してから音声が流れるまでに1〜2秒かかると、ユーザーはすぐにタブを閉じてしまいます。
9,900万個のパラメータを持つオープンソースモデル「Supertonic 3」は魅力的な代替手段です。ローカルサーバーで直接動かせば、API費用を0円に抑えることができます。
ただし、Pythonのサンプルコードを数行動かしてみることと、実際のプロダクションデプロイは全く別物です。低スペックサーバーでこのモデルを立ち上げる瞬間に直面するメモリボトルネックと非同期処理の問題を、自ら解決した方法をまとめました。
Supertonic 3のONNX重みファイルのサイズは約305MBです。モデルを最初にメモリに読み込んだ際、常駐メモリ（RSS）は350MB前後を維持します。
問題は、ユーザーからリクエストが送信され、44.1kHzのオーディオテンソルの演算を開始するときに発生します。瞬間的なピークメモリが900MBを超えます。1 vCPU / 1GB RAM仕様の最安値インスタンスを使用すると、LinuxのOOM Killerが作動してPythonプロセスを即座に強制終了します。
安定した運用のための最低ラインは、2 vCPU / 2GB RAMインスタンスです。
| サーバーインスタンス仕様 | アイドル時RAM | 演算ピークRAM | 平均CPU使用率 | リアルタイムファクター (RTF) | 月間予想コスト削減額 |
|---|---|---|---|---|---|
| 1 vCPU / 1GB RAM | 280 MB | 890 MB (強制終了のリスク) | 98% | 0.85 (1秒の音声生成に0.85秒要す) | $130 (商用API比較) |
| 2 vCPU / 2GB RAM | 320 MB | 920 MB (安全圏) | 48% (スレッド制限時) | 0.28 (1秒の音声生成に0.28秒要す) | $120 (GPUインスタンス比較) |
| 4 vCPU / 4GB RAM | 350 MB | 950 MB | 25% | 0.15 | $90 (過剰割り当てインスタンスの調整) |
低スペックCPUサーバーで複数のリクエストが流入した際にCPU使用率が100%に急騰する現象を防ぐには、ONNXランタイムのスレッドプールを手動で制御する必要があります。
intra_op_num_threadsをサーバーの物理コア数（2個）に合わせます。execution_modeをORT_SEQUENTIALにし、inter_op_num_threadsを1に指定して不要なコンテキストスイッチングを防ぎます。enable_cpu_mem_arenaを有効にして頻繁なヒープメモリの再割り当てを防ぎ、allow_spinningを0に設定してCPUが空ループを回して待機する現象をオフにします。`python
import onnxruntime as ort
def get_optimized_session_options(cpu_cores: int = 2) -> ort.SessionOptions:
options = ort.SessionOptions()
options.intra_op_num_threads = cpu_cores
options.execution_mode = ort.ExecutionMode.ORT_SEQUENTIAL
options.inter_op_num_threads = 1
options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
options.enable_cpu_mem_arena = True
options.add_session_config_entry("session.dynamic_block_base", "4")
options.add_session_config_entry("session.intra_op.allow_spinning", "0")
return options
`
このオプションを適用してワーカーを起動すると、2 vCPU環境で平均CPU使用率を50%以下に抑えることができます。高価なGPUサーバーを使わなくても、月120ドル規模のインフラコストを節約できます。
ローカル環境やデプロイコンテナでSDKを読み込む際、C++の動的ライブラリの衝突が頻発します。
ImportError: DLL load failedが発生した場合は、Microsoft Visual C++ 2015-2022再頒布可能パッケージをインストールし、Pythonが64ビット仮想環境であることを確認します。libomp.dylibエラーが発生します。ターミナルでbrew install libompを実行し、環境変数にlibのパス(export DYLD_LIBRARY_PATH="$(brew --prefix libomp)/lib:$DYLD_LIBRARY_PATH")を追加します。python:3.10-slimイメージを使用する場合は、build-essentialとlibgomp1パッケージをaptで事前にインストールしておきます。環境を整えたら、入力テキストのクリーナーを適用する必要があります。英語の略語、数字、記号が混入すると、モデルの発音が崩れたり不自然な機械音を出力するためです。
`python
import re
from typing import Dict
class SupertonicTextNormalizer:
def init(self):
self.lexicon_map: Dict[str, str] = {
"FastAPI": "ファスト エピアイ",
"SaaS": "サース",
"TTS": "ティーティーエス",
"ONNX": "オニキス",
"Python": "パイソン",
"SDK": "エスディーケー",
"API": "エーピーアイ",
}
self.currency_pattern = re.compile(r'(\d+)\sウォン')
self.date_pattern = re.compile(r'(\d{4})年\s(\d{1,2})月\s*(\d{1,2})日')
self.time_pattern = re.compile(r'(\d{1,2}):(\d{2})')
self.special_char_pattern = re.compile(r'[^\w\s.,!?~<>]')
def normalize(self, text: str) -> str:
if not text or not text.strip():
raise ValueError("入力テキストが空です。")
for word, pronunciation in self.lexicon_map.items():
text = re.sub(rf'\b{re.escape(word)}\b', pronunciation, text, flags=re.IGNORECASE)
text = self.date_pattern.sub(r'\1年 \2月 \3日', text)
text = self.time_pattern.sub(r'\1時 \2分', text)
tags = re.findall(r'<[^>]+>', text)
text_placeholder = re.sub(r'<[^>]+>', ' ___TAG___ ', text)
text_cleaned = self.special_char_pattern.sub('', text_placeholder)
for tag in tags:
text_cleaned = text_cleaned.replace('___TAG___', tag, 1)
return re.sub(r'\s+', ' ', text_cleaned).strip()
`
C++推論モジュールが特定のテキストパターンで無限待機状態に陥るのを防ぐため、asyncio.wait_forでタイムアウトを設定し、失敗時にはあらかじめ用意したエラー案内音声を返す防御コードを配置します。
`python
import asyncio
import logging
logger = logging.getLogger("TTSPipeline")
async def synthesize_with_fallback(tts_engine, text: str, voice_style, timeout_sec: float = 3.0) -> bytes:
try:
normalizer = SupertonicTextNormalizer()
cleaned_text = normalizer.normalize(text)
loop = asyncio.get_running_loop()
wav_data = await asyncio.wait_for(
loop.run_in_executor(
None,
lambda: tts_engine.synthesize(text=cleaned_text, lang="ko", voice_style=voice_style)
),
timeout=timeout_sec
)
return wav_data
except Exception as err:
logger.error(f"TTS推論失敗またはタイムアウト: {err}")
with open("static/audio/fallback_system_error.wav", "rb") as f:
return f.read()
`
この前処理パイプラインを経由することで、不正確な発音による音声再生エラーが大幅に減少します。QA段階で発音の不具合を一つずつ確認して修正する時間を、毎週5〜6時間ほど節約できます。
FastAPIの非同期ルーター内で同期関数であるSupertonic 3の推論器をそのまま実行すると問題が発生します。C++の演算が完了するまで単一のイベントループ全体が停止し、他のユーザーの軽いAPIリクエストまで全て待機状態に陥ります。
演算集約的な推論作業は、別のProcessPoolExecutorにオフロードする必要があります。
`python
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from concurrent.futures import ProcessPoolExecutor
import asyncio
import io
import os
app = FastAPI()
process_pool = ProcessPoolExecutor(max_workers=min(4, os.cpu_count() or 1))
def sync_tts_inference(text: str, voice_style_name: str):
from supertonic import TTS
tts = TTS(auto_download=False)
style = tts.get_voice_style(voice_style_name)
wav, _ = tts.synthesize(text=text, lang="ko", voice_style=style)
return wav.tobytes()
@app.post("/api/v1/tts/realtime")
async def generate_speech_realtime(text: str, voice: str = "M1"):
loop = asyncio.get_running_loop()
try:
audio_bytes = await loop.run_in_executor(process_pool, sync_tts_inference, text, voice)
return StreamingResponse(io.BytesIO(audio_bytes), media_type="audio/wav")
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
`
生成されたオーディオをディスクにファイルとして書き込んでから再度読み込んで返す構造は、低スペックサーバーのディスクI/Oを消耗させます。ファイル保存を挟まず、io.BytesIOを用いてメモリ上で直接ストリーミングする方がはるかに高速です。
トラフィックが集中してキューが長くなったり、長文を処理しなければならない場合は、RedisキューとCeleryワーカーでリクエストを分離します。
task_idを即座に発行し、HTTP 202でレスポンスを返します。一時ファイルのキャッシュをどうしてもディスクに残さざるを得ない構造の場合は、バックグラウンドクリーンアップタスクを実行してディスクフル（Disk Full）障害を防止します。
`python
import os
import time
import glob
AUDIO_CACHE_DIR = "/tmp/supertonic_audio_cache"
MAX_FILE_AGE_SECONDS = 600
def cleanup_ephemeral_audio_files():
now = time.time()
if not os.path.exists(AUDIO_CACHE_DIR):
return
for filepath in glob.glob(os.path.join(AUDIO_CACHE_DIR, "*.wav")):
try:
if now - os.path.getmtime(filepath) > MAX_FILE_AGE_SECONDS:
os.remove(filepath)
except Exception:
pass
`
プロセスの分離とインメモリストリーミングを備えることで、2 vCPUインスタンスでも同時リクエスト処理時のp95遅延時間を200ミリ秒前後に抑えることができます。外部APIの料金高騰の心配なく、独立したオンデバイス音声サービスを安定してサービスに組み込むことができます。