Qwen3-ASR流式处理实战：实时语音转写技术

想象一下这样的场景：一场重要的线上会议正在进行，你需要实时记录下每个人的发言要点；或者，你正在观看一场外语直播，希望能立刻看到翻译字幕。在这些情况下，等音频全部录完再转成文字，显然已经来不及了。

这就是实时语音转写的魅力所在——声音一边说，文字一边就出来了，几乎没有延迟。今天，我们就来聊聊如何用Qwen3-ASR这个强大的开源模型，自己动手搭建一套这样的实时转写系统。

1. 为什么你需要关注流式语音转写？

在深入技术细节之前，我们先搞清楚流式处理到底解决了什么问题。

传统的语音识别，就像是你把一整段录音交给一个速记员，他听完后再把整理好的文稿给你。这个过程需要等待，而且如果录音很长，等待时间就更久。

而流式处理，更像是这个速记员就坐在你旁边，你一边说，他一边写。你说完一句话，他几乎同时就把这句话的文字版递给你。这种体验上的差异是巨大的。

几个典型的应用场景：

• 实时会议记录：线上会议中，发言内容实时变成文字，方便后续整理和搜索。
• 直播字幕生成：为主播或视频内容添加实时字幕，提升观看体验。
• 语音助手交互：更自然的对话体验，减少用户等待时间。
• 客服电话实时分析：在通话过程中实时分析客户情绪和需求。

Qwen3-ASR在这方面做得相当不错，它专门提供了流式处理的接口，让我们能够以很低的延迟获取识别结果。

2. 准备工作：环境与认证

在开始写代码之前，我们需要做一些准备工作。别担心，步骤不多，跟着做就行。

2.1 获取API密钥

首先，你需要一个访问Qwen3-ASR服务的凭证。如果你使用阿里云百炼平台，可以按照以下步骤获取：

1. 登录阿里云百炼控制台
2. 在API密钥管理页面创建新的密钥
3. 复制生成的sk-开头的密钥字符串

安全提示：千万不要把API密钥直接写在代码里提交到公开仓库。推荐的做法是设置环境变量。

# Linux/Mac
export DASHSCOPE_API_KEY="你的API密钥"

# Windows (PowerShell)
$env:DASHSCOPE_API_KEY="你的API密钥"

2.2 安装必要的Python库

Qwen3-ASR提供了多种调用方式，对于实时流式处理，我们主要使用WebSocket协议。确保你的Python环境已经安装了必要的库：

pip install dashscope websocket-client

如果你之前安装过不同版本的websocket库，可能会遇到冲突，可以先卸载再安装：

pip uninstall websocket-client websocket -y
pip install websocket-client

注意：不要把你的代码文件命名为websocket.py，这会导致导入冲突。起个其他名字，比如realtime_asr.py。

3. 核心概念：流式处理是如何工作的？

在写代码之前，我们先简单了解一下流式语音识别的底层原理。这样你在调试的时候，就知道每一步在干什么。

关键点1：WebSocket连接 流式处理使用WebSocket而不是普通的HTTP请求。WebSocket允许服务器和客户端之间建立持久连接，双方可以随时互相发送数据。这就像打电话一样，接通后可以一直通话，而不需要像发短信那样每次都要重新建立连接。

关键点2：音频分块发送 我们不会一次性发送整个音频文件，而是把它切成很多小片段，比如每0.1秒的音频数据作为一个数据包发送。服务器收到一个数据包就处理一个，处理完就返回对应的文字结果。

关键点3：语音活动检测（VAD） VAD的作用是判断什么时候开始说话，什么时候结束。Qwen3-ASR支持两种模式：

• 服务器端VAD：由服务器自动检测
• 手动控制：由客户端决定什么时候开始和结束

对于大多数场景，使用服务器端VAD更方便，我们主要介绍这种方式。

4. 实战：构建你的第一个实时转写程序

现在让我们动手写代码。我会带你一步步构建一个完整的实时语音转写客户端。

