assemblyai-streaming

AssemblyAI ストリーミングとライブ文字起こしスキル

概要

このスキルを使って、AssemblyAI の以下のサービスと連携するコードを作成・保守します。

• WebSocket（wss://streaming.assemblyai.com/v3/ws）経由のストリーミング音声認識（STT）
• REST（https://api.assemblyai.com/v2/transcript）経由の非同期／事前録音済み音声のSTT
• 文字起こしにClaude/GPT/Gemini系モデルを適用するLLM Gateway（https://llm-gateway.assemblyai.com）

主にストリーミング／ライブ文字起こし、会議の議事録作成ツール、音声エージェントを扱いますが、非同期ワークフローと後処理についても説明します。

このスキルは、Python（推奨）とBashを利用できるClaude Code環境を前提としています。

---

使用する場面

次のような場合にこのスキルを使用します。

• マイク、電話の音声ストリーム、または音声ファイルからのリアルタイム文字起こしを実装する場合。
• ライブ会議の議事録作成ツール（Zoom/Teams/Meet）を構築する場合。特に、要約、アクションアイテム、重要なポイントを扱う場合。
• レイテンシと自然なターンテイキングが重要な音声エージェントを実装する場合。
• 他のSTTプロバイダー（OpenAI/Deepgram/Google/AWSなど）からAssemblyAIに移行する場合。
• LLM Gatewayを通じてLLMを音声に適用し、要約、質疑応答、トピックタグ付け、カスタムプロンプトなどを行う場合。

次のような場合は、このスキルを使用しないでください。

• AssemblyAI固有のロジックを伴わない、一般的なHTTPクライアントの使用が目的の場合。
• 別のSTTベンダーを対象としていることが明らかな場合。
• 環境内でAPIキーを安全に保存または使用できない場合。

---

AssemblyAIの基本的な考え方

1. 押さえておきたい製品

1. 事前録音済み音声の音声認識（非同期）

   • REST API：POST /v2/transcript → GET /v2/transcript/{id}
   • URL、アップロード、S3などからのファイル向けに設計されています。
   • 要約、トピック検出、感情分析、個人情報のマスキング、チャプターなどの追加モデルに対応しています。

2. ストリーミング音声認識

   • WebSocket：wss://streaming.assemblyai.com/v3/ws
   • 低レイテンシで、変更されない文字起こし結果が得られます（約300ms）。
   • ターン検出機能が組み込まれており、音声エージェントやライブ字幕に適しています。

3. LLM Gateway

   • REST API：https://llm-gateway.assemblyai.com の POST /v1/chat/completions
   • 複数のLLM（Claude、GPT、Geminiなど）を統一的に利用できます。
   • 「文字起こしに対するLLM処理」のワークフロー向けに設計されています。

2. 主なモデル設定（非同期）

• speech_models：["slam-1", "universal"] など
  • Slam-1：英語の精度が最も高く、keyterms_promptに対応しており、医療や技術的な会話に適しています。
  • Universal：多言語に対応しており、言語が不明な場合の適切なデフォルトです。
• language_code と language_detection：
  • 言語がわかっている場合は language_code を使用します。
  • 不明な場合は language_detection: true を使用します。必要に応じて language_confidence_threshold を設定します。
• keyterms_prompt：
  • 認識精度を高めるための分野固有の単語やフレーズ（医療用語、製品名など）。
• 追加のインテリジェンス機能：summarization、iab_categories、content_safety、entity_detection、auto_chapters、sentiment_analysis、speaker_labels、auto_highlights、redact_pii など。

3. 主なモデル設定（ストリーミング）

接続 URL：

• 米国：wss://streaming.assemblyai.com/v3/ws
• EU：wss://streaming.eu.assemblyai.com/v3/ws

重要なクエリパラメーター：

• sample_rate（必須）：例：16000
• format_turns（ブール値）：整形済みの最終文字起こしを返します。低レイテンシの音声エージェントでは使用を避けてください。
• speech_model：universal-streaming-english（デフォルト）または universal-streaming-multi。
• keyterms_prompt：用語の JSON エンコード済みリスト。例：["AssemblyAI", "Slam-1", "Keanu Reeves"]。
• ターン検出：
  • end_of_turn_confidence_threshold（0.0～1.0、デフォルトは約0.4）
  • min_end_of_turn_silence_when_confident（ミリ秒、デフォルトは約400）
  • max_turn_silence（ミリ秒、デフォルトは約1280）

ヘッダー：

• Authorization: <API_KEY> またはバックエンドが発行する有効期間の短い token クエリパラメーターのいずれかを使用します。

メッセージ：

• クライアントが送信するもの：
  • バイナリー音声チャンク（各50～1000ミリ秒）。
  • 任意の JSON メッセージ：{"type": "UpdateConfig", ...}、{"type": "Terminate"}、{"type": "ForceEndpoint"}。
• サーバーが送信するもの：
  • id と expires_at を含む Begin イベント。
  • 次の内容を含む Turn イベント：
    • transcript（変更されない部分的な文字起こし／最終文字起こし）
    • utterance（意味的に完結したチャンク）
    • end_of_turn（ブール値）
    • turn_is_formatted（ブール値）
    • タイムスタンプと信頼度を含む words 配列。
  • 要約統計を含む Termination イベント。

4. リージョンとデータレジデンシー

