# Qwen3-ASR-1.7B实战教程：FFmpeg+Python批量切片+ASR+文本后处理流水线 ## 1. 项目概述与目标 今天我要分享一个完整的语音识别实战项目，教你如何用Qwen3-ASR-1.7B模型搭建一个高效的音频处理流水线。这个方案特别适合需要处理大量音频文件的场景，比如会议录音转写、课程录制整理、播客内容分析等。 **学完本教程你能掌握**： - 如何用FFmpeg批量切割长音频文件 - 如何用Python自动化调用Qwen3-ASR模型 - 如何对识别结果进行智能后处理 - 如何构建完整的端到端处理流程 **前置知识要求**： - 基本的Python编程经验 - 会在命令行中运行简单命令 - 不需要深度学习背景，我会一步步带你操作 ## 2. 环境准备与工具安装 ### 2.1 核心工具安装 首先确保你的系统中安装了以下工具： ```bash # 安装FFmpeg（Ubuntu/Debian系统） sudo apt update sudo apt install ffmpeg # 安装Python依赖包 pip install requests tqdm pydub ``` 如果你的系统是Windows，可以从FFmpeg官网下载预编译版本，然后添加到系统路径中。 ### 2.2 Qwen3-ASR模型访问 确保你的Qwen3-ASR-1.7B服务已经正常运行，可以通过浏览器访问Web界面。记下你的服务地址，格式通常是： ``` https://gpu-你的实例ID-7860.web.gpu.csdn.net/ ``` ## 3. 音频预处理：FFmpeg批量切片 长音频文件直接识别往往效果不好，而且容易出错。我们先把它切成小段，每段30秒左右，这样识别准确率更高。 ### 3.1 单文件切片方法 ```bash # 将长音频切成30秒一段的小文件 ffmpeg -i input.mp3 -f segment -segment_time 30 -c copy output_%03d.mp3 ``` 这个命令会把input.mp3文件按30秒一段进行切割，生成output_001.mp3、output_002.mp3等文件。 ### 3.2 批量处理脚本 创建`split_audio.py`文件： ```python import os import subprocess from pathlib import Path def batch_split_audio(input_folder, output_folder, segment_time=30): """ 批量切割音频文件 :param input_folder: 输入文件夹路径 :param output_folder: 输出文件夹路径 :param segment_time: 每段时长（秒） """ # 创建输出文件夹 Path(output_folder).mkdir(exist_ok=True) # 支持的文件格式 audio_extensions = ['.mp3', '.wav', '.flac', '.m4a'] for file in Path(input_folder).iterdir(): if file.suffix.lower() in audio_extensions: output_pattern = f"{output_folder}/{file.stem}_%03d{file.suffix}" cmd = [ 'ffmpeg', '-i', str(file), '-f', 'segment', '-segment_time', str(segment_time), '-c', 'copy', output_pattern ] try: subprocess.run(cmd, check=True, capture_output=True) print(f"成功切割: {file.name}") except subprocess.CalledProcessError as e: print(f"切割失败 {file.name}: {e}") # 使用示例 if __name__ == "__main__": batch_split_audio("raw_audio", "splitted_audio", segment_time=30) ``` ## 4. 调用Qwen3-ASR进行语音识别 现在我们来编写调用Qwen3-ASR模型的Python代码。 ### 4.1 基础识别函数 创建`asr_client.py`文件： ```python import requests import json import os from pathlib import Path from tqdm import tqdm class QwenASRClient: def __init__(self, base_url): self.base_url = base_url.rstrip('/') self.api_url = f"{self.base_url}/run/predict" def transcribe_audio(self, audio_path, language="auto"): """ 识别单个音频文件 :param audio_path: 音频文件路径 :param language: 语言类型，默认auto自动检测 :return: 识别结果文本 """ try: with open(audio_path, 'rb') as f: files = {'audio': (os.path.basename(audio_path), f)} data = {'language': language} response = requests.post(self.api_url, files=files, data=data) response.raise_for_status() result = response.json() return result['data'][0] # 返回识别文本 except Exception as e: print(f"识别失败 {audio_path}: {e}") return None def batch_transcribe(self, audio_folder, output_file="transcriptions.txt", language="auto"): """ 批量识别音频文件 """ audio_files = list(Path(audio_folder).glob("*.mp3")) + \ list(Path(audio_folder).glob("*.wav")) + \ list(Path(audio_folder).glob("*.flac")) results = [] for audio_file in tqdm(audio_files, desc="识别进度"): text = self.transcribe_audio(audio_file, language) if text: results.append({ 'file': audio_file.name, 'text': text }) # 保存结果 with open(output_file, 'w', encoding='utf-8') as f: for result in results: f.write(f"{result['file']}|{result['text']}\n") return results # 使用示例 if __name__ == "__main__": # 替换为你的实际地址 client = QwenASRClient("https://gpu-你的实例ID-7860.web.gpu.csdn.net") # 批量识别 results = client.batch_transcribe("splitted_audio", "raw_transcriptions.txt") ``` ## 5. 文本后处理与优化 原始识别结果可能存在一些格式问题，我们需要进行后处理。 ### 5.1 智能后处理脚本 创建`text_processor.py`文件： ```python import re from datetime import datetime class TextProcessor: @staticmethod def clean_text(text): """ 清理和优化识别文本 """ if not text: return "" # 移除多余空格 text = re.sub(r'\s+', ' ', text).strip() # 处理标点符号 text = re.sub(r'([,.!?])([^ ])', r'\1 \2', text) # 处理数字和英文混排 text = re.sub(r'([0-9])([a-zA-Z])', r'\1 \2', text) text = re.sub(r'([a-zA-Z])([0-9])', r'\1 \2', text) return text @staticmethod def merge_transcriptions(transcriptions, max_gap=5): """ 合并连续的转录片段 :param transcriptions: 转录结果列表 :param max_gap: 最大间隔时间（秒） :return: 合并后的文本 """ if not transcriptions: return "" merged_text = [] current_speaker = None current_text = [] for trans in transcriptions: # 这里可以根据需要添加说话人识别逻辑 text = trans['text'] if text: # 只处理非空文本 current_text.append(text) # 合并所有文本片段 full_text = ' '.join(current_text) # 进行最终清理 full_text = TextProcessor.clean_text(full_text) return full_text def process_transcription_file(input_file, output_file): """ 处理转录结果文件 """ transcripts = [] with open(input_file, 'r', encoding='utf-8') as f: for line in f: parts = line.strip().split('|', 1) if len(parts) == 2: file_name, text = parts # 从文件名中提取时间信息（如果有） transcripts.append({ 'file': file_name, 'text': text }) # 按文件名排序（确保时间顺序） transcripts.sort(key=lambda x: x['file']) # 合并文本 merged_text = TextProcessor.merge_transcriptions(transcripts) # 保存结果 with open(output_file, 'w', encoding='utf-8') as f: f.write(f"处理时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") f.write(f"处理文件数: {len(transcripts)}\n") f.write("-" * 50 + "\n") f.write(merged_text) return merged_text # 使用示例 if __name__ == "__main__": process_transcription_file("raw_transcriptions.txt", "final_transcription.txt") ``` ## 6. 完整流水线整合 现在我们把所有步骤整合成一个完整的流水线脚本。 创建`main_pipeline.py`文件： ```python import os from pathlib import Path from split_audio import batch_split_audio from asr_client import QwenASRClient from text_processor import process_transcription_file def main(): # 配置参数 input_audio = "long_audio.mp3" # 你的长音频文件 temp_folder = "temp_splits" raw_transcript = "raw_transcript.txt" final_output = "final_output.txt" asr_url = "https://gpu-你的实例ID-7860.web.gpu.csdn.net" # 替换为你的地址 print("开始音频处理流水线...") # 步骤1：创建临时文件夹 Path(temp_folder).mkdir(exist_ok=True) # 步骤2：切割音频（如果输入是单个长文件） print("切割音频文件...") batch_split_audio(".", temp_folder, segment_time=30) # 步骤3：语音识别 print("开始语音识别...") client = QwenASRClient(asr_url) client.batch_transcribe(temp_folder, raw_transcript) # 步骤4：文本后处理 print("进行文本后处理...") final_text = process_transcription_file(raw_transcript, final_output) # 步骤5：清理临时文件（可选） # import shutil # shutil.rmtree(temp_folder) print(f"处理完成！结果保存在: {final_output}") print(f"识别总字数: {len(final_text)}") if __name__ == "__main__": main() ``` ## 7. 进阶功能与优化建议 ### 7.1 错误重试机制 在`asr_client.py`中添加重试逻辑： ```python import time from tenacity import retry, stop_after_attempt, wait_exponential class QwenASRClient: # ... 其他代码不变 ... @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) def transcribe_audio_with_retry(self, audio_path, language="auto"): """ 带重试机制的语音识别 """ return self.transcribe_audio(audio_path, language) ``` ### 7.2 进度保存与恢复 对于大量文件处理，可以添加进度保存功能： ```python import json def save_progress(progress_data, progress_file="progress.json"): with open(progress_file, 'w', encoding='utf-8') as f: json.dump(progress_data, f, ensure_ascii=False) def load_progress(progress_file="progress.json"): if os.path.exists(progress_file): with open(progress_file, 'r', encoding='utf-8') as f: return json.load(f) return {} ``` ### 7.3 批量处理文件夹 如果你有多个音频文件需要处理： ```python def process_multiple_audios(input_folder, output_folder): """ 处理整个文件夹的音频文件 """ audio_files = list(Path(input_folder).glob("*.mp3")) + \ list(Path(input_folder).glob("*.wav")) for audio_file in audio_files: output_file = Path(output_folder) / f"{audio_file.stem}_transcribed.txt" # 为每个文件创建临时处理文件夹 temp_folder = Path("temp") / audio_file.stem temp_folder.mkdir(parents=True, exist_ok=True) # 执行完整的处理流程 batch_split_audio(str(audio_file.parent), str(temp_folder), segment_time=30) # ... 其他处理步骤 ... ``` ## 8. 总结与建议 通过这个完整的流水线，你现在可以： 1. **高效处理长音频**：用FFmpeg自动切割成适合识别的小段 2. **批量语音识别**：用Python自动化调用Qwen3-ASR模型 3. **智能文本处理**：对识别结果进行清理和优化 4. **完整流程管理**：从原始音频到最终文本的一站式处理 **实用建议**： - 对于会议录音，30秒的切片长度通常效果最好 - 如果识别准确率不高，尝试手动指定语言而不是用auto - 处理大量文件时，记得添加进度保存功能 - 可以根据需要调整后处理逻辑，比如添加专业术语校正 **常见问题解决**： - 如果识别速度慢，检查网络连接和GPU资源 - 如果出现识别错误，尝试减少切片长度（比如改为20秒） - 确保音频文件质量良好，背景噪音尽量小 这个方案特别适合需要处理大量音频内容的场景，比如媒体公司、教育机构、企业会议记录等。你可以根据自己的需求进一步定制和优化这个流水线。 --- > **获取更多AI镜像** > > 想探索更多AI镜像和应用场景？访问 [CSDN星图镜像广场](https://ai.csdn.net/?utm_source=mirror_blog_end)，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。