Transcribe v2.0 完全ガイド（Machine Learning & AI）

概要

Amazon Transcribe は、機械学習によって音声をテキストに自動変換する自動音声認識（ASR: Automatic Speech Recognition）サービスです。録音済み音声ファイル（バッチ処理）またはリアルタイムストリーミング音声の両方に対応し、話者識別・カスタム語彙・PII（個人識別情報）検出・コールセンター特化分析を提供します。医療専門用語に対応した Transcribe Medical、リアルタイムチャット字幕生成、議事録自動作成、複数話者の発言分離など、エンタープライズグレードの音声AI機能を備えています。

課題と特徴

従来型ASRの課題

• 話者識別の手動作業: 複数人の通話で「誰が何を話したか」を手動で判別する必要
• 専門用語の誤認識: 医療・法律・金融用語が汎用モデルでは正しく認識されない
• 個人情報の手動マスキング: コンプライアンス対応に人手作業が必要
• スケーリングの複雑さ: 大量の音声処理にGPUインフラ構築が必要

Transcribeの特徴

アーキテクチャ

graph TB
    subgraph Input["音声入力"]
        A["リアルタイム音声<br/>WebSocket/HTTP2"]
        B["S3 保存済み音声<br/>MP3/WAV/FLAC"]
    end
    
    subgraph Processing["Transcribe処理"]
        C["言語検出"]
        D["モデル選択<br/>Standard/Medical"]
        E["カスタムVocab適用"]
        F["話者識別"]
        G["PII検出"]
        H["感情分析<br/>Call Analytics"]
    end
    
    subgraph Output["出力"]
        I["JSON/VTT字幕"]
        J["Call Analytics<br/>スコア"]
        K["医療コード<br/>ICD-10等"]
    end
    
    A --> C
    B --> C
    C --> D
    D --> E
    E --> F
    F --> G
    G --> H
    H --> I
    H --> J
    H --> K

コアコンポーネント

1. Streaming Transcription（リアルタイム）

import boto3
from amazon_transcribe.client import TranscribeStreamingClient
from amazon_transcribe.model import TranscriptEvent
import asyncio

async def stream_transcribe():
    client = TranscribeStreamingClient(region='ap-northeast-1')
    
    # ストリーミング開始
    stream = await client.start_stream_transcription(
        language_code='ja-JP',
        media_sample_rate_hz=16000,
        media_encoding='pcm',
        vocabulary_name='medical-terms',  # カスタム語彙
    )
    
    # マイク入力 → ストリーム送信
    async def write_chunks():
        with open('audio.wav', 'rb') as audio:
            for chunk in iter(lambda: audio.read(1024), b''):
                await stream.input_stream.send_audio_event(audio_chunk=chunk)
        await stream.input_stream.end_stream()
    
    # リアルタイム結果受信
    async def read_stream():
        async for transcript_event in stream:
            if isinstance(transcript_event, TranscriptEvent):
                for result in transcript_event.transcript.results:
                    if not result.is_partial:
                        print(f"最終: {result.alternatives[0].transcript}")
                    else:
                        print(f"中間: {result.alternatives[0].transcript}")
    
    await asyncio.gather(write_chunks(), read_stream())

asyncio.run(stream_transcribe())

2. Batch Transcription（非同期）

import boto3

transcribe = boto3.client('transcribe', region_name='ap-northeast-1')

# バッチ開始
response = transcribe.start_transcription_job(
    TranscriptionJobName='meeting-2026-04-26',
    Media={'MediaFileUri': 's3://my-bucket/audio/meeting.mp3'},
    MediaFormat='mp3',
    LanguageCode='ja-JP',
    Settings={
        'ShowSpeakerLabels': True,
        'MaxSpeakerLabels': 5,
        'ChannelIdentification': False,  # ステレオ音声の左右分離
        'VocabularyName': 'legal-terms',
        'VocabularyFilterName': 'profanity-filter',
        'VocabularyFilterMethod': 'mask',  # mask/remove
        'ContentRedaction': {
            'RedactionType': 'PII',
            'RedactionOutput': 'redacted'  # redacted/redacted_and_unredacted
        }
    },
    OutputBucketName='my-bucket',
    OutputKey='transcripts/',
    Tags=[
        {'Key': 'Project', 'Value': 'customer-support'},
        {'Key': 'Department', 'Value': 'quality-assurance'}
    ]
)

job_name = response['TranscriptionJob']['TranscriptionJobName']

# ジョブ完了を確認
waiter = transcribe.get_waiter('transcription_job_completed')
waiter.wait(TranscriptionJobName=job_name)

# 結果取得
result = transcribe.get_transcription_job(TranscriptionJobName=job_name)
print(f"ステータス: {result['TranscriptionJob']['TranscriptionJobStatus']}")
print(f"出力URI: {result['TranscriptionJob']['Transcript']['TranscriptFileUri']}")

3. Speaker Diarization（話者識別）

# 出力例: 複数話者を自動識別して分離
{
  "jobName": "meeting-2026-04-26",
  "results": {
    "transcripts": [{
      "transcript": "本日はご参加ありがとうございます。"
    }],
    "speaker_labels": {
      "speakers": 2,
      "segments": [
        {
          "start_time": "0.10",
          "end_time": "2.50",
          "speaker_label": "spk_0",
          "items": [...]
        },
        {
          "start_time": "2.70",
          "end_time": "4.50",
          "speaker_label": "spk_1",
          "items": [...]
        }
      ]
    }
  }
}

