音声データを処理する前処理パイプラインと実行基盤を作った話

概要編では、日本語の大規模音声基盤モデル(LLM-TTS)をスケールさせ、TTS性能を検証した結果を紹介しました。これを実現するためには、50万時間規模(wav換算で約80TB)の日本語音声を学習データとして用意する必要があります。

しかし、収集した音声は書き起こしを持たず、多くは雑音を含むため、そのまま学習に用いることは難しく、この規模の前処理には、以下の課題があります。

書き起こし・雑音除去・トークン化という性質の異なる複数の変換工程を、コスト制約の中で効率よく回す必要がある
逐次処理では現実的な時間で完了しないため、大規模な並列実行基盤が求められる
数百TBに及ぶデータの管理自体がボトルネックになりやすい

本記事では、まず前処理パイプラインの全体像と各処理の内容を説明し、次にステージ1を効率よく回す工夫、最後にそれを大規模に実行するための基盤について述べます。

関連記事

概要編：日本語音声基盤モデルをスケーリングさせ、TTS性能を見てみる

前処理編：この記事

事前学習編：（後日公開予定）

事後学習編：（後日公開予定）

1. 前処理パイプライン

1-1. 全体像

LLM-TTSの学習には、テキストトークン列と音声トークン列の2種類の入力が必要です。

テキストトークン列とは、文字列を離散的なトークン列に変換したものです。音声トークン列とは、ニューラルオーディオコーデックで、音声波形を離散的なトークン列として表現したものです。事前学習では、この2種類のトークン列を学習させます。

収集した音声を2つのトークン列に変換するまでにGPUを用いる処理として、

言語識別・書き起こし（ASR）
雑音除去（音声強調）
音声トークン化

の3つがあります。これらを以下の2つのステップに分けて実行しました。

ステージ1（言語識別・書き起こし + 雑音除去）: 収集した音声を入力として、書き起こしと雑音除去済み音声を出力する
ステージ2（音声トークン化）: ステージ1の音声を入力として、音声トークン列を生成し、ステージ1の書き起こしとともに出力する

2段に分けることで、音声トークナイザの突然の変更等に対応できるようにしました。ステージ1後の雑音除去済み音声＋書き起こしがあれば、異なる音声トークナイザを用いる場合であっても手戻りが小さくなります。

以下にパイプライン全体の流れを示します。

前処理パイプラインの全体像

1-2. ステージ1: 言語識別・書き起こし、雑音除去

言語識別・書き起こし

言語識別・書き起こし（ASR: Automatic Speech Recognition）では、収集した音声に対して言語判定と書き起こしを行います。

書き起こしのためのASRとして、faster-whisperのlarge-v3-turboモデルを使用しました。 faster-whisperは、OpenAIのWhisperモデルをCTranslate2で再実装したもので、精度の低下を抑えて推論速度を向上させています。また、Whisperは書き起こしだけでなく、言語識別タスクでも学習されています。そのため今回は言語識別でもfaster-whisperで行っています。

まず、収集した音声は、言語識別を用いて言語を判定し、日本語・英語以外と判定された音声を除外しました。収集した音声のファイル数の約8割が日本語・英語の音声でした。その後、音声に対して書き起こしを行い、結果は発話単位で分割し、各発話単位にタイムスタンプとテキストを付与します。発話単位で分割されたそれぞれを、以降セグメントと呼びます。

雑音除去

雑音除去では、音声強調の枠組みで、あらゆる劣化が含まれる音から音声（人間の発声器官から生成される音）を抽出することで、学習データとしての品質向上を図っています。これにより、学習したモデルが元の音声に含まれるような雑音を生成することを防げる他、学習データのドメインが可聴音全体から音声までドメインを狭めることが可能となり学習が容易になります。

今回は、言語識別やASRを掛けた後に音声強調を適用しました。ASRや言語識別では他言語や無音区間など一部のサンプルが除去されます。そのため先に音声強調を掛けると一部のサンプルに対する強調結果が無駄になってしまうからです。

音声強調にはSidonを使用しました。Sidonは、大規模データセット作成のための音声強調モデルで、ストリーミング推論に対応しています。出力は48kHzの音声ですが、データ容量削減のため24kHzにリサンプリングして保存しています。

1-3. ステージ1を効率よく回す工夫

ステージ1のGPUにはNVIDIA L4（VRAM 24GB）を使用しています。モデルや入力音声によってVRAMが足りなくなるので、処理の各所でメモリ制約を考慮した工夫が必要になりました。

言語識別・書き起こし: 入力音声の30分チャンク分割

faster-whisperのバッチ推論は、音声全体を固定長の断片に分割してからまとめて推論するため、入力音声が長いほどバッチサイズが増大し、VRAMを圧迫します。長時間音声をそのまま入力するとOut of Memoryが発生するため、入力音声を30分ごとの断片に分割して逐次的に書き起こしを行っています。

雑音除去: 固定長チャンクによるストリーミング処理

