Azure OpenAI 最近增强了其语音识别产品，推出了两款令人印象深刻的模型：GPT-4o-transcribe 和 GPT-4o-mini-transcribe。它们的一个关键特性是利用 WebSocket 连接进行实时音频流转录。这为开发者们提供了构建语音转文本应用的先进工具。本文将深入探讨这些模型的工作原理，并提供一个用 Python 实现的实用示例。

理解 OpenAI 的实时转录 API

与常规的用于音频转录的 REST API 不同，Azure OpenAI 的实时 API 支持通过 WebSocket 或 WebRTC 连接持续流式传输音频数据。这种方法对于需要即时转录反馈的应用特别有价值，例如实时字幕、会议转录或语音助手。

标准转录 API 和实时 API 之间的关键区别在于，实时转录会话通常不包含来自模型的响应，而是专注于实时地将语音转换为文本。

GPT-4o-transcribe 和 GPT-4o-mini-transcribe：功能概览

Azure OpenAI 推出了两款专门的转录模型：

  • GPT-4o-transcribe：功能齐全的转录模型，具有高准确性。
  • GPT-4o-mini-transcribe：一个更轻量、更快速的模型，准确性略有降低，但延迟更低。

两种模型都通过 WebSocket 连接，使开发人员能够直接从麦克风或其他来源流式传输音频以进行即时转录。这些模型专为实时 API 基础架构设计。

设置环境

首先，需要使用必要的库来设置 Python 环境：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

import os
import json
import base64
import threading
import pyaudio # 用于处理音频
import websocket # 用于 WebSocket 通信
from dotenv import load_dotenv # 用于加载环境变量

# 从 .env 文件加载环境变量 (例如 azure.env)
# 确保你的 .env 文件中有 AZURE_OPENAI_STT_TTS_KEY 和 AZURE_OPENAI_STT_TTS_ENDPOINT
load_dotenv('azure.env') 

OPENAI_API_KEY = os.environ.get("AZURE_OPENAI_STT_TTS_KEY")
if not OPENAI_API_KEY:
    raise RuntimeError("❌ 环境变量 AZURE_OPENAI_STT_TTS_KEY 未设置！")

AZURE_OPENAI_ENDPOINT = os.environ.get('AZURE_OPENAI_STT_TTS_ENDPOINT')
if not AZURE_OPENAI_ENDPOINT:
    raise RuntimeError("❌ 环境变量 AZURE_OPENAI_STT_TTS_ENDPOINT 未设置！")

# OpenAI 实时 API (转录模型) 的 WebSocket 端点
# 将 https 替换为 wss 用于 WebSocket 连接
url = f"""{AZURE_OPENAI_ENDPOINT.replace('https', 'wss')}/openai/realtime
?api-version=2025-04-01-preview&intent=transcription"""
headers = { "api-key": OPENAI_API_KEY }

# 音频流参数 (16位 PCM, 24kHz 单声道)
# 注意: 原文提及16kHz, 但示例代码中使用24000Hz。此处保留代码中的24000Hz。
RATE = 24000  # 采样率
CHANNELS = 1  #声道数
FORMAT = pyaudio.paInt16  # 音频格式 (16位)
CHUNK = 1024  # 每次读取的音频块大小

# 初始化 PyAudio
audio_interface = pyaudio.PyAudio()
# 打开音频流
stream = audio_interface.open(format=FORMAT,
                              channels=CHANNELS,
                              rate=RATE,
                              input=True,  # 设置为输入流
                              frames_per_buffer=CHUNK)
建立 WebSocket 连接

以下代码建立到 OpenAI 实时 API 的连接并配置转录会话：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