# 解析例
for segment in speaker_labels['segments']:
    speaker = segment['speaker_label']
    text = extract_segment_text(segment)
    print(f"{speaker}: {text}")

4. Custom Vocabulary（カスタム語彙）

import csv
import io

# 専門用語ファイル作成
vocab_data = """
Phrase,IPA,SoundsLike,DisplayAs
Amazon Web Services,@m@z,@n w.@b s.@rvi.s,Amazon Web Services
HIPAA,hI.p.@,hip ay,HIPAA
ニューラルネットワーク,n.@.r.@l n.@t.w.@rk,,ニューラルネットワーク
COVID-19,k.o.v.Id 19,,COVID-19
"""

# Vocabularyをインポート
response = transcribe.import_vocabulary(
    VocabularyName='medical-2026',
    LanguageCode='ja-JP',
    VocabularyFileUri='s3://my-bucket/vocabulary.txt',
    Tags=[
        {'Key': 'Domain', 'Value': 'Healthcare'}
    ]
)

# 文字起こしで使用
response = transcribe.start_transcription_job(
    TranscriptionJobName='medical-record',
    Media={'MediaFileUri': 's3://my-bucket/doctor-notes.mp3'},
    MediaFormat='mp3',
    LanguageCode='ja-JP',
    Settings={
        'VocabularyName': 'medical-2026'
    }
)

5. Custom Language Model（カスタムモデル）

# トレーニングデータ準備（テキスト）
training_data = """
抗菌薬治療を開始した患者の予後は良好である。
アナフィラキシーは即座にエピネフリン投与で対応する必要がある。
薬剤耐性菌による感染症は治療が困難である。
"""

# モデルトレーニング開始
response = transcribe.create_language_model(
    LanguageCode='ja-JP',
    BaseModelName='NarrowBand',  # NarrowBand(電話)/WideBand(高品質)
    ModelName='medical-model-2026',
    DataAccessRoleArn='arn:aws:iam::123456789:role/TranscribeRole',
    InputDataConfig={
        'S3Uri': 's3://my-bucket/training-data/',
        'TuningDataS3Uri': 's3://my-bucket/tuning-data/',
        'DataFormat': 'text'
    },
    Tags=[
        {'Key': 'Domain', 'Value': 'Medical'}
    ]
)

# トレーニング完了を確認
model_name = response['LanguageModel']['ModelName']
waiter = transcribe.get_waiter('language_model_ready')
waiter.wait(ModelName=model_name)

# 文字起こしで使用
response = transcribe.start_transcription_job(
    TranscriptionJobName='medical-transcription',
    Media={'MediaFileUri': 's3://my-bucket/medical-audio.mp3'},
    LanguageCode='ja-JP',
    Settings={
        'LanguageModelName': 'medical-model-2026'
    }
)

6. Call Analytics（コールセンター分析）

transcribe = boto3.client('transcribe', region_name='ap-northeast-1')

# Call Analytics ジョブ開始
response = transcribe.start_call_analytics_job(
    DataAccessRoleArn='arn:aws:iam::123456789:role/TranscribeRole',
    Media={'MediaFileUri': 's3://my-bucket/call-recording.wav'},
    OutputLocation='s3://my-bucket/call-analytics/',
    ContentRedaction={
        'RedactionType': 'PII',
        'RedactionOutput': 'redacted'
    }
)

job_name = response['CallAnalyticsJob']['CallAnalyticsJobName']

# ジョブ完了を待機
import time
while True:
    result = transcribe.get_call_analytics_job(CallAnalyticsJobName=job_name)
    if result['CallAnalyticsJob']['CallAnalyticsJobStatus'] == 'COMPLETED':
        break
    time.sleep(10)

# 結果の取得と分析
output_uri = result['CallAnalyticsJob']['Transcript']['TranscriptFileUri']

# Comprehend と連携して感情分析を追加
comprehend = boto3.client('comprehend', region_name='ap-northeast-1')
sentiment = comprehend.detect_sentiment(
    Text=transcript_text,
    LanguageCode='ja'
)

print(f"全体感情: {sentiment['Sentiment']}")
print(f"感情スコア: {sentiment['SentimentScore']}")

7. PII Redaction（個人情報マスキング）

# PII自動検出とマスキング
response = transcribe.start_transcription_job(
    TranscriptionJobName='customer-support-call',
    Media={'MediaFileUri': 's3://my-bucket/support-audio.mp3'},
    MediaFormat='mp3',
    LanguageCode='ja-JP',
    ContentRedaction={
        'RedactionType': 'PII',
        'RedactionOutput': 'redacted'  # 削除
    },
    Settings={
        'ShowSpeakerLabels': True,
        'MaxSpeakerLabels': 2
    }
)

# 出力例
"""
元: "私の電話番号は090-1234-5678です。クレジットカード番号は1234-5678-9012-3456です。"
出力: "私の電話番号は[PII]です。クレジットカード番号は[PII]です。"
"""