Sidonでは、前チャンクの隠れ状態を引き継ぐことでチャンク間をシームレスに再構成しています。公式の実装では96秒の固定長チャンク（GPUに入力する固定長の音声断片）で推論していますが、今回はこれを10秒に短縮してストリーミング推論を行っています。チャンクサイズを固定長かつ短くすることで、JITコンパイルのキャッシュが有効に機能し、推論速度の向上にも寄与しています。

1-4. ステージ2: 音声トークン化

ステージ1で得られた24kHz音声を入力として、音声トークン列に変換します。

音声トークンとしてNeuCodecを用いました。 NeuCodecは、音声波形を圧縮し、有限スカラー量子化によって離散トークンの列に変換するニューラル音声コーデックです。この音声トークンを用いて、モデル学習を行います。

出力されるJSONLの各行は以下のような構造を持ちます。

{
  "id": "000000051649_0114",
  "text": "書き起こしテキスト",
  "codes": {"neu_tokens": [3303, 3175, 56535, ...]},
  "duration_seconds": 26.65
}

neu_tokensがNeuCodecの出力するトークン列で、モデル訓練時に使われます。

2. 実行基盤

ここからは、前処理パイプラインを50万時間規模で実行するための基盤について説明します。まずデータ形式とジョブ管理ツールの選定を述べ、次にジョブ内のマルチプロセス設計とジョブのスケールアウト、最後に運用で得られた知見をまとめます。

なお、以降ではジョブを「1台のインスタンス上で動く1つの処理単位」、ワーカーを「1つのジョブ内部で特定の処理を担当するプロセス」の意味で使い分けます。

2-1. データ形式とジョブ管理ツールの選定

データ形式: WebDataset

50万時間分の音声データを個々のファイルとしてファイルシステム上で管理すると、ファイル数が数億規模になります。このことにより、以下の問題が生じます。

ファイル本体に加えてインデックスも保持する必要があるため、大量のファイルが存在すると追加のストレージ容量が発生する
大量の小さなファイルへのランダムアクセスはストレージのI/O性能を著しく低下させ、ファイルのリスト取得やコピーだけでも膨大な時間がかかる

また、このほかにもデータ全体をメモリに載せることは不可能であり、ストリーミング的に読み込みながら処理する仕組みが必要です。

これらの問題に対して、WebDatasetを採用しました。WebDatasetは、複数のファイルをtarアーカイブ(以降、シャードと呼びます)にまとめ、シーケンシャルアクセスで効率的に読み出すための仕組みです。

PyTorchのIterableDataset互換のインターフェースを提供しており、ストリーミング処理のパイプラインを容易に構築できます。tarアーカイブによるファイルのまとめ管理でファイル数の問題が解消され、シーケンシャルアクセスによってI/Oがボトルネックになりにくくなります。さらにtarアーカイブが安定してどのようなプラットフォーム上でも動作し、管理が楽というメリットもあります。

ジョブ管理: SkyPilot

クラウド環境で安く抑えるために、スポットインスタンス（クラウドプロバイダの余剰リソースを割安に利用できるインスタンス）を活用します。しかし、スポットインスタンスにはいくつかの運用上の課題があります。

リソースの空き状況はリージョン（データセンターの地域）ごとに異なり、特定のリージョンでは確保できないことがある
プロバイダ側の都合で予告なく強制停止される
複数リージョン・複数プロバイダにまたがるインスタンスを手動で管理するのは非現実的である

これらの課題に対して、SkyPilotを活用しました。SkyPilotは、複数のクラウドプロバイダ・リージョンにまたがってスポットインスタンスを統一的に管理できるツールです。SkyPilotの具体的な活用方法は、スケールアウトの節であわせて説明します。

2-2. ジョブ内のマルチプロセス設計

前処理パイプラインは、「データのダウンロード」「GPU上での推論」「結果の書き出しとアップロード」という性質の異なる3つの処理から構成されています。これらを単一のプロセスで逐次実行すると、GPUがI/O待ちでアイドルになる時間が生まれ、GPUリソースを有効活用できません。

そこで、処理の性質ごとにプロセスを分離し、パイプライン的に並行動作させることでGPUの稼働率を上げています。また、1つのGPU上で、言語識別・書き起こしと雑音除去を直列に実行するTranscription Workerを2並列で動かしています。プロセス間の受け渡しには、Main Process → Transcription Worker間の task_q、Transcription Worker → Writer Worker間の writer_q という2つのキューを用いています。以下にジョブ内部のプロセス構成を示します。

ジョブ内部のプロセス構成

Main Processがシャードの取得とダウンロード、task_q への投入を担当し、Transcription Workerは推論に専念します。Transcription Workerは、結果の圧縮・書き出し・アップロードをWriter Workerに任せるので、次の音声の処理にすぐ移行できます。

データセットのダウンロードにおいて

WebDatasetはオブジェクトストレージのストリーミング読み出しに対応していますが、スポットインスタンスでは地理的に分散したリージョンを使う関係で接続断が頻発しやすく、途中からの再開も困難です。そのため、シャード単位で並列ダウンロードしてローカルに取得してから処理する方式にしました。

ステージ間の同期

ステージ1が完了したシャードを元に、ステージ2が順次処理を開始できるようにしています。

