Qwen3-ASR-1.7B实战：如何用Python调用API实现批量转录-平芜编程栈

Qwen3-ASR-1.7B实战：如何用Python调用API实现批量转录

1. 项目概述与核心价值

你是不是经常需要处理大量的音频文件转录工作？无论是会议记录、访谈内容还是语音笔记，手动转录既耗时又容易出错。传统的语音识别工具要么需要联网上传（有隐私风险），要么识别准确率不高（特别是对方言和专业术语的处理）。

今天我要介绍的Qwen3-ASR-1.7B镜像，完美解决了这些问题。这是一个基于阿里巴巴17亿参数大模型的本地语音识别工具，支持20多种语言和方言，包括中文、英文、粤语等。最重要的是，它完全在本地运行，不需要网络连接，彻底保护你的语音隐私。

通过本文，你将学会如何用Python编写一个批量转录脚本，实现：

自动扫描指定文件夹内的音频文件（支持MP3、WAV、M4A等格式）
调用Qwen3-ASR的API接口进行高精度转录
将识别结果保存为文本文件，保持原文件名对应
处理大量文件时的进度管理和错误重试机制

我会提供完整的代码示例和实战技巧，即使你是Python新手也能轻松上手。让我们开始吧！

2. 环境准备与镜像部署

2.1 获取并启动Qwen3-ASR镜像

首先，你需要在CSDN星图平台找到Qwen3-ASR-1.7B镜像。在镜像广场搜索"Qwen3-ASR"，选择1.7B版本，点击"立即部署"。

推荐配置：

GPU实例：至少8GB显存（T4或同等级别）
系统：Ubuntu 20.04或更高版本
端口：确保8000端口对外开放

部署完成后，通过SSH连接到你的实例，或者使用平台提供的Web终端。

2.2 验证服务状态

镜像内置了FastAPI服务，默认监听8000端口。首先检查服务是否正常启动：

# 查看服务进程 ps aux | grep uvicorn # 检查端口监听 netstat -tlnp | grep 8000 # 测试API健康状态 curl http://localhost:8000/health

如果一切正常，你应该看到类似这样的响应：

{"status":"healthy","model":"Qwen3-ASR-1.7B"}

2.3 安装必要的Python库

在你的本地开发环境或另一个服务器上，安装调用API所需的库：

pip install requests tqdm python-dotenv

这些库的作用分别是：

requests：用于发送HTTP请求到API
tqdm：显示进度条，方便监控批量处理进度
python-dotenv：管理环境变量，保护API密钥等敏感信息

3. Python API调用核心代码

3.1 基础调用函数

让我们从最简单的单文件转录开始。创建一个名为asr_client.py的文件：

import requests import json import os from pathlib import Path class QwenASRClient: def __init__(self, base_url="http://localhost:8000"): self.base_url = base_url self.transcribe_url = f"{base_url}/transcribe" def transcribe_audio(self, audio_path): """转录单个音频文件""" try: with open(audio_path, 'rb') as audio_file: files = {'audio_file': audio_file} response = requests.post(self.transcribe_url, files=files) if response.status_code == 200: result = response.json() return result['text'] else: print(f"错误: {response.status_code} - {response.text}") return None except Exception as e: print(f"转录失败: {str(e)}") return None # 使用示例 if __name__ == "__main__": client = QwenASRClient() result = client.transcribe_audio("test_audio.wav") if result: print("转录结果:", result)

这个基础版本已经可以处理单个文件了。但我们需要更强大的批量处理能力。

3.2 批量处理增强版

现在让我们扩展这个类，添加批量处理功能：

import requests import json import os from pathlib import Path from tqdm import tqdm import time class BatchASRClient: def __init__(self, base_url="http://localhost:8000", max_retries=3): self.base_url = base_url self.transcribe_url = f"{base_url}/transcribe" self.max_retries = max_retries self.supported_formats = ['.wav', '.mp3', '.m4a', '.flac', '.ogg'] def is_supported_format(self, file_path): """检查文件格式是否支持""" return Path(file_path).suffix.lower() in self.supported_formats def find_audio_files(self, directory): """查找目录中的所有音频文件""" audio_files = [] directory_path = Path(directory) for format in self.supported_formats: audio_files.extend(directory_path.glob(f"**/*{format}")) return [str(file) for file in audio_files] def transcribe_with_retry(self, audio_path, retry_delay=2): """带重试机制的转录函数""" for attempt in range(self.max_retries): try: with open(audio_path, 'rb') as audio_file: files = {'audio_file': audio_file} response = requests.post(self.transcribe_url, files=files, timeout=30) if response.status_code == 200: return response.json()['text'] else: print(f"尝试 {attempt + 1} 失败: {response.status_code}") except requests.exceptions.Timeout: print(f"尝试 {attempt + 1}: 超时") except Exception as e: print(f"尝试 {attempt + 1} 错误: {str(e)}") if attempt < self.max_retries - 1: time.sleep(retry_delay * (attempt + 1)) return None def batch_transcribe(self, input_dir, output_dir=None): """批量转录目录中的所有音频文件""" if output_dir is None: output_dir = input_dir # 创建输出目录（如果不存在） Path(output_dir).mkdir(parents=True, exist_ok=True) # 查找所有音频文件 audio_files = self.find_audio_files(input_dir) print(f"找到 {len(audio_files)} 个音频文件") results = [] for audio_file in tqdm(audio_files, desc="转录进度"): text = self.transcribe_with_retry(audio_file) if text: # 保存结果 output_file = Path(output_dir) / f"{Path(audio_file).stem}.txt" with open(output_file, 'w', encoding='utf-8') as f: f.write(text) results.append({ 'audio_file': audio_file, 'text_file': str(output_file), 'success': True }) else: results.append({ 'audio_file': audio_file, 'success': False, 'error': '转录失败' }) return results # 使用示例 if __name__ == "__main__": client = BatchASRClient(base_url="http://你的服务器IP:8000") # 批量转录 results = client.batch_transcribe( input_dir="./audio_files", output_dir="./transcription_results" ) # 打印统计信息 success_count = sum(1 for r in results if r['success']) print(f"处理完成: {success_count}/{len(results)} 成功")