# 検出PIIタイプ一覧
pii_types = [
    'CREDIT_DEBIT_NUMBER',
    'PHONE_NUMBER',
    'EMAIL',
    'SSN',
    'ADDRESS',
    'NAME',
    'DATE',
    'ACCOUNT_NUMBER',
    'PIN',
    'DRIVER_ID',
    'PASSPORT_ID',
    'BANK_ACCOUNT_NUMBER'
]

8. Toxicity Detection（不適切な言葉検出）

# 不適切な言葉を自動検出
response = transcribe.start_transcription_job(
    TranscriptionJobName='call-with-toxicity-check',
    Media={'MediaFileUri': 's3://my-bucket/call.wav'},
    LanguageCode='ja-JP',
    Settings={
        'ToxicityDetection': [
            {
                'Severity': 'HIGH'  # HIGH/MEDIUM/LOW
            }
        ]
    }
)

# 出力に含まれるToxicity スコア
"""
toxicity_detection_results: [
    {
        "Toxicity": {
            "Sentiment": "NEGATIVE",
            "Confidence": 0.95
        },
        "Timestamps": {
            "Begin": "10.5",
            "End": "12.3"
        }
    }
]
"""

主要ユースケース

1. コールセンター品質管理

# 通話品質を自動スコアリング
import boto3

transcribe = boto3.client('transcribe')
comprehend = boto3.client('comprehend')

def quality_score_call(audio_uri):
    # Step 1: Call Analytics で分析
    response = transcribe.start_call_analytics_job(
        Media={'MediaFileUri': audio_uri},
        DataAccessRoleArn='arn:aws:iam::123456789:role/Role'
    )
    
    # Step 2: 感情スコア、解決済み、通話時間を取得
    # Step 3: ダッシュボードで可視化
    return {
        'duration': result['Duration'],
        'sentiment': result['SentimentBySegment'],
        'issue_resolved': result['IssueResolved'],
        'agent_score': result['AgentScore']
    }

2. 自動議事録生成

import boto3
import json

s3 = boto3.client('s3')
transcribe = boto3.client('transcribe')
bedrock = boto3.client('bedrock-runtime')

def generate_meeting_minutes(meeting_audio_s3_uri):
    # Step 1: Transcribe で文字起こし
    response = transcribe.start_transcription_job(
        TranscriptionJobName='meeting-minutes',
        Media={'MediaFileUri': meeting_audio_s3_uri},
        LanguageCode='ja-JP',
        Settings={'ShowSpeakerLabels': True, 'MaxSpeakerLabels': 10}
    )
    
    # Step 2: 文字起こし取得
    transcript = get_transcript_from_s3(response['TranscriptionJob']['Transcript']['TranscriptFileUri'])
    
    # Step 3: Bedrock Claude で要約・議事録生成
    prompt = f"""
    以下の会議の文字起こしから、以下の形式で議事録を作成してください：
    
    1. 議題
    2. 出席者
    3. 決定事項
    4. アクションアイテム（担当者・期限付き）
    5. 次回会議日時
    
    文字起こし:
    {transcript}
    """
    
    response = bedrock.invoke_model(
        modelId='anthropic.claude-3-sonnet-20240229-v1:0',
        body=json.dumps({'prompt': prompt, 'max_tokens': 2000})
    )
    
    return json.loads(response['body'].read())['completion']

3. 多言語字幕自動生成

import boto3

transcribe = boto3.client('transcribe')
translate = boto3.client('translate')
polly = boto3.client('polly')

def auto_subtitle_pipeline(video_s3_uri, languages=['en', 'ja', 'es']):
    # Step 1: 英語で文字起こし
    response = transcribe.start_transcription_job(
        TranscriptionJobName='video-transcription',
        Media={'MediaFileUri': video_s3_uri},
        LanguageCode='en-US',
        OutputBucketName='my-bucket'
    )
    
    # Step 2: 各言語に翻訳
    subtitles = {}
    for lang in languages:
        translated = translate.translate_text(
            Text=transcript,
            SourceLanguageCode='en',
            TargetLanguageCode=lang
        )
        subtitles[lang] = translated['TranslatedText']
    
    # Step 3: VTT字幕ファイル生成
    return generate_vtt_files(subtitles)

4. 医療診察記録の自動化

import boto3
import json

transcribe = boto3.client('transcribe')
comprehend = boto3.client('comprehend')

def medical_note_generation(doctor_audio):
    # Step 1: Medical Transcribe で医療用語を認識
    response = transcribe.start_transcription_job(
        TranscriptionJobName='medical-dictation',
        Media={'MediaFileUri': doctor_audio},
        LanguageCode='ja-JP',
        Specialty='HEALTHCARE',  # 医療特化
        ContentIdentification='PII'  # PHI自動マスキング
    )
    
    transcript = get_transcript(response)
    
    # Step 2: Comprehend Medical で医療エンティティ抽出
    response = comprehend.detect_medical_entities(
        Text=transcript,
        LanguageCode='ja'
    )
    
    # Step 3: 電子カルテフォーマット生成
    soap_note = {
        'subjective': extract_patient_complaints(transcript),
        'objective': extract_vital_signs(transcript),
        'assessment': extract_diagnosis(response),
        'plan': extract_treatment_plan(transcript)
    }
    
    return soap_note