ステージ1の処理済みシャード情報のステージ間の受け渡しは、DBもメッセージキューを使わず、テキストファイルを介して実現しています。

ステージ間の同期（処理済みシャード一覧への追記）

ステージ1では各シャード処理完了時に処理済みシャードをテキストファイルへ追記し、ステージ2のジョブがこのファイルを読み込みます。ステージ1の各シャード処理完了ごとに追記されていく構造のため、ステージ2はこの更新を読み進めることで、2つのステージをパイプライン的に並行して動かせる設計になっています。

同じシャードを複数のジョブが取得しないように、読み出し側の設計で回避しています。各ジョブには担当するシャード番号（ステージ2ではシャード範囲）が与えられ、ジョブは各シャードに対してのみ処理を行います。ジョブ投入時点でレンジが排他的に分割されているため、同じシャードを2つのジョブが同時に処理することは起こりません。

DBやメッセージキューを使わないことで、追加のインフラを管理せずに済みます。ネットワークが論理的に隔離されている計算環境であっても、処理済みシャード情報を同じオブジェクトストレージにファイルとして置けるため、外部サービスを立てる必要がないです。状態を見たいときはテキストファイルを読むだけでよく、運用も簡単です。

2-3. ジョブのスケールアウト

個々のジョブの設計に加えて、数十〜数百台のインスタンスを同時に運用するための仕組みが必要です。今回、運用にはSkyPilotを用いました。以降では、SkyPilotのいくつかの具体的な活用方法を説明します。

自動リージョン選択

SkyPilotでは、リージョン候補を列挙しておくと、スポットの空き状況と価格を自動的に比較し、最適なリージョンを選択してジョブを起動します。今回は複数のプロバイダかつ20以上のリージョンを候補に設定し、特定の地域でスポットが枯渇しても別のリージョンで起動できるようにしました。

自動リカバリ

SkyPilotのManaged Jobは、スポットインスタンスが強制停止されると自動的にRECOVERING状態へ遷移し、別リージョンでの再起動を試みます。これにより、強制停止が発生してもジョブが自動的に復旧し、手動での再起動が不要になります。

宣言的なインスタンス数維持

SkyPilot自体にはオートスケーリング機能がないため、spot_managerという自作コンポーネントで目標台数を維持しています。以下にspot_managerの設定例を示します。

spot_manager:
  poll_interval_sec: 300
  max_launch_per_cycle: 10
  targets:
    gcp-task-a:
      type: task-a
      count: 50
      template: skypilot_yaml/task-a_gcp_spot.yaml
    gcp-task-b:
      type: task-b
      count: 10
      template: skypilot_yaml/task-b_gcp_spot.yaml

ポーリング間隔（今回は5分で実施）ごとにSkyPilotのジョブキューからアクティブジョブ数を取得し、目標台数との差分だけ新規ジョブを起動します。

spot_manager のポーリングループ

RECOVERING状態のジョブもアクティブとしてカウントすることで、SkyPilotの自動リカバリと重複して起動することを防いでいます。各ジョブには{target_name}-{uuid4短縮}の形式でユニークな名前が付与され、ジョブ名のプレフィックスでフィルタリングすることで、複数種類のジョブを同じSkyPilotコントローラーで混在管理できます。

2-4. 運用で得られた知見

同時起動数と強制停止率の関係

あるクラウドプロバイダでは、1アカウントで50台以上のスポットインスタンスを同時に起動すると、強制停止率が急激に上昇する傾向が見られました。強制停止が発生するとジョブの再起動にオーバーヘッドがかかるため、1プロバイダあたりの同時起動数を50台に制限し、それ以上のスケールが必要な場合は複数のクラウドプロバイダに分散させる運用としました。

起動直後の不安定性

スポットインスタンスは毎回クリーンな状態から起動するため、OS起動直後はパッケージマネージャのロック（/var/lib/dpkg/lock-frontendなど）がまだ解放されていないことがあります。セットアップスクリプトでapt installが失敗した際は30秒待ってリトライするようにしています。

3. まとめ

本記事では、50万時間規模の音声データをLLM-TTSの訓練データへ変換するためのデータセット作成パイプラインについて述べました。

前処理パイプラインは、再利用可能性を考慮して言語識別・書き起こし・雑音除去のステージ1と音声トークン化のステージ2に分離しました。ステージ1のGPU（L4, 24GB VRAM）については、ASRの30分チャンク分割、Sidonのストリーミング処理、2並列のTranscription Workerといった工夫で対処しています。

実行基盤としては、WebDatasetによるI/O効率化、プロセス分離によるGPU稼働率の向上、SkyPilotと自作のspot_managerによるマルチプロバイダ・マルチリージョンでのスポットインスタンス管理を組み合わせました。

今後は、データセットの収集から前処理まで全てを一貫したパイプラインとして構築する等の運用面でのさらなる自動化を進めることで、運用負荷を下げていけると考えています。

関連記事

概要編：日本語音声基盤モデルをスケーリングさせ、TTS性能を見てみる

前処理編：この記事

事前学習編：（後日公開予定）

事後学習編：（後日公開予定）