deepgram-python-speech-to-text

Deepgram Speech-to-Text を使用する（Python SDK）

/v1/listen を介して、録音済み音声の基本的な文字起こし（ASR）（REST）またはライブ音声の文字起こし（WebSocket）を行います。

この製品を使用する場面

• REST（transcribe_url / transcribe_file） — ファイル全体または URL を一度だけ文字起こしします。バッチジョブ、字幕作成のパイプライン、オフライン分析に使用します。
• WebSocket（listen.v1.connect） — 継続的にストリーミング文字起こしします。ライブ字幕、リアルタイムのマイク入力、通話音声に使用します。

次の場合は、別のスキルを使用してください。

• 音声の要約、感情分析、トピック、インテント、話者分離、またはマスキングが必要な場合 → deepgram-python-audio-intelligence（同じエンドポイントで、パラメーターが異なります）。
• ターン交代／ターン終了イベントが必要な場合 → deepgram-python-conversational-stt（v2 / Flux）。
• 全二重の対話型アシスタント（STT + LLM + TTS + 関数呼び出し）が必要な場合 → deepgram-python-voice-agent。

認証

import os
from dotenv import load_dotenv
load_dotenv()

from deepgram import DeepgramClient

client = DeepgramClient()  # reads DEEPGRAM_API_KEY from env
# or: DeepgramClient(api_key=os.environ["DEEPGRAM_API_KEY"])

すべてのリクエストで送信するヘッダー：Authorization: Token <api_key>（Bearer ではありません）。

クイックスタート — REST（録音済み URL）

response = client.listen.v1.media.transcribe_url(
    url="https://dpgr.am/spacewalk.wav",
    model="nova-3",
    smart_format=True,
    punctuate=True,
)
transcript = response.results.channels[0].alternatives[0].transcript

クイックスタート — REST（録音済みファイル）

with open("audio.wav", "rb") as f:
    audio_bytes = f.read()

response = client.listen.v1.media.transcribe_file(
    request=audio_bytes,
    model="nova-3",
)

request= には、生の bytes または bytes のイテレーターを指定できます（大きなファイルをチャンクごとにストリーミングできます）。ファイルハンドルは渡さないでください。

クイックスタート — WebSocket（中間結果を含むライブストリーミング）

ライブ文字起こしでは、中間（部分）結果と確定結果が出力されます。interim_results=True を渡し、is_final を確認することで、部分テキストをリアルタイムで表示し、話者が間を置いたら最終的な文字起こし結果で上書きできます。

import threading
from deepgram.core.events import EventType
from deepgram.listen.v1.types import (
    ListenV1Results, ListenV1Metadata,
    ListenV1SpeechStarted, ListenV1UtteranceEnd,
)

with client.listen.v1.connect(
    model="nova-3",
    interim_results=True,    # ← emit partial results while user is still speaking
    utterance_end_ms=1000,   # silence (ms) before server emits UtteranceEnd
    vad_events=True,         # SpeechStarted events
    smart_format=True,
) as conn:
    # Mutable container so the on_message closure can update state without `global`
    state = {"last_interim_len": 0}

    def on_message(m):
        if isinstance(m, ListenV1Results) and m.channel and m.channel.alternatives:
            transcript = m.channel.alternatives[0].transcript
            if not transcript:
                return
            if m.is_final:
                # Final segment: overwrite the running interim line, newline if utterance ended
                pad = " " * max(0, state["last_interim_len"] - len(transcript))
                end = "\n" if m.speech_final else ""
                print(f"\r{transcript}{pad}", end=end, flush=True)
                state["last_interim_len"] = 0
            else:
                # Interim: keep overwriting the same console line as the user speaks
                print(f"\r{transcript}", end="", flush=True)
                state["last_interim_len"] = len(transcript)
        elif isinstance(m, ListenV1UtteranceEnd):
            print()  # newline; UtteranceEnd fires after final results when audio goes silent
        elif isinstance(m, ListenV1SpeechStarted):
            pass  # optional: reset UI when a new utterance begins

    conn.on(EventType.OPEN,    lambda _: print("connected"))
    conn.on(EventType.MESSAGE, on_message)
    conn.on(EventType.CLOSE,   lambda _: print("\nclosed"))
    conn.on(EventType.ERROR,   lambda e: print(f"\nerr: {e}"))

    # Start receive loop in background so we can send concurrently
    threading.Thread(target=conn.start_listening, daemon=True).start()

    for chunk in audio_chunks:         # raw PCM bytes at declared encoding/sample_rate
        conn.send_media(chunk)

    conn.send_finalize()               # flush final partial before closing