5. リアルタイムライブ字幕

import asyncio
from amazon_transcribe.client import TranscribeStreamingClient

async def live_caption_service(websocket_connection):
    """ウェビナー・ライブストリーム用リアルタイム字幕"""
    
    client = TranscribeStreamingClient(region='ap-northeast-1')
    
    stream = await client.start_stream_transcription(
        language_code='ja-JP',
        media_sample_rate_hz=16000,
        media_encoding='pcm',
        show_speaker_label=True
    )
    
    async def broadcast_captions():
        async for transcript_event in stream:
            for result in transcript_event.transcript.results:
                # リアルタイムで WebSocket 経由で字幕配信
                await websocket_connection.send(json.dumps({
                    'caption': result.alternatives[0].transcript,
                    'is_final': not result.is_partial
                }))

6. コンタクトセンター自動ルーティング

import boto3

transcribe = boto3.client('transcribe')
connect = boto3.client('connect')
comprehend = boto3.client('comprehend')

def auto_routing_call(call_recording_uri):
    """通話内容から自動的に適切な部門にルーティング"""
    
    # Step 1: リアルタイム Transcribe で初期フレーズを認識
    # Step 2: キーフレーズ抽出で通話理由を判定
    response = comprehend.detect_key_phrases(
        Text=initial_transcript,
        LanguageCode='ja'
    )
    
    # Step 3: ルーティング先を決定
    keywords = extract_keywords(response)
    
    if '請求' in keywords:
        queue_id = 'billing_queue'
    elif 'テクニカル' in keywords:
        queue_id = 'technical_queue'
    else:
        queue_id = 'general_queue'
    
    # Step 4: Amazon Connect でキューに送信
    connect.put_contact_attributes(
        InstanceId='instance_id',
        InitialContactId='contact_id',
        Attributes={'routing_priority': queue_id}
    )

7. ポッドキャスト・オーディオコンテンツ処理

import boto3

transcribe = boto3.client('transcribe')
translate = boto3.client('translate')
comprehend = boto3.client('comprehend')

def podcast_processing_pipeline(audio_file_s3):
    """ポッドキャスト音声 → 字幕 + 文字起こし + トランスクリプト + 多言語"""
    
    # Step 1: 文字起こし（Speech Marks で時間情報も取得）
    response = transcribe.start_transcription_job(
        TranscriptionJobName='podcast-episode',
        Media={'MediaFileUri': audio_file_s3},
        LanguageCode='en-US',
        Settings={
            'ShowSpeakerLabels': True,
            'MaxSpeakerLabels': 5
        }
    )
    
    # Step 2: キーフレーズ抽出でエピソードのトピック抽出
    topics = comprehend.detect_key_phrases(
        Text=transcript,
        LanguageCode='en'
    )
    
    # Step 3: 多言語版を生成
    translations = {
        'ja': translate.translate_text(
            Text=transcript,
            SourceLanguageCode='en',
            TargetLanguageCode='ja'
        ),
        'es': translate.translate_text(
            Text=transcript,
            SourceLanguageCode='en',
            TargetLanguageCode='es'
        )
    }
    
    return {
        'transcript': transcript,
        'topics': topics,
        'translations': translations,
        'speaker_segments': response['SpeakerLabels']
    }

8. 顧客サポートチャット翻訳

import asyncio
import boto3

transcribe = boto3.client('transcribe')
translate = boto3.client('translate')
comprehend = boto3.client('comprehend')

async def multilingual_support_chat(customer_audio, customer_language):
    """多言語サポート: 顧客の音声 → テキスト → 日本語 → 感情分析"""
    
    # Step 1: 顧客音声を認識
    transcript = transcribe.start_transcription_job(...)
    
    # Step 2: 日本語に翻訳（サポート担当者用）
    translated = translate.translate_text(
        Text=transcript,
        SourceLanguageCode=customer_language,
        TargetLanguageCode='ja'
    )
    
    # Step 3: 感情分析で対応優先度判定
    sentiment = comprehend.detect_sentiment(
        Text=translated['TranslatedText'],
        LanguageCode='ja'
    )
    
    # Step 4: ネガティブな感情の場合はエスカレーション
    if sentiment['Sentiment'] == 'NEGATIVE':
        escalate_to_supervisor()
    
    return {
        'customer_transcript': transcript,
        'support_agent_text': translated['TranslatedText'],
        'sentiment': sentiment['Sentiment'],
        'priority': 'HIGH' if sentiment['Sentiment'] in ['NEGATIVE', 'MIXED'] else 'NORMAL'
    }

9. 法的文書の音声記録処理

import boto3

transcribe = boto3.client('transcribe')
comprehend = boto3.client('comprehend')

def legal_deposition_processing(audio_uri):
    """法的供述書の音声記録を自動処理"""
    
    # Step 1: PII保護付きで文字起こし
    response = transcribe.start_transcription_job(
        TranscriptionJobName='deposition',
        Media={'MediaFileUri': audio_uri},
        LanguageCode='en-US',
        ContentRedaction={
            'RedactionType': 'PII',
            'RedactionOutput': 'redacted_and_unredacted'  # 両方保持
        }
    )
    
    # Step 2: エンティティ抽出で関係者・日付を抽出
    entities = comprehend.detect_entities(
        Text=transcript,
        LanguageCode='en'
    )
    
    # Step 3: キーフレーズで重要な法的主張を抽出
    phrases = comprehend.detect_key_phrases(
        Text=transcript,
        LanguageCode='en'
    )
    
    return {
        'full_transcript': transcript,
        'redacted_transcript': redacted,
        'key_entities': filter_entities(entities, ['PERSON', 'ORGANIZATION', 'DATE']),
        'key_phrases': phrases,
        'timestamp_segments': response['SpeakerLabels']
    }

