deepgram-python-speech-to-text

使用 Deepgram 语音转文本（Python SDK）

通过 /v1/listen 转录预录制音频（REST）或实时音频（WebSocket）的基础语音识别（ASR）。

何时使用此产品

• REST（transcribe_url / transcribe_file） — 一次性转录完整文件或 URL。适用于批处理任务、字幕流水线和离线分析。
• WebSocket（listen.v1.connect） — 连续流式转录。适用于实时字幕、实时麦克风输入和电话音频。

以下情况请使用其他 skill：

• 如果需要对音频生成摘要、情感、主题、意图、说话人分离或脱敏结果 → 使用 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()

异步 / 延迟结果模式

“异步”有两种不同的含义，请勿混淆。

1. Python async/await（同步风格，即时返回结果）

AsyncDeepgramClient 会返回 Awaitable[<完整响应>]。在你执行 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，稍后发送结果）

传入 callback="https://your.app/webhook"，请求会立即返回一个 request_id。Deepgram 会在后台处理音频，并将最终结果 POST 到你的 Webhook URL。没有轮询端点——你的服务器必须可访问，才能接收结果。

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}，稍后将转录结果 POST 到 Webhook	超长文件；无需长时间保持 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>——传入 access_token=... 时，自定义 DeepgramClient 会安装 Bearer 覆盖项（参见 src/deepgram/client.py）。使用长期有效的 API 密钥时发送 Bearer <api_key> 会导致失败。
2. 编码必须与音频匹配。 声明 encoding="linear16" 却发送 Opus，会导致输出乱码或返回 400 错误。
3. 正确关闭流。 退出 WSS 上下文前调用 send_finalize()，否则最后一段部分结果会丢失。
4. 在较长的 WSS 会话中发送保活消息。 若空闲时间超过约 10 秒，服务端会关闭连接。请发送 KeepAlive 消息或音频块。
5. 智能功能仅支持 REST。 summarize、topics、intents、sentiment、detect_language 均不支持通过 WSS 使用——参见 deepgram-python-audio-intelligence。
6. transcribe_file(request=...) 接受字节或迭代器，而不是文件句柄。
7. nova-3 是当前的旗舰 STT 模型。 可通过 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）。