Azure 语音转文本 REST API Python 实现

azure-speech-to-text-rest-py
分类编程
作者Agentic Awesome Skills 社区
许可MIT
评分4.90/5
使用13.6K

Azure 语音转文本 REST API (短音频)

一个简单的 REST API,用于短音频文件(最多 60 秒)的语音转文本转录。无需 SDK,仅需 HTTP 请求。

前置条件

1. Azure 订阅 - 免费创建
2. 语音资源 - 在 Azure 门户 中创建
3. 获取凭据 - 部署后,前往 资源 > 密钥和终结点

环境变量

bash
# 必填
AZURE_SPEECH_KEY=<your-speech-resource-key>
AZURE_SPEECH_REGION=<region>  # 例如 eastus, westus2, westeurope

备选:直接使用终结点

AZURE_SPEECH_ENDPOINT=https://<region>.stt.speech.microsoft.com

安装

bash
pip install requests

快速上手

python
import os
import requests

def transcribe_audio(audio_file_path: str, language: str = "en-US") -> dict:
"""使用 REST API 转录短音频文件(最多 60 秒)。"""
region = os.environ["AZURE_SPEECH_REGION"]
api_key = os.environ["AZURE_SPEECH_KEY"]

url = f"https://{region}.stt.speech.microsoft.com/speech/recognition/conversation/cognitiveservices/v1"

headers = {
"Ocp-Apim-Subscription-Key": api_key,
"Content-Type": "audio/wav; codecs=audio/pcm; samplerate=16000",
"Accept": "application/json"
}

params = {
"language": language,
"format": "detailed" # 或 "simple"
}

with open(audio_file_path, "rb") as audio_file:
response = requests.post(url, headers=headers, params=params, data=audio_file)

response.raise_for_status()
return response.json()

使用示例

result = transcribe_audio("audio.wav", "en-US") print(result["DisplayText"])

音频要求

| 格式 | 编解码器 | 采样率 | 备注 |
|--------|-------|-------------|-------|
| WAV | PCM | 16 kHz, 单声道 | 推荐 |
| OGG | OPUS | 16 kHz, 单声道 | 文件体积更小 |

限制条件:

  • 音频最长 60 秒

  • 发音评估:最长 30 秒

  • 不支持部分/中间结果(仅返回最终结果)

Content-Type 请求头

python
# WAV PCM 16kHz
"Content-Type": "audio/wav; codecs=audio/pcm; samplerate=16000"

OGG OPUS

"Content-Type": "audio/ogg; codecs=opus"

响应格式

简单格式 (默认)

python
params = {"language": "en-US", "format": "simple"}
json
{
  "RecognitionStatus": "Success",
  "DisplayText": "Remind me to buy 5 pencils.",
  "Offset": "1236645672289",
  "Duration": "1236645672289"
}

详细格式

python
params = {"language": "en-US", "format": "detailed"}
json
{
  "RecognitionStatus": "Success",
  "Offset": "1236645672289",
  "Duration": "1236645672289",
  "NBest": [
    {
      "Confidence": 0.9052885,
      "Display": "What's the weather like?",
      "ITN": "what's the weather like",
      "Lexical": "what's the weather like",
      "MaskedITN": "what's the weather like"
    }
  ]
}

分块传输 (推荐)

为了降低延迟,可以分块流式传输音频:

python
import os
import requests

def transcribe_chunked(audio_file_path: str, language: str = "en-US") -> dict:
"""分块流式传输音频以降低延迟"""