这个增强版本提供了完整的批量处理能力，包括文件查找、格式验证、重试机制和进度显示。

4. 高级功能与实战技巧

4.1 处理长音频文件

Qwen3-ASR-1.7B支持长音频转录，但对于特别长的文件（超过30分钟），我们建议先进行分割处理。这里提供一个简单的分割函数：

from pydub import AudioSegment import math def split_long_audio(audio_path, chunk_length_ms=1800000): """将长音频分割为30分钟 chunks""" audio = AudioSegment.from_file(audio_path) duration_ms = len(audio) chunks = math.ceil(duration_ms / chunk_length_ms) split_files = [] for i in range(chunks): start_ms = i * chunk_length_ms end_ms = min((i + 1) * chunk_length_ms, duration_ms) chunk = audio[start_ms:end_ms] chunk_file = f"{Path(audio_path).stem}_part{i+1}.wav" chunk.export(chunk_file, format="wav") split_files.append(chunk_file) return split_files

4.2 结果后处理与格式优化

转录结果可能需要一些后处理来提高可读性：

def postprocess_text(text): """对转录结果进行后处理""" # 基本的文本清理 lines = text.split('\n') processed_lines = [] for line in lines: line = line.strip() if not line: continue # 添加标点符号（如果缺失） if not line.endswith(('.', '!', '?', '。', '！', '？')): line += '.' # 首字母大写 if line and len(line) > 1: line = line[0].upper() + line[1:] processed_lines.append(line) return '\n'.join(processed_lines)

4.3 完整的实战脚本

结合所有功能，这是一个完整的实战脚本：

import argparse from pathlib import Path from batch_asr_client import BatchASRClient def main(): parser = argparse.ArgumentParser(description='批量语音转录工具') parser.add_argument('--input-dir', required=True, help='输入音频目录') parser.add_argument('--output-dir', help='输出文本目录') parser.add_argument('--api-url', default='http://localhost:8000', help='API地址') parser.add_argument('--max-retries', type=int, default=3, help='最大重试次数') args = parser.parse_args() # 初始化客户端 client = BatchASRClient( base_url=args.api_url, max_retries=args.max_retries ) # 执行批量转录 results = client.batch_transcribe( input_dir=args.input_dir, output_dir=args.output_dir ) # 生成报告 success_count = sum(1 for r in results if r['success']) print(f"\n处理完成!") print(f"成功: {success_count}/{len(results)}") if success_count < len(results): print("\n失败的文件:") for result in results: if not result['success']: print(f" - {result['audio_file']}") if __name__ == "__main__": main()

使用方式：

python batch_transcribe.py --input-dir ./meeting_recordings --output-dir ./transcripts --api-url http://192.168.1.100:8000

5. 性能优化与错误处理

5.1 并发处理提高效率

对于大量文件，串行处理速度太慢。我们可以使用多线程并发处理：

import concurrent.futures from tqdm import tqdm def concurrent_batch_transcribe(self, input_dir, output_dir=None, max_workers=4): """并发批量转录""" if output_dir is None: output_dir = input_dir Path(output_dir).mkdir(parents=True, exist_ok=True) audio_files = self.find_audio_files(input_dir) results = [] with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: # 提交所有任务 future_to_file = { executor.submit(self.transcribe_with_retry, file): file for file in audio_files } # 处理结果 for future in tqdm(concurrent.futures.as_completed(future_to_file), total=len(audio_files), desc="并发转录"): audio_file = future_to_file[future] try: text = future.result() if text: output_file = Path(output_dir) / f"{Path(audio_file).stem}.txt" with open(output_file, 'w', encoding='utf-8') as f: f.write(text) results.append({'audio_file': audio_file, 'success': True}) else: results.append({'audio_file': audio_file, 'success': False}) except Exception as e: results.append({'audio_file': audio_file, 'success': False, 'error': str(e)}) return results