10. ライブイベント字幕配信

import boto3
import asyncio

class LiveEventCaptioning:
    def __init__(self, event_id):
        self.transcribe = boto3.client('transcribe')
        self.event_id = event_id
        self.captions_queue = asyncio.Queue()
    
    async def caption_stream(self, audio_stream_uri):
        """ライブイベント用キャプション配信"""
        
        stream = await self.transcribe.start_stream_transcription(
            language_code='en-US',
            media_sample_rate_hz=16000,
            media_encoding='pcm',
            show_speaker_label=True
        )
        
        async for event in stream:
            for result in event.transcript.results:
                caption_obj = {
                    'text': result.alternatives[0].transcript,
                    'is_final': not result.is_partial,
                    'timestamp': result.start_time,
                    'speaker': result.speaker_label if hasattr(result, 'speaker_label') else None
                }
                
                # キャプション配信（WebSocket/SSE/HLS）
                await self.captions_queue.put(caption_obj)
                
                if caption_obj['is_final']:
                    # 最終版をログに記録
                    log_final_caption(self.event_id, caption_obj)

設定・操作の具体例

CLI 例（5+）

1. バッチ文字起こしジョブ開始

aws transcribe start-transcription-job \
  --transcription-job-name 'my-job-2026' \
  --language-code 'ja-JP' \
  --media-format 'mp3' \
  --media '{"MediaFileUri":"s3://my-bucket/audio.mp3"}' \
  --output-bucket-name 'my-bucket' \
  --settings \
    'ShowSpeakerLabels=true,MaxSpeakerLabels=5,VocabularyName=medical-terms' \
  --region 'ap-northeast-1'

2. ジョブ進捗確認

aws transcribe get-transcription-job \
  --transcription-job-name 'my-job-2026' \
  --region 'ap-northeast-1' \
  --query 'TranscriptionJob.TranscriptionJobStatus' \
  --output text

3. Custom Vocabulary インポート

aws transcribe import-vocabulary \
  --vocabulary-name 'legal-2026' \
  --language-code 'ja-JP' \
  --vocabulary-file-uri 's3://my-bucket/vocabulary.txt' \
  --region 'ap-northeast-1'

4. Call Analytics ジョブ開始

aws transcribe start-call-analytics-job \
  --call-analytics-job-name 'call-analysis-001' \
  --media '{"MediaFileUri":"s3://my-bucket/call.wav"}' \
  --output-location 's3://my-bucket/analytics/' \
  --data-access-role-arn 'arn:aws:iam::123456789:role/TranscribeRole' \
  --region 'ap-northeast-1'

5. Language Model トレーニング開始

aws transcribe create-language-model \
  --language-code 'ja-JP' \
  --base-model-name 'NarrowBand' \
  --model-name 'medical-model-v1' \
  --input-data-config \
    'S3Uri=s3://my-bucket/training/,DataFormat=text' \
  --data-access-role-arn 'arn:aws:iam::123456789:role/TranscribeRole' \
  --region 'ap-northeast-1'

SDK 例（5+）

1. Python: リアルタイムストリーミング

import asyncio
from amazon_transcribe.client import TranscribeStreamingClient

async def transcribe_stream():
    client = TranscribeStreamingClient(region='ap-northeast-1')
    
    # サンプルレート 16kHz、PCM形式
    stream = await client.start_stream_transcription(
        language_code='ja-JP',
        media_sample_rate_hz=16000,
        media_encoding='pcm',
        show_speaker_label=True,
        max_speaker_labels=3
    )
    
    # マイク入力をシミュレート
    with open('input_audio.wav', 'rb') as f:
        async def write_chunks():
            for chunk in iter(lambda: f.read(4096), b''):
                await stream.input_stream.send_audio_event(audio_chunk=chunk)
            await stream.input_stream.end_stream()
        
        async def read_transcripts():
            async for transcript_event in stream:
                for result in transcript_event.transcript.results:
                    if not result.is_partial:
                        speaker = getattr(result, 'speaker_label', 'Unknown')
                        text = result.alternatives[0].transcript
                        print(f"[{speaker}] {text}")
        
        await asyncio.gather(write_chunks(), read_transcripts())

asyncio.run(transcribe_stream())

2. JavaScript/TypeScript: バッチジョブ監視

import { TranscribeClient, GetTranscriptionJobCommand, 
         StartTranscriptionJobCommand } from "@aws-sdk/client-transcribe";