中間／確定フラグの意味

• is_final = False — 中間の推定結果です。修正される可能性があります。確定前のスタイル（薄い色、斜体）で表示し、次のメッセージが届いたら上書きします。
• is_final = True, speech_final = False — 確定したセグメントですが、話者はまだ話しています。文字起こしに追加してください。後続の確定結果が届きます。
• is_final = True, speech_final = True — セグメントが確定し、発話も終了しました（無音を検出）。行を確定して、新しい行を開始します。
• from_finalize = True — 明示的な send_finalize() 呼び出しによって生成された確定結果です（自然なエンドポイント検出ではありません）。「フラッシュを要求した」のか「話者が間を置いた」のかを区別するのに役立ちます。

send_finalize() を送信すると、サーバーは確定結果をただちに出力します（例：ユーザーが「停止」をクリックした場合）。正常に終了するには、send_finalize の後に send_close_stream() を送信します。

WSS のメッセージタイプは deepgram.listen.v1.types にあります。

非同期版

from deepgram import AsyncDeepgramClient
client = AsyncDeepgramClient()

response = await client.listen.v1.media.transcribe_url(url=..., model="nova-3")

async with client.listen.v1.connect(model="nova-3") as conn:
    # same .on(...) handlers, then:
    await conn.start_listening()

非同期／遅延結果のパターン

「非同期」には明確に異なる意味が2つあります。混同しないでください。

1. Python の async/await（同期スタイル、即時に結果を取得）

AsyncDeepgramClient は Awaitable[<full response>] を返します。結果は後からではなく、await したときに返されます。FastAPI、aiohttp、または任意の asyncio アプリへの統合に使用します。

import asyncio
from deepgram import AsyncDeepgramClient

client = AsyncDeepgramClient()

async def transcribe(url: str) -> str:
    response = await client.listen.v1.media.transcribe_url(
        url=url,
        model="nova-3",
        smart_format=True,
    )
    # `response` is the FULL transcription — no polling, no callback, just await.
    return response.results.channels[0].alternatives[0].transcript

text = asyncio.run(transcribe("https://dpgr.am/spacewalk.wav"))

2. コールバック URL による遅延処理（Webhook、結果は後で POST）

callback="https://your.app/webhook" を渡すと、リクエストは request_id を返してすぐに終了します。Deepgram はバックグラウンドで音声を処理し、最終結果を Webhook URL に POST します。ポーリング用のエンドポイントはありません。結果を受け取るには、サーバーにアクセスできる状態にしておく必要があります。

response = client.listen.v1.media.transcribe_url(
    url="https://dpgr.am/spacewalk.wav",
    callback="https://your.app/deepgram-webhook",
    callback_method="POST",         # or "PUT"
    model="nova-3",
    smart_format=True,
)
print(f"Accepted; tracking id: {response.request_id}")
# response is a "listen accepted" — NOT the transcript. Wait for your webhook.

Webhook には、同期的な transcribe_url 呼び出しで受け取るものと同じ JSON 本文が送信されます。非常に長いファイルを処理する場合や、リクエストを開いたままにしたくない場合に使用します。

