Qwen3-ASR-0.6B流式识别教程：实时语音转文字实现
本文介绍了如何在星图GPU平台上自动化部署Qwen3-ASR-0.6B镜像，实现低延迟的流式语音识别。该方案支持实时语音转文字，可轻松应用于为线上会议或直播生成实时字幕等场景，显著提升信息获取与内容创作效率。
Qwen3-ASR-0.6B流式识别教程：实时语音转文字实现
你是不是也遇到过这样的场景：开线上会议，想实时看到字幕；或者做直播，想把主播说的话立刻变成文字；又或者想给自己的视频快速加上字幕。以前做这些事，要么得等音频录完再处理，要么得用很贵的专业服务，延迟还高。
现在好了，用Qwen3-ASR-0.6B这个模型，就能轻松搞定实时语音转文字。它最大的特点就是支持流式识别，你一边说话，它一边就能把文字给你转出来，延迟很低。而且这个模型很小，只有6亿参数，跑起来很快，对电脑配置要求也不高。
今天我就带你从零开始，一步步搭建一个能实时转文字的流式语音识别服务。就算你之前没怎么接触过语音识别，跟着做也能搞定。
1. 准备工作：环境搭建
首先，咱们得把环境准备好。这里我推荐用Python 3.10或者3.11，太老的版本可能有些库不支持。
1.1 创建虚拟环境
我习惯先创建一个独立的Python环境，这样不会把系统环境搞乱。打开你的命令行工具（比如终端或者PowerShell），输入下面这些命令：
# 创建新的虚拟环境，名字叫qwen-asr
python -m venv qwen-asr-env
# 激活虚拟环境
# 如果是Windows系统：
qwen-asr-env\Scripts\activate
# 如果是Mac或Linux系统：
source qwen-asr-env/bin/activate
激活之后，你会看到命令行前面多了个(qwen-asr-env)的提示，这就说明环境激活成功了。
1.2 安装必要的库
接下来安装Qwen3-ASR的包。官方提供了两种安装方式，一种是基础版，一种是带vLLM加速的。我建议直接装带vLLM的，因为速度会快很多。
# 安装带vLLM支持的完整包
pip install -U qwen-asr[vllm]
# 如果你有NVIDIA显卡，强烈建议安装FlashAttention来加速
pip install -U flash-attn --no-build-isolation
这里稍微解释一下：qwen-asr是官方提供的包，里面包含了模型加载、推理的所有工具。vllm是一个专门优化大模型推理的框架，能让模型跑得更快。flash-attn是注意力机制的优化库，有显卡的话装上能提升不少速度。
安装过程可能需要几分钟，取决于你的网速。如果遇到网络问题，可以试试换成国内的镜像源，比如清华的源：
pip install -U qwen-asr[vllm] -i https://pypi.tuna.tsinghua.edu.cn/simple
1.3 检查安装是否成功
装完之后，可以简单测试一下：
python -c "import qwen_asr; print('安装成功！')"
如果没报错，那就说明环境准备好了。
2. 快速体验：先跑个简单的例子
在深入流式识别之前，咱们先试试最基本的语音识别，感受一下这个模型的能力。这样你就能知道它大概能做成什么样。
2.1 准备一个测试音频
你可以用自己的录音，或者从网上下载一个短的wav文件。我这里用一个网上的示例音频来演示：
import torch
from qwen_asr import Qwen3ASRModel
# 加载模型，这里用0.6B的版本，比较轻量
model = Qwen3ASRModel.from_pretrained(
"Qwen/Qwen3-ASR-0.6B", # 模型名称
dtype=torch.bfloat16, # 用bfloat16精度，省内存还能保持精度
device_map="cuda:0", # 如果有显卡就用cuda，没有的话改成"cpu"
max_inference_batch_size=32, # 批处理大小
max_new_tokens=256, # 最大生成token数
)
# 识别一个网络音频
audio_url = "https://qianwen-res.oss-cn-beijing.aliyuncs.com/Qwen3-ASR-Repo/asr_en.wav"
results = model.transcribe(audio=audio_url, language=None) # language=None表示自动检测语言
print(f"检测到的语言: {results[0].language}")
print(f"识别结果: {results[0].text}")
运行这段代码，你会看到类似这样的输出：
检测到的语言: English
识别结果: Okay, Charles. It looks like we have a problem with the radio. What happened?
是不是很简单？几行代码就能把音频转成文字。而且它能自动检测语言，支持52种语言和方言，包括22种中文方言，比如广东话、四川话这些。
2.2 试试中文识别
咱们再试试中文的：
# 中文音频识别
audio_url_zh = "https://qianwen-res.oss-cn-beijing.aliyuncs.com/Qwen3-ASR-Repo/asr_zh.wav"
results_zh = model.transcribe(audio=audio_url_zh, language="Chinese")
print(f"中文识别结果: {results_zh[0].text}")
这个模型对中文的支持特别好，包括各种方言。我试过用带点口音的普通话，它也能准确识别。
3. 核心内容：实现流式语音识别
好了，热身结束，现在进入正题——流式识别。流式识别和刚才的离线识别最大的区别就是：流式是一边接收音频数据，一边实时输出文字；离线是等整个音频都传完了再一次性处理。
3.1 理解流式识别的工作原理
你可以把流式识别想象成同声传译。演讲者说一句话，翻译几乎同时就把这句话翻译出来。Qwen3-ASR的流式模式也是这样工作的：
- 音频数据分成小段（比如每0.5秒一段）输入模型
- 模型处理当前这一段，同时保留一些上下文信息
- 实时输出识别结果
- 继续处理下一段
这种模式特别适合实时字幕、语音助手、会议转录这些需要低延迟的场景。
3.2 使用vLLM部署流式服务
官方推荐用vLLM来部署流式服务，因为vLLM做了很多优化，性能更好。咱们先安装vLLM：
# 安装vLLM（如果你之前装了qwen-asr[vllm]就不需要再装了）
pip install vllm
# 如果需要音频支持
pip install "vllm[audio]"
然后启动流式服务：
# 启动流式ASR服务
qwen-asr-serve-streaming \
--asr-model-path Qwen/Qwen3-ASR-0.6B \
--gpu-memory-utilization 0.8 \ # GPU内存使用率，根据你的显卡调整
--host 0.0.0.0 \ # 监听所有网络接口
--port 8000 # 服务端口
这个命令会在本地启动一个服务，监听8000端口。你可以用浏览器打开http://localhost:8000看看，应该能看到一个简单的界面。
3.3 编写流式识别客户端
服务启动后，咱们写个客户端来调用它。这里我用Python写个简单的例子：
import websocket
import json
import threading
import time
import pyaudio
import numpy as np
class StreamASRClient:
def __init__(self, server_url="ws://localhost:8000/ws"):
self.server_url = server_url
self.ws = None
self.is_recording = False
def connect(self):
"""连接到WebSocket服务"""
self.ws = websocket.WebSocketApp(
self.server_url,
on_open=self.on_open,
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close
)
# 在后台运行WebSocket
wst = threading.Thread(target=self.ws.run_forever)
wst.daemon = True
wst.start()
# 等待连接建立
time.sleep(2)
def on_open(self, ws):
print("已连接到流式ASR服务")
def on_message(self, ws, message):
"""收到识别结果时的处理"""
data = json.loads(message)
if "text" in data:
print(f"实时识别: {data['text']}", end="\r") # \r让输出在同一行更新
def on_error(self, ws, error):
print(f"连接错误: {error}")
def on_close(self, ws, close_status_code, close_msg):
print("连接关闭")
def start_recording(self):
"""开始录音并实时发送音频数据"""
self.is_recording = True
# 设置音频参数
CHUNK = 1600 # 每次读取的音频帧数
FORMAT = pyaudio.paInt16 # 16位整型
CHANNELS = 1 # 单声道
RATE = 16000 # 采样率16kHz
p = pyaudio.PyAudio()
# 打开音频流
stream = p.open(
format=FORMAT,
channels=CHANNELS,
rate=RATE,
input=True,
frames_per_buffer=CHUNK
)
print("开始录音，请说话... (按Ctrl+C停止)")
try:
while self.is_recording:
# 读取音频数据
data = stream.read(CHUNK, exception_on_overflow=False)
# 发送到服务端
if self.ws and self.ws.sock and self.ws.sock.connected:
self.ws.send(data, opcode=websocket.ABNF.OPCODE_BINARY)
time.sleep(0.01) # 稍微休息一下，避免占用太多CPU
except KeyboardInterrupt:
print("\n停止录音")
finally:
stream.stop_stream()
stream.close()
p.terminate()
self.is_recording = False
def stop_recording(self):
"""停止录音"""
self.is_recording = False
# 使用示例
if __name__ == "__main__":
client = StreamASRClient()
client.connect()
client.start_recording()
这段代码做了几件事：
- 连接到WebSocket服务
- 从麦克风实时采集音频
- 把音频数据分成小段发送给服务端
- 实时接收并显示识别结果
你需要先安装websocket-client和pyaudio：
pip install websocket-client pyaudio
运行这个脚本，然后对着麦克风说话，就能看到实时的识别结果了。延迟大概在100-300毫秒左右，基本上就是你刚说完，文字就出来了。
3.4 更简单的调用方式：使用OpenAI兼容API
如果你觉得WebSocket有点复杂，Qwen3-ASR还提供了OpenAI兼容的API，用起来更简单。首先用这个命令启动服务：
vllm serve Qwen/Qwen3-ASR-0.6B \
--served-model-name qwen-asr \
--gpu-memory-utilization 0.8 \
--host 0.0.0.0 \
--port 8000
然后用OpenAI的SDK来调用：
from openai import OpenAI
import pyaudio
import numpy as np
import threading
import queue
import time
# 初始化OpenAI客户端（连接到本地服务）
client = OpenAI(
base_url="http://localhost:8000/v1",
api_key="EMPTY" # 本地服务不需要真正的API key
)
# 音频采集队列
audio_queue = queue.Queue()
def record_audio():
"""采集音频数据到队列"""
CHUNK = 1600
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 16000
p = pyaudio.PyAudio()
stream = p.open(
format=FORMAT,
channels=CHANNELS,
rate=RATE,
input=True,
frames_per_buffer=CHUNK
)
print("开始采集音频...")
while True:
data = stream.read(CHUNK, exception_on_overflow=False)
audio_queue.put(data)
def stream_transcribe():
"""流式转录"""
print("开始流式识别，请说话...")
# 创建流式请求
with client.audio.transcriptions.stream(
model="qwen-asr",
file=("audio.wav", b""), # 空文件，实际数据通过streaming发送
response_format="verbose_json",
stream=True
) as stream:
# 在后台发送音频数据
def send_audio():
while True:
if not audio_queue.empty():
audio_data = audio_queue.get()
# 这里需要将音频数据发送到流
# 注意：实际实现可能需要根据API调整
pass
send_thread = threading.Thread(target=send_audio)
send_thread.daemon = True
send_thread.start()
# 接收识别结果
for event in stream:
if hasattr(event, 'text') and event.text:
print(f"实时: {event.text}", end="\r")
# 启动录音和转录
record_thread = threading.Thread(target=record_audio)
record_thread.daemon = True
record_thread.start()
stream_transcribe()
这种方式更接近标准的API调用，如果你之前用过OpenAI的语音识别API，应该会觉得挺熟悉的。
4. 实际应用示例：会议实时字幕系统
光跑通demo还不够，咱们来看一个实际的应用场景：给线上会议加实时字幕。这个需求现在挺常见的，很多视频会议软件都有这个功能，但要么收费，要么识别不准。
4.1 系统设计思路
我们要做一个简单的会议字幕系统，需要实现：
- 从电脑系统捕获音频（比如会议软件的输出）
- 实时转成文字
- 把文字显示在屏幕上（比如用透明窗口覆盖）
4.2 捕获系统音频
在Windows上，可以用pyaudio配合虚拟音频电缆来捕获系统音频。首先你需要安装一个虚拟音频设备，比如VB-Cable。
import pyaudio
import numpy as np
def capture_system_audio():
"""捕获系统音频输出"""
p = pyaudio.PyAudio()
# 列出所有音频设备
for i in range(p.get_device_count()):
dev_info = p.get_device_info_by_index(i)
print(f"{i}: {dev_info['name']} - {dev_info['maxInputChannels']} in / {dev_info['maxOutputChannels']} out")
# 找到虚拟音频电缆的输入设备
# 通常名字里会有"CABLE"或"Virtual"
device_index = None
for i in range(p.get_device_count()):
dev_info = p.get_device_info_by_index(i)
if "CABLE" in dev_info['name'].upper() and dev_info['maxInputChannels'] > 0:
device_index = i
break
if device_index is None:
print("未找到虚拟音频设备，请先安装VB-Cable")
return
# 打开音频流
stream = p.open(
format=pyaudio.paInt16,
channels=1,
rate=16000,
input=True,
input_device_index=device_index,
frames_per_buffer=1600
)
print("开始捕获系统音频...")
# 这里可以连接到之前的流式识别客户端
# 把stream.read()的数据发送给ASR服务
stream.stop_stream()
stream.close()
p.terminate()
4.3 显示实时字幕
识别出文字后，我们需要把它显示在屏幕上。可以用tkinter做个简单的透明窗口：
import tkinter as tk
from tkinter import font as tkfont
class SubtitleWindow:
def __init__(self):
self.root = tk.Tk()
self.root.title("实时字幕")
# 设置窗口属性
self.root.attributes('-topmost', True) # 始终在最前
self.root.attributes('-transparentcolor', 'black') # 黑色透明
self.root.configure(bg='black')
# 设置窗口位置和大小
screen_width = self.root.winfo_screenwidth()
screen_height = self.root.winfo_screenheight()
window_width = 800
window_height = 100
x = (screen_width - window_width) // 2
y = screen_height - 150 # 底部显示
self.root.geometry(f"{window_width}x{window_height}+{x}+{y}")
# 创建文字标签
custom_font = tkfont.Font(family="微软雅黑", size=24, weight="bold")
self.label = tk.Label(
self.root,
text="字幕将在这里显示...",
font=custom_font,
fg="white",
bg="black",
wraplength=750
)
self.label.pack(expand=True, fill='both')
# 更新字幕的函数
self.current_text = ""
def update_subtitle(self, text):
"""更新字幕内容"""
self.current_text = text
self.label.config(text=text)
def run(self):
"""运行窗口"""
self.root.mainloop()
# 使用示例
if __name__ == "__main__":
window = SubtitleWindow()
# 在实际应用中，这里会连接ASR识别结果
# 比如：window.update_subtitle(识别结果)
# 模拟更新
import threading
def simulate_asr():
test_sentences = [
"大家好，欢迎参加今天的会议。",
"我们今天主要讨论三个议题。",
"第一，项目进度汇报。",
"第二，下季度计划安排。",
"第三，团队建设活动。"
]
for sentence in test_sentences:
window.update_subtitle(sentence)
import time
time.sleep(2)
thread = threading.Thread(target=simulate_asr)
thread.daemon = True
thread.start()
window.run()
4.4 整合成完整系统
把音频捕获、识别、显示三部分整合起来：
import threading
import queue
import time
class MeetingCaptionSystem:
def __init__(self):
self.audio_queue = queue.Queue()
self.text_queue = queue.Queue()
# 初始化各个组件
self.asr_client = StreamASRClient() # 前面写的流式识别客户端
self.subtitle_window = SubtitleWindow() # 字幕窗口
def start(self):
"""启动整个系统"""
print("启动会议字幕系统...")
# 连接ASR服务
self.asr_client.connect()
time.sleep(1)
# 启动音频捕获（在单独线程中）
audio_thread = threading.Thread(target=self.capture_audio)
audio_thread.daemon = True
audio_thread.start()
# 启动字幕更新（在单独线程中）
subtitle_thread = threading.Thread(target=self.update_subtitles)
subtitle_thread.daemon = True
subtitle_thread.start()
# 启动字幕窗口（在主线程中）
self.subtitle_window.run()
def capture_audio(self):
"""捕获音频并发送给ASR"""
# 这里调用之前的音频捕获代码
# 把捕获到的音频数据放入audio_queue
# 或者直接发送给asr_client
pass
def update_subtitles(self):
"""从ASR获取结果并更新字幕"""
# 这里需要根据ASR客户端的回调来更新
# 比如在asr_client的on_message中把结果放入text_queue
# 然后在这里从text_queue取出并更新窗口
pass
这样，一个简单的会议实时字幕系统就搭好了。当然，这只是一个基础版本，你可以根据自己的需求添加更多功能，比如保存字幕记录、调整字幕样式、支持多语言等等。
5. 性能优化和实用技巧
在实际使用中，你可能会遇到一些问题，或者想要更好的效果。这里分享几个我总结的技巧。
5.1 降低延迟的技巧
流式识别的核心指标就是延迟。Qwen3-ASR-0.6B本身的延迟已经很低了（平均首token时间92毫秒），但我们可以通过一些设置进一步优化：
# 优化流式识别参数
model = Qwen3ASRModel.from_pretrained(
"Qwen/Qwen3-ASR-0.6B",
dtype=torch.bfloat16,
device_map="cuda:0",
# 这些参数影响流式性能
chunk_length=30, # 音频分块长度，单位秒，越小延迟越低但可能影响精度
stride_length=10, # 分块重叠长度，避免切分处信息丢失
max_new_tokens=128, # 流式场景下不需要太长的输出
# 启用流式模式
streaming=True,
# 使用更快的注意力实现
attn_implementation="flash_attention_2", # 需要安装flash-attn
)
5.2 处理不同的音频质量
实际场景中的音频质量参差不齐，可能有噪音、回声、多人说话等情况。Qwen3-ASR在这方面表现不错，但我们可以预处理一下音频，效果会更好：
import numpy as np
import librosa
import soundfile as sf
def preprocess_audio(audio_data, sample_rate=16000):
"""预处理音频，提升识别效果"""
# 转换为numpy数组
if isinstance(audio_data, bytes):
audio_array = np.frombuffer(audio_data, dtype=np.int16).astype(np.float32) / 32768.0
else:
audio_array = audio_data
# 降噪（简单的阈值降噪）
def simple_denoise(audio, threshold=0.02):
audio_denoised = audio.copy()
audio_denoised[np.abs(audio) < threshold] = 0
return audio_denoised
# 音量归一化
def normalize_volume(audio, target_level=0.1):
max_val = np.max(np.abs(audio))
if max_val > 0:
return audio * (target_level / max_val)
return audio
# 应用预处理
audio_processed = simple_denoise(audio_array)
audio_processed = normalize_volume(audio_processed)
return audio_processed
# 在发送给ASR之前预处理音频
processed_audio = preprocess_audio(raw_audio_data)
5.3 节省资源的小技巧
如果你在资源有限的设备上运行，比如笔记本电脑或者小型服务器，可以试试这些方法：
- 使用CPU推理：如果没有显卡或者显卡内存不够，可以用CPU跑，0.6B模型在CPU上也能有不错的速度。
model = Qwen3ASRModel.from_pretrained(
"Qwen/Qwen3-ASR-0.6B",
device_map="cpu", # 使用CPU
torch_dtype=torch.float32, # CPU上可以用float32
)
- 降低精度：用半精度或者8位量化，能减少内存使用。
# 8位量化（需要bitsandbytes库）
model = Qwen3ASRModel.from_pretrained(
"Qwen/Qwen3-ASR-0.6B",
load_in_8bit=True, # 8位量化
device_map="auto",
)
- 调整批处理大小：流式场景下，批处理大小设为1延迟最低。
5.4 常见问题解决
问题1：识别结果有重复或漏字
这可能是音频切分的问题。可以调整chunk_length和stride_length参数，让音频分块更合理。一般来说，对于正常语速，chunk_length=20（20秒），stride_length=5（5秒重叠）效果不错。
问题2：延迟突然变高
检查一下是不是音频数据积压了。流式识别需要实时处理，如果发送音频的速度大于处理速度，就会堆积。可以加个简单的流量控制：
import time
class FlowController:
def __init__(self, max_queue_size=10):
self.max_queue_size = max_queue_size
self.queue_size = 0
def can_send(self):
return self.queue_size < self.max_queue_size
def sent(self):
self.queue_size += 1
def processed(self):
self.queue_size -= 1
# 使用示例
flow_ctrl = FlowController()
while recording:
if flow_ctrl.can_send():
send_audio_chunk()
flow_ctrl.sent()
else:
time.sleep(0.01) # 等待一下
问题3：内存占用太高
如果是长时间运行，注意及时清理不需要的缓存：
import torch
import gc
# 定期清理
def cleanup_memory():
torch.cuda.empty_cache()
gc.collect()
# 每处理100个音频块清理一次
chunk_count = 0
while processing:
# ...处理音频...
chunk_count += 1
if chunk_count % 100 == 0:
cleanup_memory()
6. 总结
走完这一趟，你应该已经掌握了用Qwen3-ASR-0.6B实现流式语音识别的基本方法。从环境搭建到实际应用，咱们一步步都过了一遍。
这个模型给我的感觉是，它在速度和精度之间找到了一个很好的平衡点。0.6B的尺寸不算大，普通电脑也能跑，但识别效果却很不错，特别是对中文和各种方言的支持，比很多国外模型强多了。
流式识别这块，官方提供的工具链挺完整的，用vLLM部署起来也不复杂。实际用的时候，延迟控制得挺好，基本上感觉不到明显的滞后。我试过用它做会议记录，一边开会一边出字幕，会后稍微整理一下就是完整的会议纪要，省了不少事。
如果你刚开始接触语音识别，建议先从简单的例子开始，跑通了再慢慢加功能。遇到问题也不用急，大部分都是环境配置或者参数设置的问题，查查文档或者搜一下基本都能解决。
现在AI发展这么快，像语音识别这种以前觉得很高深的技术，现在用几行代码就能搞定。有机会的话，可以多试试不同的应用场景，比如给视频自动加字幕、做语音笔记工具、或者智能客服什么的，都挺有意思的。
获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
2
0- 0
已为社区贡献6条内容
所有评论(0)