azure-speech-to-text-rest-py

随附的上游支持文件

支持文件位于相对于技能的路径中。有关其固定来源、许可证、哈希值和导入变更，请参阅 UPSTREAM-NOTICE.md。

用于短音频的 Azure Speech to Text REST API

用于转录短音频文件（最长 60 秒）的简单 REST API。无需 SDK，只需发送 HTTP 请求。

先决条件

1. Azure 订阅 — 免费创建
2. Speech 资源 — 在 Azure Portal 中创建
3. 获取凭据 — 部署后，前往资源 > Keys and Endpoint

环境变量

# Required
AZURE_SPEECH_KEY=<your-speech-resource-key>
AZURE_SPEECH_REGION=<region>  # e.g., eastus, westus2, westeurope

# Alternative: Use endpoint directly
AZURE_SPEECH_ENDPOINT=https://<region>.stt.speech.microsoft.com

安装

pip install requests

快速开始

import os
import requests

def transcribe_audio(audio_file_path: str, language: str = "en-US") -> dict:
    """Transcribe short audio file (max 60 seconds) using REST API."""
    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"  # or "simple"
    }
    
    with open(audio_file_path, "rb") as audio_file:
        response = requests.post(url, headers=headers, params=params, data=audio_file, timeout=(10, 90))
    
    response.raise_for_status()
    return response.json()

# Usage
result = transcribe_audio("audio.wav", "en-US")
if result.get("RecognitionStatus") == "Success" and result.get("NBest"):
    print(result["NBest"][0].get("Display", ""))
else:
    print("No successful detailed recognition result")

音频要求

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

限制：

• 音频最长 60 秒
• 进行发音评估时，音频最长 30 秒
• 不提供部分／中间结果（仅提供最终结果）

Content-Type 标头

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

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

响应格式

简单格式（默认）

params = {"language": "en-US", "format": "simple"}

{
  "RecognitionStatus": "Success",
  "DisplayText": "Remind me to buy 5 pencils.",
  "Offset": "1236645672289",
  "Duration": "1236645672289"
}

详细格式

params = {"language": "en-US", "format": "detailed"}

{
  "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"
    }
  ]
}

分块传输（推荐）

如需降低延迟，可分块传输音频：

import os
import requests

def transcribe_chunked(audio_file_path: str, language: str = "en-US") -> dict:
    """Stream audio in chunks for lower latency."""
    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),
        timeout=(10, 90)
    )
    
    response.raise_for_status()
    return response.json()

身份验证选项

选项 1：订阅密钥（简单）

headers = {
    "Ocp-Apim-Subscription-Key": os.environ["AZURE_SPEECH_KEY"]
}

选项 2：Bearer 令牌

import requests
import os

def get_access_token() -> str:
    """Get access token from the token endpoint."""
    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"
        },
        timeout=(10, 30)
    )
    response.raise_for_status()
    return response.text

# Use token in requests (valid for 10 minutes)
token = get_access_token()
headers = {
    "Authorization": f"Bearer {token}",
    "Content-Type": "audio/wav; codecs=audio/pcm; samplerate=16000",
    "Accept": "application/json"
}

查询参数

参数	必填	值	说明
language	是	en-US、de-DE 等	语音所用语言
format	否	simple、detailed	结果格式（默认：simple）
profanity	否	masked、removed、raw	脏话处理方式（默认：masked）

识别状态值

状态	说明
Success	识别成功
NoMatch	检测到语音，但未匹配到任何词语
InitialSilenceTimeout	仅检测到静音
BabbleTimeout	仅检测到噪声
Error	服务内部错误

脏话处理

# Mask profanity with asterisks (default)
params = {"language": "en-US", "profanity": "masked"}

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

# Include profanity as-is
params = {"language": "en-US", "profanity": "raw"}

错误处理

import requests

def transcribe_with_error_handling(audio_path: str, language: str = "en-US") -> dict | None:
    """Transcribe with proper error handling."""
    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-Subscription-Key": api_key,
                    "Content-Type": "audio/wav; codecs=audio/pcm; samplerate=16000",
                    "Accept": "application/json"
                },
                params={"language": language, "format": "detailed"},
                data=audio_file,
                timeout=(10, 90)
            )
        
        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("Forbidden: Check resource permissions, endpoint and service policy")
        else:
            print(f"Error {response.status_code}; response body withheld")
        
        return None
        
    except requests.exceptions.RequestException as e:
        print(f"Request failed: {e}")
        return None

异步版本

import os
import aiohttp
import asyncio

async def transcribe_async(audio_file_path: str, language: str = "en-US") -> dict:
    """Async version using 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(timeout=aiohttp.ClientTimeout(total=90)) 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()

# Usage
result = asyncio.run(transcribe_async("audio.wav", "en-US"))
if result.get("RecognitionStatus") == "Success" and result.get("NBest"):
    print(result["NBest"][0].get("Display", ""))
else:
    print("No successful detailed recognition result")

支持的语言

常见语言代码（参见完整列表）：

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

最佳实践

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

不适合使用此 API 的情况

在以下情况下，请改用 Speech SDK 或 Batch Transcription API：

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

参考文件

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

适用情况

当需要执行概述中描述的工作流或操作时，可使用此技能。

限制

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

示例会将音频和参考文本发送到 Azure；应先明确数据传输与保留范围，并默认使用合成录音。在附加凭据前，先根据实际获批的 Azure 资源解析并验证已配置的区域或资源终结点；不要接受来自不可信输入的任意主机名。在本地检查时长、格式和大小，限制请求数量，并确保密钥和转录内容不会写入日志。详细识别结果使用 NBest[].Display，并且必须在索引前检查 Success 和结果是否非空。发音和韵律评分是模型估算值，并非临床判断或对说话人质量的客观评估；对区域设置和 API 的支持因情况而异。请参阅官方短音频 API。