Qwen3-ASR-1.7B流式推理实战：实时语音转文字系统搭建

你是不是也遇到过这样的场景：开线上会议时，想实时看到文字记录；看外语视频时，希望能有即时字幕；或者开发一个语音助手，需要它能边听边回答。这些需求的核心，都是一个能“边听边转”的实时语音识别系统。

过去搭建这样的系统，要么延迟高得让人着急，要么准确率差强人意。现在，有了Qwen3-ASR-1.7B，事情变得简单多了。这个模型不仅识别准，还原生支持流式推理，这意味着它能像人耳听人脑想一样，处理源源不断的音频流，实现真正的低延迟转写。

今天，我就手把手带你，用Qwen3-ASR-1.7B搭建一个属于自己的实时语音转文字系统。从环境准备到代码实现，再到效果优化，咱们一步步来。即使你之前没怎么接触过语音识别，跟着做也能跑起来。

1. 环境准备与模型获取

工欲善其事，必先利其器。咱们先把“厨房”收拾好，把“食材”备齐。

1.1 创建虚拟环境

为了避免包版本冲突，强烈建议使用虚拟环境。打开你的终端，执行以下命令：

# 创建并激活一个名为 qwen_asr 的虚拟环境（以conda为例）
conda create -n qwen_asr python=3.10
conda activate qwen_asr

# 如果你习惯用 venv
# python -m venv qwen_asr_env
# source qwen_asr_env/bin/activate  # Linux/Mac
# qwen_asr_env\Scripts\activate  # Windows

1.2 安装核心依赖

Qwen3-ASR的推理主要依赖 transformers 和 vLLM。vLLM 是一个高性能的推理库，能极大提升流式推理的效率。

# 安装 PyTorch（请根据你的CUDA版本选择，以下以CUDA 12.1为例）
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121

# 安装 transformers 和 vLLM
pip install transformers vllm

# 安装一些辅助库，用于音频处理
pip install soundfile pydub websockets

1.3 获取模型

模型可以从 Hugging Face 或 ModelScope 下载。这里我们使用 Hugging Face。由于模型较大（约3.4GB），下载需要一些时间。

from transformers import AutoModelForSpeechSeq2Seq, AutoProcessor
import torch

# 指定模型ID
model_id = "Qwen/Qwen3-ASR-1.7B"

# 下载并加载模型与处理器（首次运行会自动下载）
print(f"正在下载模型 {model_id}，请耐心等待...")
model = AutoModelForSpeechSeq2Seq.from_pretrained(
    model_id,
    torch_dtype=torch.float16,  # 使用半精度以节省显存
    device_map="auto",          # 自动分配设备（GPU/CPU）
    trust_remote_code=True
)
processor = AutoProcessor.from_pretrained(model_id, trust_remote_code=True)
print("模型加载完成！")

如果网络环境不佳，你也可以先通过 git lfs clone 命令将模型仓库克隆到本地，然后从本地路径加载。

2. 理解流式推理：它为什么能“边听边转”？

在写代码之前，咱们先花两分钟搞明白，流式推理到底是怎么一回事。这能帮你更好地理解后面的代码逻辑。

你可以把非流式推理想象成“交卷式批改”：你必须把整段音频（比如一个1小时的会议录音）全部录完，交给模型，模型从头到尾看一遍，然后一次性给你全部的转写结果。等待时间长，且无法中途获取结果。

而流式推理，更像是“随堂测验批改”：音频像水流一样，一段一段地实时传给模型。模型每收到一小段（比如几百毫秒的音频），就立刻处理这一小段，并输出当前能识别出的文字。后面的音频来了，它就在之前的基础上继续识别和输出。

Qwen3-ASR通过一个叫 动态Flash Attention窗口 的机制来实现这一点。简单说，模型有一个“记忆窗口”，它只关注当前窗口内的音频和之前一点点上下文，而不是回顾整个历史。这让它既能保持一定的连贯性，又能实现极低的处理延迟。

所以，我们搭建系统的核心任务就是：如何把连续的音频流，切成小块喂给模型，并实时收集模型的输出。

3. 从文件到实时：三步上手流式推理

咱们由浅入深，先看看怎么处理一个完整的音频文件（模拟流式），再过渡到真正的实时麦克风输入。

3.1 第一步：用流式方式处理一个音频文件

假设你有一个 meeting.wav 文件。我们通过分段读取来模拟流式输入。

import soundfile as sf
import numpy as np