def on_open(ws):
    print("WebSocket 连接成功！请开始说话...")
    session_config = {
        "type": "transcription_session.update",
        "session": {
            "input_audio_format": "pcm16", # 输入音频格式
            "input_audio_transcription": {
                "model": "gpt-4o-mini-transcribe", # 可选 "gpt-4o-transcribe" 以获得更高精度
                "prompt": "Respond in English." # 提示模型输出英文，可根据需求调整
            },
            "input_audio_noise_reduction": { # 输入音频降噪配置
                "type": "near_field" # 近场降噪，适用于耳机等近距离麦克风
            },
            "turn_detection": { # 轮次检测配置 (判断用户是否说完一句话)
                "type": "server_vad", # 服务器端语音活动检测 (VAD)
                "threshold": 0.5, # VAD 阈值
                "prefix_padding_ms": 300, # 前缀填充毫秒数
                "silence_duration_ms": 200 # 静默持续毫秒数
            }
        }
    }
    # 发送会话配置信息
    ws.send(json.dumps(session_config))

    # 定义一个函数来处理麦克风音频流
    def stream_microphone():
        try:
            # WebSocketApp 会在 run_forever 循环中管理 keep_running
            # 此处检查 ws.keep_running 属性 (由 websocket-client 库设置)
            while getattr(ws, 'keep_running', True):
                audio_data = stream.read(CHUNK, exception_on_overflow=False)
                audio_base64 = base64.b64encode(audio_data).decode('utf-8')
                
                # 仅当 WebSocket 仍处于连接状态时发送数据
                if ws.sock and ws.sock.connected:
                    ws.send(json.dumps({
                        "type": "input_audio_buffer.append",
                        "audio": audio_base64
                    }))
                else:
                    print("WebSocket 已关闭，停止发送音频数据。")
                    break # 如果连接关闭则退出循环
        except Exception as e:
            print(f"音频流错误: {e}")
        # finally:
            # WebSocket 的关闭通常由 on_close 回调或 run_forever 结束时处理
            # 在此显式调用 ws.close() 可能导致竞争条件或重复关闭

    # 启动一个守护线程 (daemon thread) 来发送麦克风音频流
    # 守护线程会随主线程退出而退出
    threading.Thread(target=stream_microphone, daemon=True).start()
处理转录结果

此部分处理包含转录结果的传入 WebSocket 消息：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