async function monitorTranscriptionJob(jobName: string) {
    const client = new TranscribeClient({ region: 'ap-northeast-1' });
    
    const command = new GetTranscriptionJobCommand({ 
        TranscriptionJobName: jobName 
    });
    
    let jobStatus = 'IN_PROGRESS';
    
    while (jobStatus === 'IN_PROGRESS') {
        const response = await client.send(command);
        jobStatus = response.TranscriptionJob?.TranscriptionJobStatus || 'UNKNOWN';
        
        if (jobStatus === 'COMPLETED') {
            console.log(`✓ 文字起こし完了: ${response.TranscriptionJob?.Transcript?.TranscriptFileUri}`);
        } else if (jobStatus === 'FAILED') {
            console.error(`✗ エラー: ${response.TranscriptionJob?.FailureReason}`);
        }
        
        await new Promise(resolve => setTimeout(resolve, 5000));
    }
}

monitorTranscriptionJob('my-job-2026');

3. Go: バッチ処理パイプライン

package main

import (
	"context"
	"fmt"
	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/service/transcribe"
	"github.com/aws/aws-sdk-go-v2/service/transcribe/types"
)

func transcribeAudioFile(ctx context.Context, bucketName, objectKey string) {
	cfg, _ := config.LoadDefaultConfig(ctx, config.WithRegion("ap-northeast-1"))
	client := transcribe.NewFromConfig(cfg)
	
	input := &transcribe.StartTranscriptionJobInput{
		TranscriptionJobName: "job-golang",
		LanguageCode:         types.LanguageCodeJaJp,
		MediaFormat:          types.MediaFormatMp3,
		Media: &types.Media{
			MediaFileUri: fmt.Sprintf("s3://%s/%s", bucketName, objectKey),
		},
		Settings: &types.Settings{
			ShowSpeakerLabels: true,
			MaxSpeakerLabels:  5,
		},
		OutputBucketName: bucketName,
	}
	
	result, _ := client.StartTranscriptionJob(ctx, input)
	fmt.Printf("ジョブ開始: %s\n", *result.TranscriptionJob.TranscriptionJobName)
}

4. Java: Custom Vocabulary 管理

import software.amazon.awssdk.services.transcribe.TranscribeClient;
import software.amazon.awssdk.services.transcribe.model.*;

public class VocabularyManager {
    public static void importVocabulary(String vocabularyName, String s3Uri) {
        TranscribeClient client = TranscribeClient.builder()
            .region(Region.AP_NORTHEAST_1)
            .build();
        
        ImportVocabularyRequest request = ImportVocabularyRequest.builder()
            .vocabularyName(vocabularyName)
            .languageCode(LanguageCode.JA_JP)
            .vocabularyFileUri(s3Uri)
            .mergeStrategy(VocabularyFilterMethod.OVERWRITE)
            .build();
        
        try {
            ImportVocabularyResponse response = client.importVocabulary(request);
            System.out.println("Vocabulary imported: " + response.vocabulary().vocabularyName());
        } finally {
            client.close();
        }
    }
}

5. C#: Call Analytics 結果処理

using Amazon.TranscribeService;
using Amazon.TranscribeService.Model;
using System;
using System.Threading.Tasks;

