当前位置：首页 > news >正文

Qwen3-ASR-1.7B实操手册：如何用Python API调用模型实现批量音频转写

news 2026/3/27 4:07:17

Qwen3-ASR-1.7B实操手册：如何用Python API调用模型实现批量音频转写

1. 项目简介

Qwen3-ASR-1.7B是基于阿里云通义千问团队开源的中量级语音识别模型开发的本地智能语音转文字工具。相比之前的0.6B版本，这个1.7B版本在复杂长难句和中英文混合语音的识别准确率上有显著提升，特别适合处理会议录音、视频字幕、访谈记录等对精度要求较高的场景。

这个工具最大的特点是完全本地运行，不需要联网，所有音频处理都在本地完成，确保了数据隐私安全。模型针对GPU进行了FP16半精度优化，显存需求约4-5GB，在保证精度的同时兼顾了硬件适配性。

2. 环境准备与安装

2.1 系统要求

在开始之前，请确保你的系统满足以下要求：

Python 3.8或更高版本
NVIDIA GPU（推荐8GB以上显存）
CUDA 11.7或更高版本
至少10GB的可用磁盘空间

2.2 安装依赖包

打开终端或命令提示符，执行以下命令安装必要的依赖：

pip install torch torchaudio transformers streamlit pip install soundfile librosa pydub

这些包分别用于：

torch和torchaudio：深度学习框架和音频处理
transformers：加载和使用语音识别模型
streamlit：构建可视化界面（可选）
soundfile、librosa、pydub：音频文件处理

2.3 验证安装

创建一个简单的测试脚本来验证环境是否正确配置：

import torch print(f"PyTorch版本: {torch.__version__}") print(f"CUDA可用: {torch.cuda.is_available()}") print(f"GPU数量: {torch.cuda.device_count()}") if torch.cuda.is_available(): print(f"当前GPU: {torch.cuda.get_device_name(0)}")

3. 基础概念快速入门

3.1 什么是语音识别

语音识别就像是一个"听觉翻译官"，它能把人说话的声音转换成文字。想象一下，你对着手机说"今天天气怎么样"，手机就能显示出这句话的文字——这就是语音识别在做的事情。

Qwen3-ASR-1.7B是这个领域的专业选手，特别擅长处理复杂的长句子和中英文混合的情况，比如技术会议中经常出现的"这个API的throughput需要optimize一下"这样的内容。

3.2 为什么选择本地部署

使用本地部署的语音识别有几个明显优势：

隐私安全：你的音频数据不会上传到任何服务器，完全在本地处理
无网络依赖：即使没有互联网连接也能正常工作
无使用限制：不像一些在线服务有调用次数限制
响应快速：不需要网络传输，处理速度更快

4. 核心API调用方法

4.1 初始化语音识别模型

首先让我们看看如何加载和使用这个语音识别模型：

from transformers import AutoModelForSpeechSeq2Seq, AutoProcessor import torch # 初始化模型和处理器 model_id = "Qwen/Qwen3-ASR-1.7B" # 加载模型，使用FP16精度节省显存 model = AutoModelForSpeechSeq2Seq.from_pretrained( model_id, torch_dtype=torch.float16, device_map="auto" ) # 加载处理器，负责音频预处理 processor = AutoProcessor.from_pretrained(model_id) print("模型加载完成，准备就绪！")

4.2 单文件语音转写示例

下面是一个完整的单文件转写示例：

import torch from transformers import AutoModelForSpeechSeq2Seq, AutoProcessor from datasets import load_dataset def transcribe_audio(audio_path): """将音频文件转换为文字""" # 加载模型和处理器（实际使用中应该全局初始化一次） model_id = "Qwen/Qwen3-ASR-1.7B" model = AutoModelForSpeechSeq2Seq.from_pretrained( model_id, torch_dtype=torch.float16, device_map="auto" ) processor = AutoProcessor.from_pretrained(model_id) # 加载音频文件 dataset = load_dataset("audio", data_files={"audio": audio_path}) sample = dataset["audio"][0] audio_data = sample["array"] sampling_rate = sample["sampling_rate"] # 处理音频并生成文字 inputs = processor( audio_data, sampling_rate=sampling_rate, return_tensors="pt" ) with torch.no_grad(): outputs = model.generate(**inputs.to(model.device)) # 解码结果 transcription = processor.batch_decode( outputs, skip_special_tokens=True )[0] return transcription # 使用示例 if __name__ == "__main__": result = transcribe_audio("你的音频文件.mp3") print(f"识别结果: {result}")

5. 批量音频处理实战

5.1 批量处理多个音频文件

在实际工作中，我们经常需要处理大量的音频文件。下面是一个批量处理的示例：