def transcribe_audio_streaming(file_path, chunk_duration_ms=1000):
    """
    以流式方式转录整个音频文件。
    :param file_path: 音频文件路径
    :param chunk_duration_ms: 每次处理的音频块时长（毫秒）
    """
    # 读取音频文件的基本信息
    audio_data, sample_rate = sf.read(file_path)
    # 确保是单声道
    if len(audio_data.shape) > 1:
        audio_data = audio_data.mean(axis=1)
    
    chunk_size = int(sample_rate * (chunk_duration_ms / 1000.0))
    total_chunks = len(audio_data) // chunk_size + (1 if len(audio_data) % chunk_size else 0)
    
    print(f"开始流式转录，采样率：{sample_rate}Hz，总块数：{total_chunks}")
    
    # 这里为了简化，我们循环模拟“流”
    for i in range(total_chunks):
        start = i * chunk_size
        end = start + chunk_size
        audio_chunk = audio_data[start:end]
        
        if len(audio_chunk) == 0:
            continue
        
        # 准备模型输入
        inputs = processor(
            audio=audio_chunk,
            sampling_rate=sample_rate,
            return_tensors="pt",
            padding=True
        ).to(model.device)
        
        # 使用 generate 进行流式推理（注意 streamer 参数）
        # 在实际完整流式API中，这里应使用专门的流式生成方法。
        # 以下代码演示逻辑，生产环境需参考官方流式示例。
        with torch.no_grad():
            # 模拟：在实际中，这里应调用支持流式输出的generate
            # 例如：outputs = model.generate(**inputs, streamer=streamer, ...)
            # 我们这里先用非流式生成示意结果
            predicted_ids = model.generate(**inputs, max_new_tokens=256)
            transcription = processor.batch_decode(predicted_ids, skip_special_tokens=True)[0]
        
        print(f"[块 {i+1}/{total_chunks}] 转录: {transcription}")
    
    print("文件转录完成。")

# 使用示例
# transcribe_audio_streaming("meeting.wav")

这段代码的关键在于分块循环。它把长音频切成1秒（可调）的小块，依次处理，并打印出每一块的识别结果。在真正的流式API中，model.generate 会配合一个 streamer 对象，能够实时地、逐词（token）地输出结果，而不是等整句话说完。

3.2 第二步：接入真正的麦克风流

处理文件是“模拟”，现在我们来点真的。下面这段代码会从你的麦克风实时采集音频，并实时转写。

import pyaudio
import numpy as np
import threading
import queue
import time

# 音频参数
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 16000  # Qwen3-ASR推荐16kHz
CHUNK_DURATION_MS = 500  # 每次处理的音频时长（毫秒）
CHUNK = int(RATE * CHUNK_DURATION_MS / 1000)  # 计算对应的样本数

# 创建一个队列，用于在音频采集线程和识别线程之间传递数据
audio_queue = queue.Queue()

def audio_capture_callback(in_data, frame_count, time_info, status):
    """PyAudio回调函数，负责采集音频数据并放入队列。"""
    audio_data = np.frombuffer(in_data, dtype=np.int16).astype(np.float32) / 32768.0
    audio_queue.put(audio_data)
    return (None, pyaudio.paContinue)

def start_audio_capture():
    """启动麦克风音频采集线程。"""
    p = pyaudio.PyAudio()
    stream = p.open(format=FORMAT,
                    channels=CHANNELS,
                    rate=RATE,
                    input=True,
                    frames_per_buffer=CHUNK,
                    stream_callback=audio_capture_callback)
    stream.start_stream()
    print("麦克风采集已启动... (按 Ctrl+C 停止)")
    return p, stream

def transcribe_from_mic():
    """从麦克风队列中获取音频并实时转写。"""
    p, stream = start_audio_capture()
    
    try:
        while True:
            # 从队列中获取音频块（这里会阻塞直到有数据）
            audio_chunk = audio_queue.get()
            
            # 准备输入
            inputs = processor(
                audio=audio_chunk,
                sampling_rate=RATE,
                return_tensors="pt",
                padding=True
            ).to(model.device)
            
            # 进行推理（此处仍为非流式生成，仅作演示）
            # 重要：在生产环境中，此处应使用模型的流式推理接口。
            # 例如，利用vLLM的异步流式接口。
            with torch.no_grad():
                predicted_ids = model.generate(**inputs, max_new_tokens=64)  # 限制生成token数以降低延迟
                text = processor.batch_decode(predicted_ids, skip_special_tokens=True)[0]
            
            if text and text.strip():  # 只打印有内容的识别结果
                print(f"> {text.strip()}")
                
    except KeyboardInterrupt:
        print("\n正在停止...")
    finally:
        stream.stop_stream()
        stream.close()
        p.terminate()
        print("已停止采集。")

# 运行这个函数开始实时转写
# transcribe_from_mic()

运行这段代码，对着麦克风说话，你就能在控制台看到几乎实时的转写文字了。这里的延迟主要来自我们简单的“采集-处理-打印”循环。CHUNK_DURATION_MS 设置得越小，延迟感越低，但太小的音频块可能包含的语音信息不足，影响识别准确性。500毫秒是一个不错的起点。

3.3 第三步：构建一个简单的WebSocket服务端

如果想让转写结果通过网络发送给其他应用（比如一个网页字幕端），WebSocket是个好选择。下面我们实现一个简单的服务端。

import asyncio
import websockets
import json
import numpy as np
from io import BytesIO