4.1 基础WebSocket连接

首先，我们建立与Qwen3-ASR服务器的连接：

import os
import json
import time
import base64
import websocket
import threading
import logging
from datetime import datetime

# 设置日志，方便调试
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class QwenRealtimeASR:
    def __init__(self, api_key=None, model="qwen3-asr-flash-realtime"):
        """
        初始化实时ASR客户端
        
        参数：
        api_key: 你的API密钥，如果为None则从环境变量读取
        model: 使用的模型名称
        """
        self.api_key = api_key or os.getenv("DASHSCOPE_API_KEY")
        if not self.api_key:
            raise ValueError("请设置DASHSCOPE_API_KEY环境变量或直接传入api_key参数")
        
        self.model = model
        self.base_url = "wss://dashscope.aliyuncs.com/api-ws/v1/realtime"
        self.url = f"{self.base_url}?model={self.model}"
        
        self.ws = None
        self.is_running = False
        self.transcript_text = ""
        
    def connect(self):
        """建立WebSocket连接"""
        headers = [
            f"Authorization: Bearer {self.api_key}",
            "OpenAI-Beta: realtime=v1"
        ]
        
        logger.info(f"正在连接到: {self.url}")
        
        self.ws = websocket.WebSocketApp(
            self.url,
            header=headers,
            on_open=self._on_open,
            on_message=self._on_message,
            on_error=self._on_error,
            on_close=self._on_close
        )
        
        # 在后台运行WebSocket
        self.ws_thread = threading.Thread(target=self.ws.run_forever)
        self.ws_thread.daemon = True
        self.ws_thread.start()
        
        # 等待连接建立
        time.sleep(2)
        return self

这段代码创建了一个基础的客户端类，处理了认证和连接建立。注意我们使用了websocket.WebSocketApp，它会自动处理WebSocket协议的各种细节。

4.2 处理连接事件

WebSocket连接建立后，会有各种事件发生。我们需要为这些事件编写处理函数：

    def _on_open(self, ws):
        """连接成功建立时的回调"""
        logger.info("连接服务器成功")
        
        # 发送会话配置
        session_config = {
            "event_id": f"event_{int(time.time() * 1000)}",
            "type": "session.update",
            "session": {
                "modalities": ["text"],  # 我们只需要文本输出
                "input_audio_format": "pcm",
                "sample_rate": 16000,    # 音频采样率
                "input_audio_transcription": {
                    "language": "zh"      # 识别语言，中文
                },
                "turn_detection": {       # 使用服务器端VAD
                    "type": "server_vad",
                    "threshold": 0.0,
                    "silence_duration_ms": 400  # 静音400毫秒认为一句话结束
                }
            }
        }
        
        logger.debug(f"发送会话配置: {json.dumps(session_config, ensure_ascii=False)}")
        ws.send(json.dumps(session_config))
        
    def _on_message(self, ws, message):
        """收到服务器消息时的回调"""
        try:
            data = json.loads(message)
            event_type = data.get("type")
            
            if event_type == "session.finished":
                # 整个会话结束
                final_text = data.get("transcript", "")
                logger.info(f"最终识别结果: {final_text}")
                self.transcript_text = final_text
                self.is_running = False
                
            elif event_type == "transcript.delta":
                # 增量识别结果
                delta_text = data.get("delta", "")
                if delta_text:
                    print(f"实时输出: {delta_text}", end="", flush=True)
                    
            elif event_type == "transcript.completed":
                # 一句话识别完成
                sentence = data.get("text", "")
                if sentence:
                    print(f"\n完整句子: {sentence}")
                    
        except json.JSONDecodeError as e:
            logger.error(f"解析消息失败: {e}, 原始消息: {message}")
            
    def _on_error(self, ws, error):
        """发生错误时的回调"""
        logger.error(f"WebSocket错误: {error}")
        
    def _on_close(self, ws, close_status_code, close_msg):
        """连接关闭时的回调"""
        logger.info(f"连接关闭: {close_status_code} - {close_msg}")
        self.is_running = False