wer 延迟。"""
region = os.environ["AZURE_SPEECH_REGION"]
api_key = os.environ["AZURE_SPEECH_KEY"]

url = f"https://{region}.stt.speech.microsoft.com/speech/recognition/conversation/cognitiveservices/v1"

headers = {
"Ocp-Apim-Subscription-Key": api_key,
"Content-Type": "audio/wav; codecs=audio/pcm; samplerate=16000",
"Accept": "application/json",
"Transfer-Encoding": "chunked",
"Expect": "100-continue"
}

params = {"language": language, "format": "detailed"}

def generate_chunks(file_path: str, chunk_size: int = 1024):
with open(file_path, "rb") as f:
while chunk := f.read(chunk_size):
yield chunk

response = requests.post(
url,
headers=headers,
params=params,
data=generate_chunks(audio_file_path)
)

response.raise_for_status()
return response.json()
code
## 身份验证选项

选项 1:订阅密钥(简单)

python headers = { "Ocp-Apim-Subscription-Key": os.environ["AZURE_SPEECH_KEY"] }
code
### 选项 2:Bearer 令牌
python import requests import os

def get_access_token() -> str:
"""从令牌端点获取访问令牌。"""
region = os.environ["AZURE_SPEECH_REGION"]
api_key = os.environ["AZURE_SPEECH_KEY"]

token_url = f"https://{region}.api.cognitive.microsoft.com/sts/v1.0/issueToken"

response = requests.post(
token_url,
headers={
"Ocp-Apim-Subscription-Key": api_key,
"Content-Type": "application/x-www-form-urlencoded",
"Content-Length": "0"
}
)
response.raise_for_status()
return response.text

在请求中使用令牌(有效期 10 分钟)

token = get_access_token() headers = { "Authorization": f"Bearer {token}", "Content-Type": "audio/wav; codecs=audio/pcm; samplerate=16000", "Accept": "application/json" }
code
## 查询参数

| 参数 | 必填 | 取值 | 描述 |
|-----------|----------|--------|-------------|
| language | | en-US, de-DE 等 | 语音语言 |
| format | 否 | simple, detailed | 结果格式(默认:simple) |
| profanity | 否 | masked, removed, raw | 脏话处理(默认:masked) |

识别状态值

| 状态 | 描述 |
|--------|-------------|
| Success | 识别成功 |
| NoMatch | 检测到语音但未匹配到词汇 |
| InitialSilenceTimeout | 仅检测到静音 |
| BabbleTimeout | 仅检测到噪音 |
| Error | 内部服务错误 |

脏话处理

python

使用星号掩盖脏话(默认)

params = {"language": "en-US", "profanity": "masked"}

完全移除脏话

params = {"language": "en-US", "profanity": "removed"}

原样保留脏话

params = {"language": "en-US", "profanity": "raw"}
code
## 错误处理
python import requests

def transcribe_with_error_handling(audio_path: str, language: str = "en-US") -> dict | None:
"""带有适当错误处理的转录函数。"""
region = os.environ["AZURE_SPEECH_REGION"]
api_key = os.environ["AZURE_SPEECH_KEY"]

url = f"https://{region}.stt.speech.microsoft.com/speech/recognition/conversation/cognitiveservices/v1"

try:
with open(audio_path, "rb") as audio_file:
response = requests.post(
url,
headers={
"Ocp-Apim-Subscrip

code
tion-Key": api_key,
"Content-Type": "audio/wav; codecs=audio/pcm; samplerate=16000",
"Accept": "application/json"
},
params={"language": language, "format": "detailed"},
data=audio_file
)

if response.status_code == 200:
result = response.json()
if result.get("RecognitionStatus") == "Success":
return result
else:
print(f"Recognition failed: {result.get('RecognitionStatus')}")
return None
elif response.status_code == 400:
print(f"Bad request: Check language code or audio format")
elif response.status_code == 401:
print(f"Unauthorized: Check API key or token")
elif response.status_code == 403:
print(f"Forbidden: Missing authorization header")
else:
print(f"Error {response.status_code}: {response.text}")

return None

except requests.exceptions.RequestException as e:
print(f"Request failed: {e}")
return None

异步版本

python
import os
import aiohttp
import asyncio

async def transcribe_async(audio_file_path: str, language: str = "en-US") -> dict:
"""使用 aiohttp 的异步版本。"""
region = os.environ["AZURE_SPEECH_REGION"]
api_key = os.environ["AZURE_SPEECH_KEY"]

url = f"https://{region}.stt.speech.microsoft.com/speech/recognition/conversation/cognitiveservices/v1"

headers = {
"Ocp-Apim-Subscription-Key": api_key,
"Content-Type": "audio/wav; codecs=audio/pcm; samplerate=16000",
"Accept": "application/json"
}

params = {"language": language, "format": "detailed"}

async with aiohttp.ClientSession() as session:
with open(audio_file_path, "rb") as f:
audio_data = f.read()

async with session.post(url, headers=headers, params=params, data=audio_data) as response:
response.raise_for_status()
return await response.json()

使用示例

result = asyncio.run(transcribe_async("audio.wav", "en-US")) print(result["DisplayText"])

支持的语言

常用语言代码(详见 完整列表):

| 代码 | 语言 |
|------|----------|
| en-US | 英语 (美国) |
| en-GB | 英语 (英国) |
| de-DE | 德语 |
| fr-FR | 法语 |
| es-ES | 西班牙语 (西班牙) |
| es-MX | 西班牙语 (墨西哥) |
| zh-CN | 中文 (普通话) |
| ja-JP | 日语 |
| ko-KR | 韩语 |
| pt-BR | 葡萄牙语 (巴西) |

最佳实践

1. 使用 WAV PCM 16kHz 单声道 以获得最佳兼容性
2. 启用分块传输 (chunked transfer) 以降低延迟
3. 缓存访问令牌 9 分钟(有效期为 10 分钟)
4. 指定正确的语言 以确保识别准确
5. 需要置信度分数时 请使用 detailed 格式
6. 在生产代码中 处理所有 RecognitionStatus 的取值

何时不要使用此 API

在需要以下功能时,请改用 Speech SDK 或 批量转录 API:

  • 音频长度超过 60 秒
  • 实时流式转录
  • 部分/中间结果
  • 语音翻译
  • 自定义语音模型
  • 大量文件的批量转录

参考文件

| 文件 | 内容 |
|------|----------|
| references/pronunciation-assessment.md | 发音评估参数与评分 |

适用场景

此技能适用于 以执行概览中所描述的工作流或操作。

局限性

  • 仅在任务明确符合上述范围时使用此技能。
  • 不要将输出结果视为针对特定环境的验证、测试或专家评审的替代方案。
  • 如果缺少必要的输入、权限、安全边界或成功标准,请停止操作并请求澄清。