Closed9

Azure AI Speech ServiceのBatch transcription APIを試す（Python / Colab)

Azure
stt
hosaka313hosaka313

サービスの作成（Azure管理画面）

まずサービスを作成する。価格は無料ではなく、有料の方にした。

hosaka313hosaka313

キーの取得

作ったサービスにアクセスすると、APIリクエストに必要な各種情報がトップに出ている。

hosaka313hosaka313

Azure Blob Storageの作成

音声ファイルをアップするためのAzure Blob Storageコンテナを作成する。

ストレージアカウントをまず作成。その後、コンテナを作成。

hosaka313hosaka313

ストレージのアクセスキーを取得

「セキュリティとネットワーク」> 「アクセスキー」からキーを取得

hosaka313hosaka313

コード作成

config

シークレットはColaboratoryのシークレット管理機能を使用。

from google.colab import userdata

# ==== 1. 環境変数を入力 =========================
AZ_SPEECH_KEY      = userdata.get('AZ_SPEECH_KEY')
AZ_SPEECH_REGION   = "eastus"   # 例: japaneast
STG_ACCOUNT_URL    = "https://<account_id>.blob.core.windows.net"
STG_ACCOUNT_KEY    = userdata.get('STG_ACCOUNT_KEY')
STG_CONTAINER_NAME = "batch-in"
# (任意) カスタムモデルを使う場合は self URL をここへ
CUSTOM_MODEL_URL   = None  # 例: "https://japaneast.api.cognitive.microsoft.com/…/models/custom/abcd1234"

カスタムモデルは今回は使わない。

ファイルアップロード

from google.colab import files

# ==== 2. Colab に音声ファイルをアップロード ======
uploaded = files.upload()
local_audio = list(uploaded.keys())[0]           # 1 ファイル想定

Blobストレージへのアップ

import os
import uuid
import datetime
from azure.storage.blob import (
    BlobServiceClient, 
    BlobSasPermissions, 
    generate_blob_sas
)

# ==== 3. Blob へアップロードし SAS URL を発行 =====
blob_name = f"{uuid.uuid4()}_{os.path.basename(local_audio)}"
blob_service = BlobServiceClient(account_url=STG_ACCOUNT_URL,
                                 credential=STG_ACCOUNT_KEY)
container_client = blob_service.get_container_client(STG_CONTAINER_NAME)
container_client.upload_blob(name=blob_name,
                             data=open(local_audio, "rb"),
                             overwrite=True)

expiry = datetime.datetime.utcnow() + datetime.timedelta(hours=1) 
sas_token = generate_blob_sas(
    account_name=blob_service.account_name,
    container_name=STG_CONTAINER_NAME,
    blob_name=blob_name,
    account_key=STG_ACCOUNT_KEY,
    permission=BlobSasPermissions(read=True),
    expiry=expiry
)
content_url = f"{STG_ACCOUNT_URL}/{STG_CONTAINER_NAME}/{blob_name}?{sas_token}"

Batchジョブの作成

# ==== 4. バッチジョブの作成 ==================
headers = {
    "Ocp-Apim-Subscription-Key": AZ_SPEECH_KEY,
    "Content-Type": "application/json"
}
endpoint = f"https://{AZ_SPEECH_REGION}.api.cognitive.microsoft.com/speechtotext/v3.2/transcriptions"
job_body = {
    "contentUrls": [content_url],
    "locale": "ja-JP",
    "displayName": f"colab-job-{uuid.uuid4()}",
    "properties": {
        "punctuationMode": "Automatic",
        "diarizationEnabled": True,
        "diarization": { 
          "speakers": {
              "minCount": 2, "maxCount": 4 
            }
        },
        "wordLevelTimestampsEnabled": True
    }
}
if CUSTOM_MODEL_URL:
    job_body["model"] = { "self": CUSTOM_MODEL_URL }

try:
    job_resp = requests.post(endpoint, headers=headers, json=job_body)
    job_resp.raise_for_status()
except requests.HTTPError as e:
    print("status:", job_resp.status_code)
    print(job_resp.text)
    raise
job_url = job_resp.json()["self"]
print("ジョブ開始:", job_url)

# ==== 5. ステータスをポーリング ==================
def wait_job(url, headers, interval=15):
    while True:
        resp = requests.get(url, headers=headers)
        resp.raise_for_status()
        info = resp.json()
        print(f"[{time.strftime('%X')}] {info['status']}")
        if info["status"] in ("Succeeded", "Failed"):
            return info
        time.sleep(interval)

job_info = wait_job(job_url, headers)

# ---- 失敗時は理由を表示して終了 ---------------------------
if job_info["status"] == "Failed":
    err = job_info.get("error") or job_info.get("properties", {}).get("error")
    print("❌ Job failed:")
    print(json.dumps(err, indent=2, ensure_ascii=False))
    raise SystemExit

pollingを実装。話者分離は以下のように指定する。

        "diarization": { 
          "speakers": {
              "minCount": 2, "maxCount": 4 
            }
        },
hosaka313hosaka313

結果の分析

jobの結果に結果ファイルへのリンクがあるので、Getする。

# ==== 6. 結果ファイルをダウンロード ===============
files_link = job_info["links"]["files"]
result_files = requests.get(files_link, headers=headers).json()["values"]
# 最初の Transcription ファイルを取得
trans_url = next(f["links"]["contentUrl"] for f in result_files
                 if f["kind"] == "Transcription")
trans_json = requests.get(trans_url).json()

# ==== 7. 話者ごとに読みやすく表示 ================
for segment in trans_json["recognizedPhrases"]:
    speaker = segment["speaker"]
    offset  = segment["offset"]
    text    = segment["nBest"][0]["display"]
    print(f"[{offset}] Speaker {speaker}: {text}")

以下のような結果が得られる。

[PT0.16S] Speaker 1: あ、自動化しているということを。もしよければもう少しお伺いでき。
[PT5.28S] Speaker 2: ますか？なんかあのスプレッドシートで。はい、要素はだ、あっと作ったやつがどっかにあの場合分け表みたいなやつを。うん、あの、作ってるんです。あ、こっちか。はい。
[PT20.72S] Speaker 1: はいはいはい。おお、すごいで。
[PT23.2S] Speaker 2: す。いやもうね、力ずくですよ。これい。
[PT27.52S] Speaker 1: やこういうデータすごく貴重な気がす。
[PT31.76S] Speaker 2: る、いや、これは逆に。まあ、それはなんとなくあの、これは貴重なんだろうなっていうのはなんとなく想像はしてたんですけど、ちょっとこんな力ずくでやんないじゃないですか。普通はいいやで。これ全部で何行あるんだっけ？
hosaka313hosaka313

感想

  • Microsoftの常でドキュメントが読みづらく、実装例もあまりないので苦労した。
  • 話者の区切りは、話者間のインターバルがないと（重なって話すと）、精度がいまいち。
  • Gemini 2.0以降は文字起こしが優秀。今となってはそちらの方が良いのでは。GeminiならPhrase Listを登録する必要もなく、プロンプトを書けば良い。

というわけで、音声特化ではないGeminiなどのマルチモーダルLLMに対して、実装容易性、精度、コストの面でBatch transcription APIにどれだけ優位性があるのか、疑問符が付くという結論に至った。

このスクラップは5ヶ月前にクローズされました