这里有几个关键点需要注意：

1. session.update：这是我们告诉服务器如何配置识别参数的地方。server_vad表示使用服务器端的语音活动检测。
2. transcript.delta：这是流式输出的核心！服务器会实时返回识别出的文字片段。
3. transcript.completed：当VAD检测到一句话结束时，服务器会返回完整的句子。

4.3 发送音频数据

连接建立并配置好后，我们就可以开始发送音频数据了：

    def send_audio_file(self, audio_file_path):
        """
        发送本地音频文件进行实时识别
        
        参数：
        audio_file_path: PCM格式的音频文件路径
        """
        if not self.ws:
            raise RuntimeError("请先调用connect()方法建立连接")
        
        # 等待会话配置完成
        time.sleep(1)
        
        self.is_running = True
        
        logger.info(f"开始发送音频文件: {audio_file_path}")
        
        with open(audio_file_path, 'rb') as f:
            while self.is_running:
                # 每次读取0.1秒的音频数据 (16000Hz * 2字节 * 0.1秒 = 3200字节)
                audio_chunk = f.read(3200)
                
                if not audio_chunk:
                    # 文件读取完毕
                    logger.info("音频文件发送完成")
                    
                    # 告诉服务器音频发送结束
                    finish_event = {
                        "event_id": f"event_{int(time.time() * 1000)}",
                        "type": "session.finish"
                    }
                    self.ws.send(json.dumps(finish_event))
                    break
                
                # 将音频数据编码为base64
                audio_base64 = base64.b64encode(audio_chunk).decode('utf-8')
                
                # 构建音频数据事件
                audio_event = {
                    "event_id": f"event_{int(time.time() * 1000)}",
                    "type": "input_audio_buffer.append",
                    "audio": audio_base64
                }
                
                # 发送音频数据
                self.ws.send(json.dumps(audio_event))
                
                # 模拟实时采集的延迟
                time.sleep(0.1)
        
        # 等待识别完成
        while self.is_running:
            time.sleep(0.1)
            
    def close(self):
        """关闭连接"""
        if self.ws:
            self.ws.close()
        self.is_running = False

这里的关键是input_audio_buffer.append事件。我们每次发送一小段音频数据（3200字节，对应0.1秒的16kHz 16位PCM音频），然后等待0.1秒再发送下一段。这样模拟了实时音频流的效果。

4.4 完整的使用示例

现在我们把所有部分组合起来，看看如何实际使用这个客户端：

def main():
    """主函数：演示完整的使用流程"""
    
    # 1. 创建客户端实例
    # 注意：确保已经设置了DASHSCOPE_API_KEY环境变量
    asr_client = QwenRealtimeASR()
    
    try:
        # 2. 建立连接
        asr_client.connect()
        
        # 3. 发送音频文件
        # 假设你有一个PCM格式的音频文件
        audio_file = "test_audio.pcm"
        
        if os.path.exists(audio_file):
            print("开始实时语音识别...")
            print("-" * 50)
            
            # 在一个单独的线程中发送音频
            audio_thread = threading.Thread(
                target=asr_client.send_audio_file,
                args=(audio_file,)
            )
            audio_thread.start()
            
            # 等待识别完成
            audio_thread.join()
            
            print("\n" + "-" * 50)
            print(f"识别完成，最终文本：{asr_client.transcript_text}")
            
        else:
            print(f"音频文件 {audio_file} 不存在")
            print("请先准备一个PCM格式的音频文件")
            print("可以使用以下命令将MP3转换为PCM：")
            print("ffmpeg -i input.mp3 -ar 16000 -ac 1 -f s16le output.pcm")
            
    except KeyboardInterrupt:
        print("\n用户中断")
    except Exception as e:
        print(f"发生错误: {e}")
    finally:
        # 4. 清理资源
        asr_client.close()
        print("程序结束")

if __name__ == "__main__":
    main()

5. 处理真实麦克风输入