# 假设我们有一个支持真正流式输出的推理函数（这里用非流式模拟其接口）
async def handle_websocket(websocket, path):
    print("客户端已连接")
    try:
        async for message in websocket:
            # 假设客户端发送的是二进制音频数据（如PCM格式）
            audio_data = np.frombuffer(message, dtype=np.int16).astype(np.float32) / 32768.0
            
            # 调用流式推理函数（此处为模拟）
            # 真实场景下，这里应该调用一个能yield中间结果的生成器函数
            inputs = processor(
                audio=audio_data,
                sampling_rate=RATE,
                return_tensors="pt",
                padding=True
            ).to(model.device)
            
            with torch.no_grad():
                predicted_ids = model.generate(**inputs, max_new_tokens=64, do_sample=False)
                transcription = processor.batch_decode(predicted_ids, skip_special_tokens=True)[0]
            
            # 将结果以JSON格式发送回客户端
            response = {"text": transcription, "is_final": True}  # 流式中会有 is_final=False 的中间结果
            await websocket.send(json.dumps(response, ensure_ascii=False))
            
    except websockets.exceptions.ConnectionClosed:
        print("客户端连接断开")

async def main():
    # 启动WebSocket服务器，监听8765端口
    async with websockets.serve(handle_websocket, "localhost", 8765):
        print("WebSocket流式ASR服务器已启动， ws://localhost:8765")
        await asyncio.Future()  # 永久运行

# 在另一个终端运行: python -m websockets ws://localhost:8765 进行测试
# asyncio.run(main())

这个服务端接收客户端发来的音频二进制数据，识别后返回文字。在完整的流式实现中，服务端应该在模型每输出一个词或一个片段时，就立即发送一个 {"text": “部分结果”, "is_final": False} 的消息，最后再发送一个 "is_final": True 的消息。这需要模型推理端也支持真正的流式输出。

4. 关键技巧与避坑指南

在实际搭建过程中，你可能会遇到一些问题。这里分享几个我踩过坑后总结的经验。

1. 延迟与准确率的权衡

• 块大小（Chunk Size）：这是最重要的参数。太小（如200ms），上下文信息少，模型容易出错，特别是句子开头；太大（如2秒），延迟明显。建议从500ms开始调试。
• 词汇端点检测（VAD）：与其固定时间分块，不如在检测到语音时才送出一块音频。这能减少静音部分的无效处理，提升响应速度。你可以集成一个简单的VAD库（如 webrtcvad）。

2. 处理长句和上下文 流式推理的“记忆窗口”有限。如果用户说了一个很长的句子，模型在中间可能会因为忘记开头而跑偏。对于会议、访谈等场景，可以在服务端做一些后处理，比如用一个简单的规则或小模型，将连续输出的片段合并成更通顺的句子。

3. 资源管理与优化

• 显存：使用 torch.float16 半精度能显著减少显存占用。如果显存紧张，可以考虑使用 Qwen3-ASR-0.6B 版本，它在精度和效率间取得了更好的平衡。
• vLLM异步服务：对于高并发场景（如多个用户同时使用），一定要部署基于vLLM的异步推理服务。官方提供的推理框架支持这个，它能大幅提升吞吐量。
• 预热：在服务启动后，先用一段测试音频“预热”一下模型，避免第一个用户的请求因模型加载而超时。

4. 关于“真正”的流式API 本文的代码示例为了清晰易懂，在推理部分使用了非流式的 generate 方法进行模拟。要获得最佳的、逐词输出的流式体验，你需要使用Qwen3-ASR官方仓库中提供的、基于vLLM的流式推理示例。那通常会涉及创建一个 AsyncLLMEngine 并使用 generate 方法的 streaming 模式，代码结构会有所不同，但核心的“音频流处理-发送-接收结果流”的思想是相通的。

5. 把它用起来：几个落地场景想法

系统搭好了，它能干嘛？这里给你几个灵感：

• 实时会议字幕：结合上面的WebSocket例子，写一个网页客户端，就能在开会时看到实时字幕。
• 语音助手核心：接收用户的语音流，实时转成文字，再交给大语言模型（LLM）去理解和生成回答，实现流畅的语音对话。
• 直播字幕生成：接入直播流的音频，为观众提供实时字幕，特别适合教育、跨境直播等场景。
• 录音笔即时整理：虽然录音笔是离线录制，但可以在连接手机后，用流式的方式快速转写，用户几乎感觉不到等待。

---

整体走下来，用Qwen3-ASR-1.7B搭建一个实时语音转文字系统，并没有想象中那么复杂。它的流式推理能力是开箱即用的，这为我们省去了大量的底层优化工作。核心就是理解“数据流”的概念，并设计好音频采集、分块、推理和结果返回的管道。

我建议你从处理一个本地音频文件开始，跑通最基本的流程。然后尝试接入麦克风，感受一下实时的效果。最后，再根据你的具体应用场景，去设计网络服务或集成其他功能。过程中遇到问题，多去看看官方GitHub仓库的Issue和讨论，社区通常很活跃。

动手试试吧，当你第一次对着麦克风说话，看到文字几乎同步出现在屏幕上时，那种感觉还是挺棒的。

获取更多AI镜像

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