# Qwen3-ASR-0.6B代码实例：Python调用API实现批量音频转写与语言自动判别 > 本文介绍如何通过Python代码调用Qwen3-ASR-0.6B模型的API接口，实现批量音频文件的自动转写和语言识别功能。 ## 1. 环境准备与安装依赖 在开始编写代码之前，我们需要先准备好Python环境并安装必要的依赖包。 ### 1.1 安装核心依赖 打开终端或命令行，执行以下命令安装所需库： ```bash pip install requests soundfile librosa numpy ``` 这些库的作用分别是： - `requests`：用于发送HTTP请求到API接口 - `soundfile`和`librosa`：用于处理音频文件 - `numpy`：数值计算支持 ### 1.2 检查音频处理环境 确保你的系统已经安装了ffmpeg，这是处理多种音频格式的基础： ```bash # 在Ubuntu/Debian系统上 sudo apt-get install ffmpeg # 在CentOS/RHEL系统上 sudo yum install ffmpeg # 通过conda安装 conda install -c conda-forge ffmpeg ``` ## 2. 基础API调用方法 让我们先从最简单的单个音频文件识别开始，了解API的基本调用方式。 ### 2.1 单个音频文件识别代码 ```python import requests import json import base64 def transcribe_audio_single(audio_file_path, api_url, language="auto"): """ 单个音频文件转写函数 参数: audio_file_path: 音频文件路径 api_url: API接口地址 language: 语言代码，默认为"auto"自动检测 """ try: # 读取音频文件并编码为base64 with open(audio_file_path, "rb") as audio_file: audio_data = base64.b64encode(audio_file.read()).decode('utf-8') # 构建请求数据 payload = { "audio": audio_data, "language": language } # 发送POST请求 headers = {'Content-Type': 'application/json'} response = requests.post(api_url, json=payload, headers=headers) # 检查响应状态 if response.status_code == 200: result = response.json() return { "success": True, "text": result.get("text", ""), "language": result.get("language", ""), "confidence": result.get("confidence", 0) } else: return { "success": False, "error": f"API请求失败，状态码: {response.status_code}", "details": response.text } except Exception as e: return { "success": False, "error": f"处理过程中发生错误: {str(e)}" } # 使用示例 if __name__ == "__main__": # API地址（替换为你的实际地址） api_url = "https://gpu-your-instance-id-7860.web.gpu.csdn.net/transcribe" # 音频文件路径 audio_file = "sample_audio.wav" # 调用转写函数 result = transcribe_audio_single(audio_file, api_url) if result["success"]: print(f"识别结果: {result['text']}") print(f"检测语言: {result['language']}") print(f"置信度: {result['confidence']}") else: print(f"识别失败: {result['error']}") ``` ### 2.2 支持的语言代码参考 Qwen3-ASR-0.6B支持多种语言识别，以下是一些常用语言代码： ```python # 常用语言代码示例 LANGUAGE_CODES = { "auto": "自动检测", "zh": "中文普通话", "zh-yue": "粤语", "zh-sichuan": "四川话", "en": "英语", "en-US": "美式英语", "en-GB": "英式英语", "ja": "日语", "ko": "韩语", "fr": "法语", "de": "德语", "es": "西班牙语", "ru": "俄语", "ar": "阿拉伯语", # 更多语言代码... } # 使用指定语言进行识别 result = transcribe_audio_single("french_audio.mp3", api_url, language="fr") ``` ## 3. 批量音频处理实现 在实际应用中，我们经常需要处理大量的音频文件，下面实现一个批量处理的功能。 ### 3.1 批量处理类实现 ```python import os import glob import time from concurrent.futures import ThreadPoolExecutor, as_completed class BatchAudioTranscriber: def __init__(self, api_url, output_dir="results", max_workers=3): self.api_url = api_url self.output_dir = output_dir self.max_workers = max_workers # 创建输出目录 os.makedirs(output_dir, exist_ok=True) def process_single_file(self, audio_path, language="auto"): """处理单个音频文件""" result = transcribe_audio_single(audio_path, self.api_url, language) # 添加文件信息 result["filename"] = os.path.basename(audio_path) result["process_time"] = time.strftime("%Y-%m-%d %H:%M:%S") return result def process_batch(self, audio_files, language="auto"): """批量处理音频文件""" results = [] # 使用线程池并行处理 with ThreadPoolExecutor(max_workers=self.max_workers) as executor: # 提交所有任务 future_to_file = { executor.submit(self.process_single_file, file, language): file for file in audio_files } # 收集结果 for future in as_completed(future_to_file): audio_file = future_to_file[future] try: result = future.result() results.append(result) # 实时输出进度 print(f"处理完成: {audio_file}") # 保存单个结果 self.save_single_result(result) except Exception as e: print(f"处理失败 {audio_file}: {str(e)}") results.append({ "filename": audio_file, "success": False, "error": str(e) }) # 保存批量结果汇总 self.save_batch_summary(results) return results def save_single_result(self, result): """保存单个文件的结果""" if result["success"]: filename = result["filename"] base_name = os.path.splitext(filename)[0] output_file = os.path.join(self.output_dir, f"{base_name}_result.txt") with open(output_file, "w", encoding="utf-8") as f: f.write(f"文件名: {filename}\n") f.write(f"处理时间: {result['process_time']}\n") f.write(f"检测语言: {result.get('language', '未知')}\n") f.write(f"置信度: {result.get('confidence', 0)}\n") f.write("-" * 50 + "\n") f.write("转写文本:\n") f.write(result.get("text", "")) def save_batch_summary(self, results): """保存批量处理汇总""" summary_file = os.path.join(self.output_dir, "batch_summary.csv") with open(summary_file, "w", encoding="utf-8") as f: f.write("文件名,处理状态,检测语言,置信度,文本长度,处理时间\n") for result in results: filename = result["filename"] status = "成功" if result["success"] else "失败" language = result.get("language", "") confidence = result.get("confidence", 0) text_length = len(result.get("text", "")) process_time = result.get("process_time", "") f.write(f'"{filename}",{status},{language},{confidence},{text_length},{process_time}\n') # 使用示例 if __name__ == "__main__": # 初始化批量处理器 transcriber = BatchAudioTranscriber( api_url="https://gpu-your-instance-id-7860.web.gpu.csdn.net/transcribe", output_dir="./transcription_results" ) # 查找所有音频文件 audio_files = glob.glob("./audio_files/*.mp3") + glob.glob("./audio_files/*.wav") print(f"找到 {len(audio_files)} 个音频文件") # 开始批量处理 results = transcriber.process_batch(audio_files, language="auto") # 统计结果 success_count = sum(1 for r in results if r["success"]) print(f"处理完成: {success_count}/{len(audio_files)} 成功") ``` ### 3.2 支持多种音频格式 ```python def find_audio_files(directory, extensions=None): """ 查找目录中的所有音频文件 参数: directory: 要搜索的目录 extensions: 音频文件扩展名列表，默认为常见音频格式 """ if extensions is None: extensions = ['.wav', '.mp3', '.flac', '.ogg', '.m4a', '.aac'] audio_files = [] for ext in extensions: audio_files.extend(glob.glob(os.path.join(directory, f"*{ext}"))) audio_files.extend(glob.glob(os.path.join(directory, f"*{ext.upper()}"))) return audio_files # 使用示例 audio_directory = "/path/to/your/audio/files" all_audio_files = find_audio_files(audio_directory) print(f"找到 {len(all_audio_files)} 个音频文件") ``` ## 4. 高级功能与实用技巧 除了基本的转写功能，我们还可以实现一些高级特性来提升使用体验。 ### 4.1 实时进度显示与断点续传 ```python class AdvancedTranscriber(BatchAudioTranscriber): def __init__(self, api_url, output_dir="results", max_workers=3): super().__init__(api_url, output_dir, max_workers) self.processed_files = set() self.load_processed_files() def load_processed_files(self): """加载已处理的文件记录""" record_file = os.path.join(self.output_dir, ".processed_files") if os.path.exists(record_file): with open(record_file, "r") as f: self.processed_files = set(line.strip() for line in f) def save_processed_file(self, filename): """保存已处理文件记录""" self.processed_files.add(filename) record_file = os.path.join(self.output_dir, ".processed_files") with open(record_file, "a") as f: f.write(filename + "\n") def filter_processed_files(self, audio_files): """过滤掉已经处理过的文件""" return [f for f in audio_files if f not in self.processed_files] def process_batch_with_resume(self, audio_files, language="auto"): """支持断点续传的批量处理""" # 过滤已处理文件 files_to_process = self.filter_processed_files(audio_files) print(f"跳过 {len(audio_files) - len(files_to_process)} 个已处理文件") print(f"待处理: {len(files_to_process)} 个文件") # 处理剩余文件 results = [] for i, audio_file in enumerate(files_to_process): try: print(f"处理中 [{i+1}/{len(files_to_process)}]: {os.path.basename(audio_file)}") result = self.process_single_file(audio_file, language) results.append(result) # 记录已处理文件 self.save_processed_file(audio_file) # 添加延迟，避免请求过于频繁 time.sleep(0.5) except Exception as e: print(f"处理失败 {audio_file}: {str(e)}") results.append({ "filename": audio_file, "success": False, "error": str(e) }) return results ``` ### 4.2 语言检测统计与分析 ```python def analyze_language_distribution(results): """分析语言分布情况""" language_count = {} total_success = 0 for result in results: if result["success"]: language = result.get("language", "unknown") language_count[language] = language_count.get(language, 0) + 1 total_success += 1 print(f"\n语言分布分析 (共{total_success}个成功样本):") print("-" * 40) for language, count in sorted(language_count.items(), key=lambda x: x[1], reverse=True): percentage = (count / total_success) * 100 print(f"{language:15}: {count:3} 个 ({percentage:5.1f}%)") return language_count # 在批量处理后调用分析 language_stats = analyze_language_distribution(results) ``` ### 4.3 音频预处理功能 ```python import soundfile as sf import librosa import numpy as np def preprocess_audio(input_path, output_path, target_sr=16000): """ 音频预处理函数 将音频转换为模型适合的格式 """ try: # 读取音频文件 audio, sr = librosa.load(input_path, sr=None) # 重采样到目标采样率（如果需要） if sr != target_sr: audio = librosa.resample(audio, orig_sr=sr, target_sr=target_sr) # 标准化音频音量 audio = audio / np.max(np.abs(audio)) * 0.9 # 保存处理后的音频 sf.write(output_path, audio, target_sr) return True except Exception as e: print(f"音频预处理失败 {input_path}: {str(e)}") return False def process_with_preprocessing(audio_file, api_url, temp_dir="temp_audio"): """带预处理的转写函数""" os.makedirs(temp_dir, exist_ok=True) # 生成临时文件路径 temp_file = os.path.join(temp_dir, f"preprocessed_{os.path.basename(audio_file)}") # 预处理音频 if preprocess_audio(audio_file, temp_file): # 使用预处理后的音频进行转写 result = transcribe_audio_single(temp_file, api_url) # 清理临时文件 try: os.remove(temp_file) except: pass return result else: return { "success": False, "error": "音频预处理失败", "filename": audio_file } ``` ## 5. 完整示例与实战应用 下面是一个完整的实战示例，展示如何在实际项目中使用这些代码。 ### 5.1 完整项目示例 ```python import argparse import json from datetime import datetime def main(): """主函数：命令行接口""" parser = argparse.ArgumentParser(description='批量音频转写工具') parser.add_argument('input_dir', help='输入音频目录') parser.add_argument('--output', '-o', default='results', help='输出目录') parser.add_argument('--language', '-l', default='auto', help='语言代码') parser.add_argument('--workers', '-w', type=int, default=3, help='并行工作数') parser.add_argument('--api', '-a', required=True, help='API地址') args = parser.parse_args() # 创建转写器实例 transcriber = AdvancedTranscriber( api_url=args.api, output_dir=args.output, max_workers=args.workers ) # 查找音频文件 audio_files = find_audio_files(args.input_dir) print(f"在目录 {args.input_dir} 中找到 {len(audio_files)} 个音频文件") if not audio_files: print("未找到音频文件，请检查目录路径") return # 开始处理 start_time = datetime.now() results = transcriber.process_batch_with_resume(audio_files, args.language) # 分析结果 success_count = sum(1 for r in results if r["success"]) total_time = (datetime.now() - start_time).total_seconds() print(f"\n处理完成!") print(f"成功: {success_count}/{len(audio_files)}") print(f"总耗时: {total_time:.1f} 秒") print(f"平均每个文件: {total_time/len(audio_files):.1f} 秒") # 语言分布分析 analyze_language_distribution(results) # 保存详细结果 summary_file = os.path.join(args.output, f"processing_summary_{start_time.strftime('%Y%m%d_%H%M%S')}.json") with open(summary_file, 'w', encoding='utf-8') as f: json.dump({ "process_time": start_time.strftime("%Y-%m-%d %H:%M:%S"), "total_files": len(audio_files), "success_count": success_count, "total_duration": total_time, "results": results }, f, ensure_ascii=False, indent=2) print(f"\n详细结果已保存至: {summary_file}") if __name__ == "__main__": main() ``` ### 5.2 使用方式 将上述代码保存为 `audio_transcriber.py`，然后通过命令行使用： ```bash # 基本用法 python audio_transcriber.py /path/to/audio/files --api https://your-api-url # 指定语言和输出目录 python audio_transcriber.py /path/to/audio/files \ --api https://your-api-url \ --language zh \ --output ./my_results \ --workers 5 # 只处理特定格式 # 先确保目录中只有你想要的音频格式，或者修改代码中的扩展名列表 ``` ### 5.3 错误处理与日志记录 ```python import logging # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('audio_transcription.log'), logging.StreamHandler() ] ) def robust_transcribe(audio_file, api_url, retries=3): """带重试机制的转写函数""" for attempt in range(retries): try: result = transcribe_audio_single(audio_file, api_url) if result["success"]: return result else: logging.warning(f"尝试 {attempt+1} 失败: {result['error']}") except Exception as e: logging.error(f"尝试 {attempt+1} 异常: {str(e)}") # 重试前等待 if attempt < retries - 1: time.sleep(2 ** attempt) # 指数退避 return { "success": False, "error": f"所有 {retries} 次尝试均失败", "filename": audio_file } ``` ## 6. 总结与建议 通过本文的代码实例，你应该已经掌握了如何使用Python调用Qwen3-ASR-0.6B的API接口实现批量音频转写和语言自动判别功能。 ### 6.1 核心要点回顾 1. **基础API调用**：学会了如何通过HTTP请求与语音识别API交互 2. **批量处理能力**：实现了支持多线程的批量音频处理功能 3. **高级特性**：添加了断点续传、进度显示、语言分析等实用功能 4. **错误处理**：完善的异常处理和重试机制确保稳定性 ### 6.2 实用建议 1. **调整并行度**：根据API服务器的性能调整`max_workers`参数，避免请求过于频繁 2. **音频预处理**：对于质量较差的音频文件，先进行降噪和标准化处理 3. **结果验证**：对于重要应用，建议对识别结果进行人工抽样验证 4. **性能监控**：记录处理时间和成功率，便于优化和排查问题 ### 6.3 扩展思路 你可以基于这个基础框架进一步扩展： - 添加Web界面实现可视化操作 - 集成到现有的工作流系统中 - 开发实时音频流识别功能 - 添加更多音频分析功能（说话人分离、情感分析等） 现在你已经拥有了一个功能完整的音频转写工具，可以根据实际需求进行调整和扩展，满足各种语音识别应用场景。 --- > **获取更多AI镜像** > > 想探索更多AI镜像和应用场景？访问 [CSDN星图镜像广场](https://ai.csdn.net/?utm_source=mirror_blog_end)，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。