上面的例子是处理已经录制好的音频文件。但在实际应用中，我们更可能需要处理实时的麦克风输入。让我们看看如何实现：

import pyaudio
import wave
import numpy as np

class MicrophoneStreamASR(QwenRealtimeASR):
    """处理麦克风实时输入的ASR客户端"""
    
    def __init__(self, api_key=None):
        super().__init__(api_key)
        self.audio_interface = None
        self.audio_stream = None
        
    def start_microphone(self):
        """开始从麦克风采集音频"""
        import pyaudio
        
        # 音频参数
        FORMAT = pyaudio.paInt16
        CHANNELS = 1
        RATE = 16000
        CHUNK = 3200  # 0.1秒的数据
        
        self.audio_interface = pyaudio.PyAudio()
        
        # 打开音频流
        self.audio_stream = self.audio_interface.open(
            format=FORMAT,
            channels=CHANNELS,
            rate=RATE,
            input=True,
            frames_per_buffer=CHUNK
        )
        
        logger.info("麦克风已开启，开始采集音频...")
        
    def stream_from_microphone(self, duration_seconds=30):
        """
        从麦克风流式传输音频
        
        参数：
        duration_seconds: 采集时长（秒）
        """
        if not self.ws:
            raise RuntimeError("请先调用connect()方法建立连接")
        
        if not self.audio_stream:
            self.start_microphone()
        
        self.is_running = True
        
        # 等待会话配置完成
        time.sleep(1)
        
        print("开始说话吧（按Ctrl+C停止）...")
        print("-" * 50)
        
        start_time = time.time()
        
        try:
            while self.is_running and (time.time() - start_time) < duration_seconds:
                # 从麦克风读取数据
                audio_data = self.audio_stream.read(3200, exception_on_overflow=False)
                
                # 转换为base64
                audio_base64 = base64.b64encode(audio_data).decode('utf-8')
                
                # 发送到服务器
                audio_event = {
                    "event_id": f"event_{int(time.time() * 1000)}",
                    "type": "input_audio_buffer.append",
                    "audio": audio_base64
                }
                
                self.ws.send(json.dumps(audio_event))
                
        except KeyboardInterrupt:
            print("\n停止采集")
        finally:
            # 发送结束信号
            finish_event = {
                "event_id": f"event_{int(time.time() * 1000)}",
                "type": "session.finish"
            }
            self.ws.send(json.dumps(finish_event))
            
    def cleanup(self):
        """清理资源"""
        super().close()
        
        if self.audio_stream:
            self.audio_stream.stop_stream()
            self.audio_stream.close()
        
        if self.audio_interface:
            self.audio_interface.terminate()

使用这个麦克风版本很简单：

def realtime_microphone_demo():
    """实时麦克风识别演示"""
    client = MicrophoneStreamASR()
    
    try:
        client.connect()
        client.stream_from_microphone(duration_seconds=60)  # 识别60秒
        
        # 等待识别完成
        while client.is_running:
            time.sleep(0.1)
            
    except KeyboardInterrupt:
        print("\n用户中断")
    except Exception as e:
        print(f"错误: {e}")
    finally:
        client.cleanup()
        print("识别结束")

6. 高级技巧与优化建议

基本的流式处理掌握了，我们来看看如何让它工作得更好。

6.1 调整VAD参数

VAD（语音活动检测）的敏感度会影响识别效果。Qwen3-ASR允许我们调整VAD参数：

# 更敏感的VAD设置（适合安静环境）
sensitive_vad = {
    "type": "server_vad",
    "threshold": -0.5,  # 更低的阈值，更容易触发
    "silence_duration_ms": 300  # 静音300毫秒就认为一句话结束
}

# 更保守的VAD设置（适合嘈杂环境）
conservative_vad = {
    "type": "server_vad", 
    "threshold": 0.5,  # 更高的阈值，需要更清晰的语音
    "silence_duration_ms": 800  # 静音800毫秒才认为结束
}

6.2 处理识别错误和重试

