【AI Shift Advent Calendar 2023】Cloud Run Jobs + faster-whisperで構築する簡易的音声認識システム

gcloud storage buckets create gs://${BUCKET}

gcloud artifacts repositories create ${REPOSITORY} --repository-format=docker \
--location=${LOCATION} --description=${DESCRIPTION}

├── src
│   ├── Dockerfile
│   ├── cloudbuild.yaml // Cloud Build用の設定ファイル
│   └── main.py // 実行スクリプト

FROM python:3.10.6-slim

RUN apt-get update && apt-get install -y \
    build-essential \
    gcc \
    git \
    ffmpeg \
    && rm -rf /var/lib/apt/lists/*

# 必要なライブラリのインストール
RUN pip install --upgrade pip && \
    pip install -r google-cloud-storage==2.10.0 pandas==2.1.0 \
    transformers[torch]==4.33.0 faster-whisper==0.8.0

# 実行ファイルのコピー
ENV APP_HOME /app
WORKDIR $APP_HOME
COPY main.py ./

# Whisperのモデル変換
RUN ct2-transformers-converter --model openai/whisper-large-v2 --output_dir whisper-large-v2-int8 \
    --copy_files tokenizer.json --quantization int8

CMD ["/usr/local/bin/python3", "main.py"]

steps:
- name: 'gcr.io/cloud-builders/docker'
  args: [ 'build', '-t', '${LOCATION}-docker.pkg.dev/${PROJECT}/${REPOSITORY}/${IMAGE}:latest', '.' ]
options:
  machineType: 'E2_HIGHCPU_32'
images: ['${LOCATION}-docker.pkg.dev/${PROJECT}/${REPOSITORY}/${IMAGE}:latest']

gcloud builds submit --config cloudbuild.yaml .

gcloud run jobs create ${JOB} \
    --image '${LOCATION}-docker.pkg.dev/${PROJECT}/${REPOSITORY}/${IMAGE}:latest' \
    --region ${LOCATION} \
    --set-env-vars BUCKET=${BUCKET} \
    --task-timeout 3600s \
    --cpu ${CPU_NUM} --memory ${MEMORY_SIZE} \
    --execute-now

import io
import os
import pandas as pd
from google.cloud import storage
from faster_whisper import WhisperModel

INCOMING_PREFIX = "incoming"
PROCESSED_PREFIX = "processed"

def speech2text(blob, model):
    audio_data = blob.download_as_bytes()
    segments, info = model.transcribe(
        io.BytesIO(audio_data),
        language="ja",
        task="transcribe",
        vad_filter=True,
        without_timestamps=True,
    )

    processed = []
    for segment in segments:
        processed.append({"start": segment.start, "end":segment.end, "transcription":segment.text})    
    df = pd.json_normalize(processed)
    return df


def main():
    # 環境変数の取得
    bucket_name = os.getenv("BUCKET")

    # モデルの読み込み
    model = WhisperModel("whisper-large-v2-int8", device="auto", compute_type="int8", local_files_only=True)

    client = storage.Client()
    bucket_obj = client.bucket(bucket_name)

    for blob in client.list_blobs(bucket_name, prefix=f"{INCOMING_PREFIX}/"):
        # フォルダの場合はスキップ
        if blob.name.endswith("/"):
            continue    

        # faster-whisperを用いた音声認識
        df = speech2text(blob, model)

        # 処理後結果をprocessedフォルダに格納
        bare_name = blob.name[len(INCOMING_PREFIX)+1:].rsplit(".", 1)[0]
        csv_new_name = f"{PROCESSED_PREFIX}/csv/{bare_name}.csv"
        
        # 結果の保存(CSV)
        csv_blob = bucket_obj.blob(csv_new_name)
        csv_blob.upload_from_string(df.to_csv(index=False), 'text/csv')


if __name__ == "__main__":
    main()