パターン	戻り値	使用する場面
client.listen.v1.media.transcribe_url(...)	文字起こし結果全体を同期的に返します	約 10 分までのファイル。HTTP タイムアウトの制限があります
await AsyncDeepgramClient().listen.v1.media.transcribe_url(...)	文字起こし結果全体をノンブロッキングで返します	asyncio アプリ内
transcribe_url(..., callback="https://...")	すぐに {request_id} を返し、文字起こし結果は後で Webhook に POST されます	非常に長いファイル。長時間の HTTP 接続を維持する必要がありません
client.listen.v1.connect(...)（WebSocket）	音声の送信に応じてストリーミングイベントを返します	ライブ音声（マイク、電話）

動作するコールバックの例については、examples/12-transcription-prerecorded-callback.py を参照してください。

主なパラメーター

model、language、encoding、sample_rate、channels、multichannel、punctuate、smart_format、diarize、endpointing、interim_results、utterance_end_ms、vad_events、keywords、search、redact、numerals、paragraphs、utterances。

API リファレンス（階層別）

1. リポジトリ内の Fern 生成リファレンス：reference.md — 「Listen V1 Media」（REST）および「Listen V1 Connect」（WSS）のセクション。
2. 正規の OpenAPI（REST）：https://developers.deepgram.com/openapi.yaml
3. 正規の AsyncAPI（WSS）：https://developers.deepgram.com/asyncapi.yaml
4. Context7 — Deepgram のドキュメント全体を対象にした自然言語クエリ。ライブラリ ID：/llmstxt/developers_deepgram_llms_txt。
5. 製品ドキュメント：
   • https://developers.deepgram.com/reference/speech-to-text/listen-pre-recorded
   • https://developers.deepgram.com/reference/speech-to-text/listen-streaming

注意点

1. 認証情報の種類に合った認証方式を使用してください。 API キーには Authorization: Token <api_key> を使用します。一時トークン／アクセストークン（client.auth.v1.tokens.grant() または同等のサーバーから取得）には Authorization: Bearer <access_token> を使用します。カスタムの DeepgramClient は、access_token=... を渡すと Bearer のオーバーライドを設定します（src/deepgram/client.py を参照）。長期間有効な API キーに Bearer <api_key> を付けると失敗します。
2. 音声に合わせてエンコーディングを指定してください。 encoding="linear16" と指定しながら Opus を送信すると、出力が文字化けするか、400 エラーになります。
3. ストリームを適切に終了してください。 WSS コンテキストを終了する前に send_finalize() を呼び出してください。そうしないと、最後の途中結果が失われます。
4. 長時間の WSS セッションでは KeepAlive を送信してください。 約 10 秒以上アイドル状態が続くと、サーバーが接続を閉じます。KeepAlive メッセージまたは音声チャンクを送信してください。
5. Intelligence 機能は REST 専用です。 summarize、topics、intents、sentiment、detect_language は WSS では動作しません。deepgram-python-audio-intelligence を参照してください。
6. transcribe_file(request=...) が受け取るのはバイト列またはイテレーターであり、ファイルハンドルではありません。
7. 現在の主力 STT モデルは nova-3 です。 最新のモデル一覧は client.manage.v1.models.list() で確認してください。
8. 同期版の connection.start_listening() はブロッキング処理です。 音声を同時に送信できるよう、同期処理ではスレッドで、非同期処理ではタスクとして実行してください。

このリポジトリ内のサンプルファイル

• examples/10-transcription-prerecorded-url.py
• examples/11-transcription-prerecorded-file.py
• examples/12-transcription-prerecorded-callback.py
• examples/13-transcription-live-websocket.py
• tests/wire/test_listen_v1_media.py — ワイヤーレベルのフィクスチャ
• tests/manual/listen/v1/connect/main.py — ライブ WSS 接続テスト

中核となるプロダクトスキル

言語横断のDeepgram製品知識（統合されたAPIリファレンス、ドキュメント検索、実行可能な特化型レシピ、サードパーティ連携の例、MCPのセットアップ）を利用するには、中核スキルをインストールしてください。

npx skills add deepgram/skills

このSDKには各言語に適したコードスキルが含まれています。deepgram/skillsには言語横断の製品知識が含まれています（api、docs、recipes、examples、starters、setup-mcpを参照）。