网络不稳定时，连接可能会中断。我们需要添加重试机制：

class RobustRealtimeASR(QwenRealtimeASR):
    """带有重试机制的ASR客户端"""
    
    def __init__(self, api_key=None, max_retries=3):
        super().__init__(api_key)
        self.max_retries = max_retries
        self.retry_count = 0
        
    def connect_with_retry(self):
        """带重试的连接"""
        while self.retry_count < self.max_retries:
            try:
                self.connect()
                logger.info("连接成功")
                return True
            except Exception as e:
                self.retry_count += 1
                logger.warning(f"连接失败，第{self.retry_count}次重试: {e}")
                time.sleep(2 ** self.retry_count)  # 指数退避
        
        logger.error(f"连接失败，已达最大重试次数{self.max_retries}")
        return False
    
    def _on_error(self, ws, error):
        """错误处理，尝试重连"""
        logger.error(f"连接错误: {error}")
        
        if self.retry_count < self.max_retries:
            logger.info("尝试重新连接...")
            time.sleep(2)
            self.retry_count += 1
            self.connect()

6.3 实时结果后处理

有时候，我们希望对识别出的文字进行实时处理，比如过滤敏感词、添加标点等：

class PostProcessASR(QwenRealtimeASR):
    """带有后处理的ASR客户端"""
    
    def __init__(self, api_key=None):
        super().__init__(api_key)
        self.word_buffer = []  # 缓存单词
        self.sentence_buffer = ""  # 缓存句子
        
    def _on_message(self, ws, message):
        """重写消息处理，添加后处理"""
        super()._on_message(ws, message)
        
        try:
            data = json.loads(message)
            
            if data.get("type") == "transcript.delta":
                delta_text = data.get("delta", "")
                
                # 简单的后处理：添加空格分隔
                if delta_text:
                    self.word_buffer.append(delta_text)
                    
                    # 每5个单词整理一次
                    if len(self.word_buffer) >= 5:
                        sentence = " ".join(self.word_buffer)
                        self.word_buffer = []
                        
                        # 这里可以添加更多的后处理逻辑
                        # 比如：自动添加标点、纠正错别字等
                        processed = self.add_punctuation(sentence)
                        print(f"处理后: {processed}")
                        
        except Exception as e:
            logger.error(f"后处理错误: {e}")
    
    def add_punctuation(self, text):
        """简单的标点添加（实际应用中可以使用更复杂的模型）"""
        # 这里只是一个简单示例
        text = text.strip()
        if text and not text[-1] in ".!?。！？":
            text += "。"
        return text

7. 常见问题与解决方案

在实际使用中，你可能会遇到一些问题。这里列出一些常见问题及其解决方法：

问题1：连接超时或失败

• 检查API密钥是否正确
• 检查网络连接，特别是能否访问WebSocket服务
• 尝试使用不同的地域端点（北京、新加坡、美国）

问题2：识别结果不准确

• 确保音频格式正确：16kHz采样率，16位PCM，单声道
• 检查环境噪音，尝试在安静环境下测试
• 调整VAD参数，适应不同的说话风格

问题3：延迟过高

• 减少音频块的大小（但不要小于1600字节）
• 检查网络延迟，尽量使用就近的地域端点
• 考虑使用Qwen3-ASR-0.6B模型，它在效率上更有优势

问题4：内存或CPU占用过高

• 确保及时清理不再使用的音频数据
• 考虑使用异步IO，避免阻塞主线程
• 对于长时间运行的应用，定期重启连接

8. 实际应用案例

让我们看几个实际的应用场景，了解如何将流式ASR集成到真实项目中。

8.1 实时会议记录系统

