使用Watson SDK实现带说话人识别的实时连续语音转文字

刚好之前做过类似的需求，IBM Watson的Speech to Text服务其实专门提供了流式WebSocket接口来处理这种需要保持会话上下文的实时转录场景——完全不需要拆分30秒片段，拆分反而会破坏说话人识别的连贯性。下面直接给你Python实现的完整方案：

第一步：安装依赖 #

先装好需要的工具包：

pip install ibm-watson python-dotenv pyaudio

• ibm-watson：官方SDK，用来调用Watson服务
• python-dotenv：管理环境变量，避免硬编码密钥
• pyaudio：捕获麦克风实时音频流（处理预录制文件时可选，但需要流式读取逻辑）

第二步：核心实现代码 #

场景1：实时麦克风输入+说话人识别 #

纯实时场景，直接从麦克风捕获音频流发送给Watson，全程保持会话上下文：

import os
from dotenv import load_dotenv
from ibm_watson import SpeechToTextV1
from ibm_cloud_sdk_core.authenticators import IAMAuthenticator
from ibm_watson.websocket import RecognizeCallback, AudioSource
import pyaudio

# 加载环境变量（建议把密钥存在.env文件，不要硬编码）
load_dotenv()
API_KEY = os.getenv('WATSON_STT_API_KEY')
SERVICE_URL = os.getenv('WATSON_STT_URL')

# 自定义回调类，处理Watson返回的转录结果
class MyRecognizeCallback(RecognizeCallback):
    def __init__(self):
        RecognizeCallback.__init__(self)

    def on_transcription(self, transcript):
        # 解析带说话人标签的最终转录结果
        for result in transcript['results']:
            if result['final']:  # 只处理最终结果，过滤临时中间结果
                print(f"\n=== 最终转录结果 ===")
                for alt in result['alternatives']:
                    print(f"文本内容: {alt['transcript']}")
                    # 提取说话人标签信息
                    if 'speaker_labels' in alt:
                        for label in alt['speaker_labels']:
                            print(f"说话人ID: {label['speaker']}, 时间段: [{label['start_time']}s - {label['end_time']}s]")

    def on_connected(self):
        print("已连接到Watson Speech to Text服务")

    def on_error(self, error):
        print(f"服务调用出错: {error}")

    def on_listening(self):
        print("正在监听音频，请开始说话...")

def main():
    # 初始化认证和服务实例
    authenticator = IAMAuthenticator(API_KEY)
    speech_to_text = SpeechToTextV1(authenticator=authenticator)
    speech_to_text.set_service_url(SERVICE_URL)

    # 配置音频参数（必须和Watson服务要求匹配，这里是16kHz单声道PCM）
    FORMAT = pyaudio.paInt16
    CHANNELS = 1
    RATE = 16000
    CHUNK = 1024

    audio = pyaudio.PyAudio()
    # 打开麦克风音频流
    stream = audio.open(format=FORMAT, channels=CHANNELS,
                        rate=RATE, input=True,
                        frames_per_buffer=CHUNK)

    # 初始化音频源
    audio_source = AudioSource(stream, True, True)

    # 启动流式识别，关键是开启speaker_labels
    recognize_callback = MyRecognizeCallback()
    speech_to_text.recognize_using_websocket(
        audio=audio_source,
        content_type=f'audio/l16; rate={RATE}; channels={CHANNELS}',
        recognize_callback=recognize_callback,
        speaker_labels=True,  # 强制开启说话人识别
        interim_results=False,  # 不需要临时结果就设为False，减少输出干扰
        number_of_speakers=2  # 可选：指定预期说话人数，提升识别准确率
    )

    # 停止音频流
    stream.stop_stream()
    stream.close()
    audio.terminate()

if __name__ == '__main__':
    main()

场景2：预录制音频文件流式处理（模拟实时，保留上下文） #

如果是处理已有的完整音频文件，需要模拟实时发送（避免一次性上传破坏说话人上下文），可以用这段代码：

import os
from dotenv import load_dotenv
from ibm_watson import SpeechToTextV1
from ibm_cloud_sdk_core.authenticators import IAMAuthenticator
from ibm_watson.websocket import RecognizeCallback, AudioSource
import wave
import time

load_dotenv()
API_KEY = os.getenv('WATSON_STT_API_KEY')
SERVICE_URL = os.getenv('WATSON_STT_URL')

class MyRecognizeCallback(RecognizeCallback):
    def __init__(self):
        RecognizeCallback.__init__(self)

    def on_transcription(self, transcript):
        for result in transcript['results']:
            if result['final']:
                print(f"\n=== 最终转录结果 ===")
                for alt in result['alternatives']:
                    print(f"文本内容: {alt['transcript']}")
                    if 'speaker_labels' in alt:
                        for label in alt['speaker_labels']:
                            print(f"说话人ID: {label['speaker']}, 时间段: [{label['start_time']:.2f}s - {label['end_time']:.2f}s]")

    def on_connected(self):
        print("已连接到Watson服务")

    def on_error(self, error):
        print(f"服务调用出错: {error}")

def main():
    authenticator = IAMAuthenticator(API_KEY)
    speech_to_text = SpeechToTextV1(authenticator=authenticator)
    speech_to_text.set_service_url(SERVICE_URL)

    # 打开预录制音频文件（要求：WAV格式、16kHz采样率、单声道PCM）
    audio_file = "your_audio_file.wav"
    wf = wave.open(audio_file, 'rb')

    # 模拟实时发送：按chunk读取，根据音频帧率计算发送间隔
    CHUNK = 1024
    RATE = wf.getframerate()
    # 计算每个chunk对应的播放时长（秒）
    chunk_duration = CHUNK / RATE

    def audio_generator():
        while True:
            data = wf.readframes(CHUNK)
            if not data:
                break
            time.sleep(chunk_duration)  # 模拟实时播放速度，避免一次性发送全部数据
            yield data

    audio_source = AudioSource(audio_generator())

    recognize_callback = MyRecognizeCallback()
    speech_to_text.recognize_using_websocket(
        audio=audio_source,
        content_type=f'audio/l16; rate={RATE}; channels={wf.getnchannels()}',
        recognize_callback=recognize_callback,
        speaker_labels=True,
        interim_results=False
    )

    wf.close()

if __name__ == '__main__':
    main()

关键知识点说明 #

1. 为什么用WebSocket而非HTTP批量接口？
   HTTP批量接口是无状态的，每次发送音频片段都会让Watson重新初始化说话人识别模型，导致同一个说话人在不同片段里被标记为不同ID。而WebSocket是长连接，服务端会在整个会话中保持说话人模型的上下文，确保说话人ID连贯。

2. 说话人识别核心参数

   • speaker_labels=True：必须开启才会返回说话人标签
   • number_of_speakers：可选，指定预期说话人数，帮助Watson更准确聚类说话人
   • speaker_label_threshold：可选，调整识别置信度阈值（默认0.5），数值越高识别越严格

3. 音频格式要求
   流式识别推荐用audio/l16（16位PCM）、16kHz采样率、单声道，这个格式兼容性最好，识别准确率最高。如果你的音频是其他格式，需要先转码（比如用ffmpeg）。

注意事项 #

• 把IBM Cloud的API密钥和服务URL存在.env文件中，不要硬编码到代码里，避免泄露
• 确保你的Watson服务实例有足够的流式识别配额（流式和批量配额是分开的）
• 处理超长音频时，官方SDK已经内置了心跳逻辑，无需额外处理连接超时

内容的提问来源于stack exchange，提问作者Tim