• 非同期：
  • 米国：https://api.assemblyai.com
  • EU：https://api.eu.assemblyai.com
• ストリーミング：
  • 米国：wss://streaming.assemblyai.com/v3/ws
  • EU：wss://streaming.eu.assemblyai.com/v3/ws

プロジェクト内では常にベース URL を統一してください。同じデータに対して米国と EU のエンドポイントを混在させないでください。

---

セキュリティと API キー

• 常に AssemblyAI API キーを必須とし、Claude Code の出力にソースとして含めないでください。
  • 環境変数 ASSEMBLYAI_API_KEY を使用します。
  • または、コードスニペットではプレースホルダー（"<YOUR_API_KEY>"）を使用します。
• ブラウザー／クライアントのコードでは：
  • API キーを埋め込まないでください。
  • バックエンドで一時的なストリーミングトークンを生成し、WebSocket 接続にはトークンのみを渡すようユーザーに指示してください。
• 実際のキーをログやコメントに出力しないでください。

---

大まかなワークフローパターン

判断ツリー

1. 音声はライブですか？

   • はい → ストリーミング STT を使用します。
   • いいえ → 非同期 STT を使用します。

2. 応答のレイテンシ（1秒未満）が重要ですか？

   • はい → format_turns=false と慎重なターン検出を使用してストリーミングします。
   • いいえ → 非同期処理の後、要約／チャプターなどを実行します。

3. 文字起こしがバックエンドの外に出ますか？

   • はい → 共有する前に redact_pii（および任意で redact_pii_audio）を検討します。
   • いいえ → 必要に応じて未加工の文字起こしを使用します。

4. LLM ベースの処理（Q&A、構造化された要約）が必要ですか？

   • はい → 文字起こしを chat/completions 経由で LLM Gateway に渡します。

---

Claude がこのスキルを使用する際の指針

基本原則

• 利用可能な場合は公式 AssemblyAI SDK（Python／JS）を優先します。SDK をインストールできない場合に限り、requests／websocket-client に切り替えます。
• 常に次のことを行います。
  • HTTP レスポンスと WebSocket のステータスを検証します。
  • 有用なエラーメッセージを提示します（文字起こし JSON の status、error フィールド）。
  • ドキュメントに記載されたチャンクサイズの最小値／最大値（バイナリーメッセージあたり50～1000ミリ秒の音声）を守ります。
• 音声エージェントのコードでは、次の点を最適化します。
  • 変更されない部分的な文字起こし（transcript）と utterance フィールド。
  • レイテンシの最小化。追加の整形処理は避けます。

---

レシピ 1 – マイクからの最小限のストリーミング（Python SDK）

目的：マイクの音声を AssemblyAI にストリーミングし、文字起こしをリアルタイムで出力します。

Python と assemblyai + pyaudio がインストールされている環境で、ユーザーが手軽なストリーミングデモを希望する場合に使用します。

import assemblyai as aai
from assemblyai.streaming import v3 as aai_stream
import pyaudio

API_KEY = "<YOUR_API_KEY>"

aai.settings.api_key = API_KEY

SAMPLE_RATE = 16000
CHUNK_MS = 50
FRAMES_PER_BUFFER = int(SAMPLE_RATE * (CHUNK_MS / 1000.0))

def main():
    client = aai_stream.StreamingClient(
        aai_stream.StreamingClientOptions(
            api_key=API_KEY,
            api_host="streaming.assemblyai.com",  # or "streaming.eu.assemblyai.com"
        )
    )

    def on_begin(_client, event: aai_stream.BeginEvent):
        print(f"Session started: {event.id}, expires at {event.expires_at}")

    def on_turn(_client, event: aai_stream.TurnEvent):
        # Use immutable transcript text
        text = (event.transcript or "").strip()
        if not text:
            return
        # Use formatted finals only for display; keep unformatted for LLMs
        if event.turn_is_formatted:
            print(f"[FINAL] {text}")
        else:
            print(f"[PARTIAL] {text}", end="\r")

    def on_terminated(_client, event: aai_stream.TerminationEvent):
        print(f"\nTerminated. Audio duration={event.audio_duration_seconds}s")

    def on_error(_client, error: aai_stream.StreamingError):
        print(f"\nStreaming error: {error}")

    client.on(aai_stream.StreamingEvents.Begin, on_begin)
    client.on(aai_stream.StreamingEvents.Turn, on_turn)
    client.on(aai_stream.StreamingEvents.Termination, on_terminated)
    client.on(aai_stream.StreamingEvents.Error, on_error)

    client.connect(
        aai_stream.StreamingParameters(
            sample_rate=SAMPLE_RATE,
            format_turns=False,  # better latency for voice agents
        )
    )

    pa = pyaudio.PyAudio()
    stream = pa.open(
        format=pyaudio.paInt16,
        channels=1,
        rate=SAMPLE_RATE,
        input=True,
        frames_per_buffer=FRAMES_PER_BUFFER,
    )

    try:
        print("Speak into your microphone (Ctrl+C to stop)...")
        def audio_gen():
            while True:
                yield stream.read(FRAMES_PER_BUFFER, exception_on_overflow=False)
        client.stream(audio_gen())
    except KeyboardInterrupt:
        pass
    finally:
        client.disconnect(terminate=True)
        stream.stop_stream()
        stream.close()
        pa.terminate()

if __name__ == "__main__":
    main()