class MeetingTranscriber:
    """会议实时转录系统"""
    
    def __init__(self):
        self.asr_client = RobustRealtimeASR(max_retries=5)
        self.participants = {}  # 存储不同说话人的记录
        self.current_speaker = None
        
    def start_meeting(self):
        """开始会议转录"""
        print("会议转录系统启动...")
        
        if not self.asr_client.connect_with_retry():
            print("连接失败，请检查网络和API密钥")
            return
        
        # 开始从默认音频设备采集
        self.asr_client.start_microphone()
        
        # 在实际应用中，这里可以集成说话人识别
        # 暂时假设只有一个说话人
        self.current_speaker = "主持人"
        self.participants[self.current_speaker] = []
        
        print(f"开始转录，当前说话人: {self.current_speaker}")
        print("-" * 50)
        
        # 转录30分钟
        self.asr_client.stream_from_microphone(duration_seconds=1800)
        
    def save_transcript(self, filename="meeting_transcript.txt"):
        """保存转录结果"""
        with open(filename, 'w', encoding='utf-8') as f:
            f.write(f"会议记录\n")
            f.write(f"生成时间: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
            f.write("=" * 50 + "\n\n")
            
            for speaker, sentences in self.participants.items():
                if sentences:
                    f.write(f"{speaker}:\n")
                    for sentence in sentences:
                        f.write(f"  - {sentence}\n")
                    f.write("\n")
        
        print(f"转录结果已保存到: {filename}")

8.2 实时字幕生成器

class LiveSubtitleGenerator:
    """直播实时字幕生成"""
    
    def __init__(self, output_file="subtitles.srt"):
        self.asr_client = PostProcessASR()
        self.subtitles = []
        self.subtitle_index = 1
        self.start_time = None
        self.output_file = output_file
        
    def generate_subtitle(self, text, duration_seconds=3):
        """生成一个字幕片段"""
        if not self.start_time:
            self.start_time = time.time()
        
        current_time = time.time() - self.start_time
        
        # 计算时间戳（SRT格式）
        start_timestamp = self.format_timestamp(current_time)
        end_timestamp = self.format_timestamp(current_time + duration_seconds)
        
        subtitle = {
            'index': self.subtitle_index,
            'start': start_timestamp,
            'end': end_timestamp,
            'text': text
        }
        
        self.subtitles.append(subtitle)
        self.subtitle_index += 1
        
        # 实时输出到控制台（实际应用中可能输出到视频流）
        print(f"[{start_timestamp} --> {end_timestamp}]")
        print(f"{text}\n")
        
    def format_timestamp(self, seconds):
        """将秒数格式化为SRT时间戳"""
        hours = int(seconds // 3600)
        minutes = int((seconds % 3600) // 60)
        secs = int(seconds % 60)
        millis = int((seconds - int(seconds)) * 1000)
        
        return f"{hours:02d}:{minutes:02d}:{secs:02d},{millis:03d}"
    
    def save_subtitles(self):
        """保存为SRT字幕文件"""
        with open(self.output_file, 'w', encoding='utf-8') as f:
            for sub in self.subtitles:
                f.write(f"{sub['index']}\n")
                f.write(f"{sub['start']} --> {sub['end']}\n")
                f.write(f"{sub['text']}\n\n")
        
        print(f"字幕已保存到: {self.output_file}")

9. 总结

流式语音识别技术正在改变我们与机器交互的方式。通过Qwen3-ASR的实时接口，我们能够构建出响应迅速、体验自然的语音应用。

从技术实现的角度看，关键是要理解WebSocket协议的工作方式，以及如何正确地分块发送音频数据。在实际应用中，还需要考虑网络稳定性、错误处理和性能优化等问题。

我建议你从简单的文件转录开始，逐步尝试更复杂的实时应用。过程中可能会遇到各种问题，但这也是学习的一部分。Qwen3-ASR的文档和社区资源都很丰富，遇到困难时可以多查阅相关资料。

流式处理的一个有趣之处在于，它打开了许多新的可能性。不仅仅是转录文字，你还可以实时分析语音情感、检测关键词、或者与其他AI服务结合创建更智能的应用。随着技术的不断进步，实时语音交互会变得越来越普及，掌握这项技术无疑会让你在AI应用开发中占据先机。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。