class CallAnalyticsProcessor {
    public async Task ProcessCallAnalytics(string jobName) {
        var client = new AmazonTranscribeServiceClient(
            Amazon.RegionEndpoint.APNortheast1
        );
        
        var response = await client.GetCallAnalyticsJobAsync(
            new GetCallAnalyticsJobRequest {
                CallAnalyticsJobName = jobName
            }
        );
        
        if (response.CallAnalyticsJob.CallAnalyticsJobStatus == CallAnalyticsJobStatus.COMPLETED) {
            Console.WriteLine({{CONTENT}}quot;Transcript: {response.CallAnalyticsJob.Transcript.TranscriptFileUri}");
            Console.WriteLine({{CONTENT}}quot;Analytics: {response.CallAnalyticsJob.DataAccessRoleArn}");
        }
    }
}

IaC 例（CloudFormation/Terraform, 5+）

1. CloudFormation: Transcribe IAM ロール

AWSTemplateFormatVersion: '2010-09-09'
Resources:
  TranscribeRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: transcribe.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/AmazonS3FullAccess
      Policies:
        - PolicyName: TranscribePolicy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - transcribe:*
                  - kms:Decrypt
                Resource: '*'

  TranscribeOutputBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub 'transcribe-output-${AWS::AccountId}'
      VersioningConfiguration:
        Status: Enabled
      ServerSideEncryptionConfiguration:
        - ServerSideEncryptionByDefault:
            SSEAlgorithm: AES256

2. Terraform: Call Analytics パイプライン

resource "aws_transcribe_language_model" "medical" {
  language_code   = "ja-JP"
  base_model_name = "NarrowBand"
  model_name      = "medical-transcription-model"
  
  input_data_config {
    s3_uri      = aws_s3_object.training_data.id
    data_format = "text"
  }
  
  data_access_role_arn = aws_iam_role.transcribe_role.arn
  
  tags = {
    Environment = "production"
    Domain      = "healthcare"
  }
}

resource "aws_lambda_function" "call_analytics_processor" {
  filename      = "lambda_function.zip"
  function_name = "call-analytics-processor"
  role          = aws_iam_role.lambda_role.arn
  handler       = "index.handler"
  runtime       = "python3.11"
  timeout       = 60
  
  environment {
    variables = {
      TRANSCRIBE_ROLE = aws_iam_role.transcribe_role.arn
      OUTPUT_BUCKET   = aws_s3_bucket.output.id
    }
  }
}

3. CDK (Python): リアルタイムストリーミングセットアップ

from aws_cdk import (
    aws_lambda as lambda_,
    aws_s3 as s3,
    aws_iam as iam,
    aws_apigatewayv2 as apigw,
    core
)

class TranscribeStreamingStack(core.Stack):
    def __init__(self, scope: core.Construct, id: str, **kwargs):
        super().__init__(scope, id, **kwargs)
        
        # Lambda 関数でストリーミング処理
        streaming_handler = lambda_.Function(
            self, "StreamingHandler",
            runtime=lambda_.Runtime.PYTHON_3_11,
            code=lambda_.Code.from_asset("lambda"),
            handler="streaming.handler",
            timeout=core.Duration.seconds(300),
            memory_size=512,
            environment={
                "TRANSCRIBE_REGION": "ap-northeast-1"
            }
        )
        
        # IAM ポリシー
        streaming_handler.add_to_role_policy(
            iam.PolicyStatement(
                actions=[
                    "transcribe:*",
                    "s3:*"
                ],
                resources=["*"]
            )
        )
        
        # API Gateway で WebSocket エンドポイント公開
        api = apigw.HttpApi(
            self, "TranscribeAPI",
            protocol=apigw.HttpProtocol.HTTP2
        )

4. Terraform: Custom Vocabulary 管理

resource "aws_transcribe_vocabulary" "medical_terms" {
  vocabulary_name = "medical-vocabulary-2026"
  language_code   = "ja-JP"
  vocabulary_file_uri = "s3://${aws_s3_bucket.vocab.id}/medical-terms.txt"
  
  tags = {
    Domain = "Healthcare"
    Version = "2.0"
  }
}

resource "aws_s3_object" "vocab_file" {
  bucket = aws_s3_bucket.vocab.id
  key    = "medical-terms.txt"
  content = file("${path.module}/vocabularies/medical.txt")
}

5. CloudFormation: EventBridge 統合

Resources:
  TranscribeEventRule:
    Type: AWS::Events::Rule
    Properties:
      EventPattern:
        source:
          - aws.transcribe
        detail-type:
          - Transcribe Job State Change
        detail:
          status:
            - COMPLETED
      Targets:
        - Arn: !GetAtt ProcessCompletedJobsFunction.Arn
          RoleArn: !GetAtt EventBridgeRole.Arn

  ProcessCompletedJobsFunction:
    Type: AWS::Lambda::Function
    Properties:
      Runtime: python3.11
      Handler: index.handler
      Code:
        ZipFile: |
          import json
          import boto3
          
          transcribe = boto3.client('transcribe')
          comprehend = boto3.client('comprehend')
          
          def handler(event, context):
              job_name = event['detail']['TranscriptionJobName']
              # 処理ロジック
              return {'statusCode': 200}

比較表

ベストプラクティス

✅ すべき事

• 音声フォーマット最適化: MP3は圧縮なし、WAV/FLACは無圧縮で品質維持
• カスタム語彙の利用: 業界専門用語は Custom Vocabulary で精度を最大化
• PII自動マスキング: コンプライアンス要件に合わせて ContentRedaction 設定
• 並列処理の活用: 大量バッチは複数のTranscribe ジョブを並列実行
• CloudWatch Logs連携: すべての Transcribe ジョブを CloudWatch で監視
• IAM最小権限: Transcribe ロールは S3 と KMS のみに限定
• Speech Marks活用: 字幕同期のために Speech Marks を出力に含める
• エラーハンドリング: ジョブ失敗時の自動リトライと通知設定
• バージョン管理: Custom Vocabulary / Language Model は版管理

❌ してはいけない事

• 大容量ファイルの同期処理: 4時間以上は非同期バッチに限定
• マシンガン型API呼び出し: スロットリング回避のため呼び出し間隔を設定
• 本番環境でリトライなし: Transient エラーに対する指数バックオフ必須
• PII検出なしでのストレージ: GDPR/CCPA対応のため PHI/PII は必ずマスキング
• 言語設定なし: 言語自動検出は精度が低いため LanguageCode を常に指定
• デフォルト語彙のみ使用: 業界向けは Custom Vocabulary で精度向上必須
• ジョブ結果の長期キャッシュなし: S3 JSON は有効期限設定してコスト削減
• モニタリングなし: CloudWatch アラーム で失敗ジョブを自動検知
• スケーリング計画なし: 大量ジョブ時の同時実行数制限を事前に考慮

トラブルシューティング

2025-2026最新動向

1. Generative Voice（生成音声）

• 2026年: Transcribe + Polly 連携で自然な音声復元
• ユースケース: 亡くなった著名人の音声再現

2. リアルタイム多言語処理

• Transcribe + Translate + Polly のエンドツーエンド統合
• 同時翻訳字幕配信が標準化

3. Bedrock連携強化

• Claude でTranscribe結果を自動要約・ラベリング
• Call Analytics + Comprehend + Bedrock による深い洞察抽出

4. Industry-Specific Models

• Financial（金融用語）/ Legal（法律用語）モデル追加予定
• 垂直領域専用 Custom Model の自動提案

5. Streaming API 高度化

• 音声の話者変更をリアルタイムで検知・通知
• 感情スコアのストリーミング出力

6. マルチモーダル処理

• Textract でドキュメント読み込み + Transcribe で音声説明 → 統合インデックス
• Vision API (Rekognition) との相互連携

7. オンデバイス処理

• Lambda@Edge での軽量 ASR（低遅延）
• IoT デバイス向けのコンパクト Transcribe エンジン

8. ビデオ処理統合

• Transcribe + Rekognition Video の統合分析
• 動画内の複数シーンを自動分割・字幕付与

学習リソース・参考文献

公式ドキュメント（8+）

1. Amazon Transcribe Developer Guide
2. Transcribe API Reference
3. Supported Languages in Transcribe
4. Call Analytics Feature Details
5. Custom Vocabulary Guide
6. Custom Language Model Setup
7. Transcribe Medical Documentation
8. Pricing and Usage Examples

OSS・ベンダー リソース（5+）

1. AssemblyAI Python SDK - AssemblyAI ASR
2. Deepgram Python SDK - Deepgram 音声認識
3. OpenAI Whisper - オープンソース音声認識
4. speech-recognition - 複数 ASR バックエンド統合
5. pydub - Python 音声処理ライブラリ

AWS機械学習ブログ（参考記事）

• Building Multilingual Voice AI with Transcribe + Translate
• Call Center Analytics with Transcribe Call Analytics

実装例・チェックリスト

実装例: 完全な Call Center Pipeline

import boto3
import json
from datetime import datetime

class CallCenterAnalyticsPipeline:
    def __init__(self):
        self.transcribe = boto3.client('transcribe', region_name='ap-northeast-1')
        self.comprehend = boto3.client('comprehend', region_name='ap-northeast-1')
        self.dynamodb = boto3.resource('dynamodb', region_name='ap-northeast-1')
        self.table = self.dynamodb.Table('CallAnalytics')
    
    def process_call_recording(self, recording_s3_uri, call_id):
        """通話録音を分析して DynamoDB に保存"""
        
        # Step 1: Transcribe Call Analytics で分析
        response = self.transcribe.start_call_analytics_job(
            CallAnalyticsJobName=f'call-{call_id}-{datetime.now().timestamp()}',
            Media={'MediaFileUri': recording_s3_uri},
            DataAccessRoleArn='arn:aws:iam::123456789:role/TranscribeRole',
            OutputLocation='s3://my-bucket/call-analytics/'
        )
        
        job_name = response['CallAnalyticsJob']['CallAnalyticsJobName']
        
        # Step 2: ジョブ完了を待機
        import time
        while True:
            result = self.transcribe.get_call_analytics_job(CallAnalyticsJobName=job_name)
            if result['CallAnalyticsJob']['CallAnalyticsJobStatus'] == 'COMPLETED':
                break
            time.sleep(10)
        
        # Step 3: 結果を Comprehend で追加分析
        transcript = get_transcript_from_s3(
            result['CallAnalyticsJob']['Transcript']['TranscriptFileUri']
        )
        
        sentiment = self.comprehend.detect_sentiment(
            Text=transcript,
            LanguageCode='ja'
        )
        
        entities = self.comprehend.detect_entities(
            Text=transcript,
            LanguageCode='ja'
        )
        
        # Step 4: DynamoDB に保存
        self.table.put_item(
            Item={
                'call_id': call_id,
                'timestamp': int(datetime.now().timestamp()),
                'transcript_uri': result['CallAnalyticsJob']['Transcript']['TranscriptFileUri'],
                'sentiment': sentiment['Sentiment'],
                'sentiment_scores': sentiment['SentimentScore'],
                'entities': [e['Text'] for e in entities['Entities']],
                'duration_seconds': result['CallAnalyticsJob']['CallDurationSeconds'],
                'quality_score': result['CallAnalyticsJob'].get('ComplianceScore', 0)
            }
        )
        
        return {
            'call_id': call_id,
            'status': 'completed',
            'sentiment': sentiment['Sentiment'],
            'key_entities': [e['Text'] for e in entities['Entities']]
        }

採用判断チェックリスト

• [ ] 音声→テキスト変換が必要か
• [ ] リアルタイム（WebSocket）またはバッチ（S3）のいずれか選択
• [ ] 複数話者の識別が必要か（ShowSpeakerLabels=true）
• [ ] 業界専門用語を正確に認識する必要か（Custom Vocabulary）
• [ ] 個人情報を保護する必要か（ContentRedaction=PII）
• [ ] コールセンター感情分析が必要か（Call Analytics）
• [ ] 医療用語対応が必要か（Transcribe Medical）
• [ ] マルチ言語サポートが必要か
• [ ] リアルタイム字幕配信が必要か
• [ ] 自動議事録生成が必要か（Bedrock 連携）

まとめ

Transcribe は 「音声をテキストに変換する自動音声認識サービス」。バッチ・リアルタイムのいずれにも対応し、話者識別・カスタム語彙・PII マスキング・Call Analytics で、コールセンター・会議・医療・法律分野のエンタープライズグレードの音声AI処理を実現します。Translate・Polly・Comprehend・Bedrock と組み合わせた多言語音声パイプラインの中核として活用できます。

最終更新：2026-04-26 バージョン：v2.0