def on_message(ws, message):
    try:
        data = json.loads(message)
        event_type = data.get("type", "")
        # print(f"接收到事件类型: {event_type}") # 取消注释以进行调试
        # print(data) # 取消注释以查看完整数据

        # 处理实时的增量转录文本 (delta)
        if event_type == "conversation.item.input_audio_transcription.delta":
            transcript_piece = data.get("delta", "")
            if transcript_piece:
                print(transcript_piece, end='', flush=True) # 连续打印，不换行
        
        # 处理一段完整的语音转录完成事件
        if event_type == "conversation.item.input_audio_transcription.completed":
            # 确保在打印前换行，如果之前有 delta 输出
            print(f"
[分段完成]: {data['transcript']}")

        # 根据原文，"item" 类型也可能表示最终的转录结果
        if event_type == "item": 
            transcript = data.get("item", {}).get("transcript", "") # 更安全地获取 transcript
            if transcript:
                 print(f"
[最终转录条目]: {transcript}")

    except json.JSONDecodeError:
        print(f"无法解析消息为 JSON: {message}")
    except Exception as e:
        print(f"处理消息时发生错误: {e}")
        # pass # 可以选择忽略不相关的事件或特定错误
错误处理和清理

为确保正确的资源管理，实现了错误和连接关闭的处理程序：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

def on_error(ws, error):
    print(f"WebSocket 错误: {error}")
    # 可以在此处添加更复杂的错误处理逻辑，例如记录错误或尝试重新连接

def on_close(ws, close_status_code, close_msg):
    print(f"已从服务器断开连接。状态码: {close_status_code}, 消息: {close_msg}")
    # 确保音频流和 PyAudio 实例被正确关闭和终止
    if stream.is_active():
        stream.stop_stream()
    # 双重检查以确保流已停止
    if not stream.is_stopped(): 
        stream.stop_stream()
    stream.close()
    audio_interface.terminate()
    print("音频资源已释放。")
运行 WebSocket 客户端

最后，此代码初始化 WebSocket 连接并开始转录过程：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

if __name__ == "__main__": # 确保作为脚本运行时执行
    print("正在连接到 Azure OpenAI 实时语音转录 API...")
    ws_app = websocket.WebSocketApp(
        url,
        header=headers,
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )
    try:
        # run_forever() 会阻塞直到连接关闭或发生错误
        ws_app.run_forever()
    except KeyboardInterrupt:
        print("
用户请求中断程序...")
    finally:
        # 确保在退出时 WebSocket 连接已关闭
        if ws_app and getattr(ws_app, 'sock', None) and getattr(ws_app.sock, 'connected', False):
            print("正在关闭 WebSocket 连接...")
            ws_app.close()
        print("程序已退出。")

分析实现细节

会话配置

让我们分解一下会话配置的关键组件：

  • input_audio_format：指定为 "pcm16"，表示16位 PCM 音频。
  • input_audio_transcription：
    • model：指定为 "gpt-4o-mini-transcribe"（可以替换为 "gpt-4o-transcribe" 以获得更高准确性）。
    • prompt：向模型提供指令（例如："Respond in English."，意为“用英语回答”）。这有助于引导模型的输出语言和风格。
    • language：指定音频的语言，如 "zh-CN" (简体中文) 或 "hi" (印地语)。如果设为 null 或不提供，模型会尝试自动检测语言。
  • input_audio_noise_reduction：降噪类型。near_field 适用于近距离麦克风（如耳机），far_field 适用于远场麦克风（如笔记本电脑或会议室麦克风）。
  • turn_detection：配置 "server_vad"（语音活动检测）以自动检测语音轮次（即用户何时说完一句话）。可以设为 null 来关闭此功能，这种情况下客户端需要通过其他方式（如特定事件）手动触发模型响应。
    • Server VAD：模型根据音量检测语音的开始和结束，并在用户语音结束时响应。
    • Semantic VAD：这是一种更高级的轮次检测机制，它使用一个专门的轮次检测模型（结合传统的VAD）来从语义上估计用户是否已经讲完话。然后，它会根据这个概率动态地设置一个超时时间。例如，如果用户的音频以“嗯…”这样的拖尾音结束，模型会判断为轮次结束的概率较低，并等待更长时间让用户继续讲话。这种方式对于实现更自然的对话交互非常有用，但可能会带来稍高的延迟。
音频流式传输

该实现使用线程化方法将音频数据从麦克风连续流式传输到 WebSocket 连接。每个音频块都会：

  1. 从麦克风读取。
  2. 编码为 base64 字符串。
  3. 作为带有 "input_audio_buffer.append" 事件类型的 JSON 消息发送给服务器。
转录事件

系统处理来自 WebSocket 连接的几种类型的事件：

  • conversation.item.input_audio_transcription.delta：转录的增量更新。这使得可以实时显示正在识别的文本。
  • conversation.item.input_audio_transcription.completed：一个语音片段（turn）的完整转录稿。
  • item：根据原始文章的代码结构，此类型也可能用于传递最终的转录结果。
自定义选项

可以通过多种方式自定义示例代码：

  • 在 gpt-4o-transcribe（高精度）和 gpt-4o-mini-transcribe（低延迟）模型之间切换。
  • 调整音频参数，如采样率（RATE）、通道数（CHANNELS）、块大小（CHUNK），以匹配音频源或优化性能。
  • 修改 prompt 来为模型提供更多上下文信息或指定特定的语言偏好。
  • 根据实际的麦克风和环境配置 input_audio_noise_reduction（例如，从 near_field 改为 far_field）。
  • 针对不同的说话模式或应用场景调整 turn_detection 参数，或完全禁用它。
部署注意事项

在生产环境中部署此解决方案时，请考虑以下几点：

  • 身份验证：安全地存储和管理 API 密钥。避免将密钥硬编码到代码中，应使用环境变量或 Azure Key Vault 等服务。
  • 错误处理与重连：实现更健壮的错误处理逻辑，包括网络中断时的自动重新连接机制。
  • 性能优化：根据具体的用例优化音频参数和处理流程，以平衡延迟和准确性。
  • 速率限制：了解并遵守 Azure OpenAI 实时 API 的速率限制，以避免服务中断。
  • 回退策略：为连接中断或 API 故障等情况实施回退策略，例如切换到备用服务或通知用户。
  • 日志记录与监控：集成日志记录和监控系统，以便跟踪应用性能、诊断问题和审计用量。

总结

GPT-4o-transcribe 和 GPT-4o-mini-transcribe 代表了实时语音识别技术的重大进步。通过利用 WebSocket 进行连续音频流传输，这些模型使开发人员能够以最小的延迟构建响应迅速的语音转文本应用程序。

本文中展示的 Python 实现演示了如何快速搭建一个实时转录系统。这个基础可以扩展到各种应用，从实时字幕和会议转录到语音控制界面和无障碍工具。

随着这些模型的不断发展和完善，我们可以期待它们在准确性和性能上带来更多惊喜，为各行各业的语音识别应用开辟新的可能性。

请记住，在生产环境中实施这些 API 时，务必遵循 Azure OpenAI 关于 API 安全的最佳实践，包括采用安全的身份验证方法并妥善保管您的 API 密钥。