import os from pathlib import Path def batch_transcribe(audio_folder, output_file="results.txt"): """批量处理文件夹中的所有音频文件""" # 支持的音频格式 audio_extensions = {'.wav', '.mp3', '.m4a', '.ogg'} # 获取所有音频文件 audio_files = [ f for f in Path(audio_folder).iterdir() if f.suffix.lower() in audio_extensions ] print(f"找到 {len(audio_files)} 个音频文件") # 初始化模型（只初始化一次） model_id = "Qwen/Qwen3-ASR-1.7B" model = AutoModelForSpeechSeq2Seq.from_pretrained( model_id, torch_dtype=torch.float16, device_map="auto" ) processor = AutoProcessor.from_pretrained(model_id) # 处理每个文件 results = [] for i, audio_file in enumerate(audio_files, 1): print(f"处理第 {i}/{len(audio_files)} 个文件: {audio_file.name}") try: transcription = transcribe_single_file( str(audio_file), model, processor ) results.append(f"{audio_file.name}: {transcription}") except Exception as e: results.append(f"{audio_file.name}: 处理失败 - {str(e)}") # 保存结果 with open(output_file, 'w', encoding='utf-8') as f: for result in results: f.write(result + '\n') print(f"处理完成！结果已保存到 {output_file}") def transcribe_single_file(audio_path, model, processor): """处理单个音频文件""" dataset = load_dataset("audio", data_files={"audio": audio_path}) sample = dataset["audio"][0] inputs = processor( sample["array"], sampling_rate=sample["sampling_rate"], return_tensors="pt" ) with torch.no_grad(): outputs = model.generate(**inputs.to(model.device)) return processor.batch_decode(outputs, skip_special_tokens=True)[0]

5.2 添加进度显示和错误处理

为了让批量处理更加友好，我们可以添加进度显示和健壮的错误处理：

from tqdm import tqdm import warnings def robust_batch_transcribe(audio_folder, output_file="results.txt"): """带进度显示和错误处理的批量转写""" audio_files = [f for f in Path(audio_folder).iterdir() if f.suffix.lower() in {'.wav', '.mp3', '.m4a', '.ogg'}] # 初始化模型 model = AutoModelForSpeechSeq2Seq.from_pretrained( "Qwen/Qwen3-ASR-1.7B", torch_dtype=torch.float16, device_map="auto" ) processor = AutoProcessor.from_pretrained("Qwen/Qwen3-ASR-1.7B") results = [] with tqdm(total=len(audio_files), desc="处理进度") as pbar: for audio_file in audio_files: try: # 抑制不必要的警告 with warnings.catch_warnings(): warnings.simplefilter("ignore") transcription = transcribe_single_file( str(audio_file), model, processor ) results.append(f"{audio_file.name}: {transcription}") except Exception as e: results.append(f"{audio_file.name}: 错误 - {str(e)}") finally: pbar.update(1) pbar.set_postfix({"当前文件": audio_file.name[:20] + '...'}) # 保存结果 with open(output_file, 'w', encoding='utf-8') as f: f.write("\n".join(results)) return results

6. 实用技巧与优化建议

6.1 内存和性能优化

处理大量音频时，内存管理很重要：

def memory_efficient_transcribe(audio_paths, batch_size=2): """内存高效的批量处理""" model = AutoModelForSpeechSeq2Seq.from_pretrained( "Qwen/Qwen3-ASR-1.7B", torch_dtype=torch.float16, device_map="auto", low_cpu_mem_usage=True ) processor = AutoProcessor.from_pretrained("Qwen/Qwen3-ASR-1.7B") results = [] for i in range(0, len(audio_paths), batch_size): batch_paths = audio_paths[i:i+batch_size] batch_results = process_batch(batch_paths, model, processor) results.extend(batch_results) # 清理内存 torch.cuda.empty_cache() return results def process_batch(audio_paths, model, processor): """处理一个批次的音频文件""" batch_results = [] for audio_path in audio_paths: try: dataset = load_dataset("audio", data_files={"audio": audio_path}) sample = dataset["audio"][0] inputs = processor( sample["array"], sampling_rate=sample["sampling_rate"], return_tensors="pt" ) with torch.no_grad(): outputs = model.generate(**inputs.to(model.device)) transcription = processor.batch_decode( outputs, skip_special_tokens=True )[0] batch_results.append(f"{Path(audio_path).name}: {transcription}") except Exception as e: batch_results.append(f"{Path(audio_path).name}: 错误 - {str(e)}") return batch_results

6.2 处理不同音频格式

不同的音频格式可能需要不同的处理方式：

from pydub import AudioSegment import tempfile def convert_and_transcribe(audio_path, model, processor): """转换音频格式并转写""" # 如果是不是常见格式，先进行转换 if not audio_path.lower().endswith(('.wav', '.mp3', '.m4a', '.ogg')): audio = AudioSegment.from_file(audio_path) # 转换为WAV格式 with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as tmp: audio.export(tmp.name, format='wav') audio_path = tmp.name # 进行转写 dataset = load_dataset("audio", data_files={"audio": audio_path}) sample = dataset["audio"][0] inputs = processor( sample["array"], sampling_rate=sample["sampling_rate"], return_tensors="pt" ) with torch.no_grad(): outputs = model.generate(**inputs.to(model.device)) return processor.batch_decode(outputs, skip_special_tokens=True)[0]