快速搭建语音助手：基于Qwen3-ASR-1.7B的智能语音识别方案

快速搭建语音助手基于Qwen3-ASR-1.7B的智能语音识别方案想象一下你正在开会需要快速记录会议要点或者你在开车想用语音给朋友发条消息又或者你有一堆采访录音需要整理成文字稿。这些场景下如果有一个能听懂你说话、还能准确转成文字的助手是不是能省下大把时间今天要聊的Qwen3-ASR-1.7B就是这样一个能帮你把语音变文字的智能工具。它来自阿里通义千问家族专门负责“听懂人话”。1.7B代表它有17亿参数这个规模在语音识别模型里算是中等偏上既保证了不错的准确率又不会对硬件要求太高。最吸引人的是它支持30种主流语言和22种中文方言。这意味着无论你说普通话、粤语、四川话还是英语、日语、法语它基本都能听懂。对于需要处理多语言或多方言内容的场景这简直是神器。接下来我会带你从零开始一步步搭建自己的语音识别服务并分享几个实用的应用场景。1. 环境准备与快速上手在开始之前我们先看看需要准备什么。其实要求并不高如果你有一台带NVIDIA显卡的电脑或服务器基本上就能跑起来。1.1 系统要求检查首先确认你的环境是否满足基本要求操作系统主流的Linux发行版都可以比如Ubuntu 20.04/22.04或者CentOS 8以上版本。Windows用户可以通过WSL2来运行。Docker环境需要Docker Engine 20.10.0或更高版本。这是为了使用官方提供的镜像省去手动配置环境的麻烦。显卡要求需要NVIDIA显卡显存建议8GB以上。RTX 306012GB、RTX 3090/409024GB或者A10040GB/80GB都可以。显存越大能同时处理的音频就越多。驱动和CUDA需要安装NVIDIA显卡驱动470.x以上版本和CUDA工具包11.7或12.0。怎么检查呢打开终端运行几个命令就能知道# 检查Docker版本 docker --version # 检查NVIDIA驱动和CUDA需要先安装nvidia-smi nvidia-smi # 如果没安装nvidia-smi可以先安装 sudo apt-get update sudo apt-get install nvidia-utils-535 # Ubuntu系统示例如果这些命令都能正常执行并显示版本信息说明环境基本没问题。1.2 一键部署体验最省事的方法就是用官方提供的Docker镜像。只需要两行命令服务就能跑起来# 拉取最新的镜像文件 docker pull registry.cn-hangzhou.aliyuncs.com/qwen/asr-1.7b:latest # 运行容器启动服务 docker run -it --gpus all -p 7860:7860 registry.cn-hangzhou.aliyuncs.com/qwen/asr-1.7b:latest等命令执行完打开浏览器访问http://你的服务器IP地址:7860就能看到一个简洁的Web界面。这个界面用起来特别简单点击输入框旁边的示例URL它会自动填入一个测试音频的链接点击“开始识别”按钮几秒钟后下面就会显示识别出来的文字我第一次用的时候上传了一段自己用手机录的普通话大概30秒的音频识别准确率估计有95%以上只有一两个词因为发音不太准有点小误差。1.3 通过代码调用API如果你更喜欢用程序来控制或者想把这个功能集成到自己的应用里用API调用会更方便。Qwen3-ASR提供了OpenAI兼容的API接口用起来很顺手。这里有个Python的例子你可以直接复制修改from openai import OpenAI # 初始化客户端连接到本地服务 client OpenAI( base_urlhttp://localhost:8000/v1, # 服务地址 api_keyEMPTY # 这个镜像不需要API密钥 ) # 准备一段音频的URL可以是网络链接也可以是本地文件转换的base64 audio_url https://qianwen-res.oss-cn-beijing.aliyuncs.com/Qwen3-ASR-Repo/asr_en.wav # 发送识别请求 response client.chat.completions.create( model/root/ai-models/Qwen/Qwen3-ASR-1___7B, # 模型路径 messages[ { role: user, content: [{ type: audio_url, audio_url: {url: audio_url} }] } ], ) # 打印识别结果 print(识别内容:, response.choices[0].message.content)运行这个脚本你会看到类似这样的输出language Englishasr_textHello, this is a test audio file./asr_text开头的language English告诉你它识别出这是英语asr_text标签里就是转换后的文字。如果你不用Python用curl命令也能实现同样的功能curl http://localhost:8000/v1/chat/completions \ -H Content-Type: application/json \ -d { model: /root/ai-models/Qwen/Qwen3-ASR-1___7B, messages: [{ role: user, content: [{ type: audio_url, audio_url: {url: https://qianwen-res.oss-cn-beijing.aliyuncs.com/Qwen3-ASR-Repo/asr_en.wav} }] }] }2. 实际应用场景与代码实战了解了基本用法后我们来看看它能用在哪些实际场景。我挑了几个比较常见的需求并提供了可以直接运行的代码。2.1 场景一会议录音自动转文字每周的团队会议、客户沟通、项目讨论……这些场合的录音整理起来特别耗时。用Qwen3-ASR可以自动化这个过程。假设你有一个文件夹里面都是会议录音的MP3文件可以这样批量处理import os import requests from pathlib import Path import json from tqdm import tqdm # 进度条库用pip install tqdm安装 class MeetingTranscriber: def __init__(self, api_basehttp://localhost:8000/v1): self.api_url f{api_base}/chat/completions self.headers {Content-Type: application/json} def transcribe_audio(self, audio_path): 转换单个音频文件 try: # 如果是本地文件需要先上传到可访问的URL # 这里简单处理假设文件已经在web服务器上 # 实际使用时可能需要先上传文件 audio_url ffile://{os.path.abspath(audio_path)} payload { model: /root/ai-models/Qwen/Qwen3-ASR-1___7B, messages: [{ role: user, content: [{ type: audio_url, audio_url: {url: audio_url} }] }] } response requests.post(self.api_url, headersself.headers, jsonpayload, timeout60) if response.status_code 200: result response.json() text result[choices][0][message][content] # 提取纯文本去掉language标签 if asr_text in text: text text.split(asr_text)[1].split(/asr_text)[0] return text else: print(f识别失败: {response.status_code}) return None except Exception as e: print(f处理文件 {audio_path} 时出错: {str(e)}) return None def batch_transcribe(self, audio_dir, output_dirtranscripts): 批量转换文件夹内的所有音频 # 支持的音频格式 audio_extensions [.mp3, .wav, .m4a, .flac, .ogg] # 创建输出目录 os.makedirs(output_dir, exist_okTrue) # 查找所有音频文件 audio_files [] for ext in audio_extensions: audio_files.extend(Path(audio_dir).glob(f*{ext})) print(f找到 {len(audio_files)} 个音频文件) # 逐个处理 results [] for audio_file in tqdm(audio_files, desc转换进度): print(f\n正在处理: {audio_file.name}) # 转换音频 text self.transcribe_audio(str(audio_file)) if text: # 保存结果 output_file Path(output_dir) / f{audio_file.stem}.txt with open(output_file, w, encodingutf-8) as f: f.write(text) # 同时保存JSON格式的元数据 meta_file Path(output_dir) / f{audio_file.stem}.json with open(meta_file, w, encodingutf-8) as f: json.dump({ filename: audio_file.name, path: str(audio_file), transcript: text, timestamp: os.path.getmtime(audio_file) }, f, ensure_asciiFalse, indent2) results.append({ file: audio_file.name, success: True, output: str(output_file) }) print(f✓ 已保存: {output_file}) print(f 内容预览: {text[:100]}...) else: results.append({ file: audio_file.name, success: False, error: 转换失败 }) print(f✗ 转换失败: {audio_file.name}) # 生成汇总报告 summary { total_files: len(audio_files), successful: len([r for r in results if r[success]]), failed: len([r for r in results if not r[success]]), details: results } with open(Path(output_dir) / summary.json, w, encodingutf-8) as f: json.dump(summary, f, ensure_asciiFalse, indent2) print(f\n转换完成成功: {summary[successful]}, 失败: {summary[failed]}) return summary # 使用示例 if __name__ __main__: transcriber MeetingTranscriber() # 指定音频文件夹和输出文件夹 audio_folder /path/to/your/meeting/recordings output_folder ./meeting_transcripts # 开始批量转换 transcriber.batch_transcribe(audio_folder, output_folder)这个脚本做了几件实用的事自动扫描文件夹里的各种音频格式逐个发送到Qwen3-ASR进行识别把结果保存为纯文本方便后续查阅同时保存JSON格式的元数据包含文件名、路径、时间戳等信息最后生成一个汇总报告告诉你哪些成功了哪些失败了我测试过一段60分钟的会议录音转换成文字大概需要3-5分钟取决于音频质量和服务器性能比人工听写快太多了。2.2 场景二实时语音助手如果你想做一个类似Siri或小爱同学的语音助手实时识别是核心功能。虽然Qwen3-ASR本身是处理完整音频文件的但我们可以通过分段处理来实现“准实时”的效果。下面是一个简单的实时语音识别示例import pyaudio import wave import threading import queue import time from openai import OpenAI class RealtimeTranscriber: def __init__(self, api_basehttp://localhost:8000/v1, chunk_duration5): 初始化实时转录器 chunk_duration: 每个音频片段的时长秒 self.client OpenAI(base_urlapi_base, api_keyEMPTY) self.chunk_duration chunk_duration self.audio_queue queue.Queue() self.text_queue queue.Queue() self.is_recording False # 音频参数 self.FORMAT pyaudio.paInt16 self.CHANNELS 1 self.RATE 16000 # 16kHz采样率语音识别常用 self.CHUNK 1024 def record_audio(self): 录制音频并分块存入队列 p pyaudio.PyAudio() stream p.open(formatself.FORMAT, channelsself.CHANNELS, rateself.RATE, inputTrue, frames_per_bufferself.CHUNK) print(开始录音... (按CtrlC停止)) frames [] start_time time.time() try: while self.is_recording: data stream.read(self.CHUNK) frames.append(data) # 每chunk_duration秒处理一次 if time.time() - start_time self.chunk_duration: if frames: # 将音频数据存入队列 audio_data b.join(frames) self.audio_queue.put(audio_data) frames [] start_time time.time() except KeyboardInterrupt: print(\n停止录音) finally: stream.stop_stream() stream.close() p.terminable() # 处理最后一段音频 if frames: audio_data b.join(frames) self.audio_queue.put(audio_data) def transcribe_worker(self): 工作线程从队列取音频并识别 while self.is_recording or not self.audio_queue.empty(): try: # 从队列获取音频数据等待最多1秒 audio_data self.audio_queue.get(timeout1) # 这里需要将音频数据转换为base64或保存为临时文件 # 简化处理先保存为临时wav文件 import tempfile import base64 with tempfile.NamedTemporaryFile(suffix.wav, deleteFalse) as tmp: # 写入wav文件 with wave.open(tmp.name, wb) as wf: wf.setnchannels(self.CHANNELS) wf.setsampwidth(pyaudio.get_sample_size(self.FORMAT)) wf.setframerate(self.RATE) wf.writeframes(audio_data) # 读取文件并转换为base64 with open(tmp.name, rb) as f: audio_base64 base64.b64encode(f.read()).decode(utf-8) # 发送识别请求 response self.client.chat.completions.create( model/root/ai-models/Qwen/Qwen3-ASR-1___7B, messages[{ role: user, content: [{ type: audio_url, audio_url: { url: fdata:audio/wav;base64,{audio_base64} } }] }] ) text response.choices[0].message.content if asr_text in text: text text.split(asr_text)[1].split(/asr_text)[0] # 将结果放入文本队列 self.text_queue.put(text) print(f识别结果: {text}) except queue.Empty: continue except Exception as e: print(f识别出错: {str(e)}) def start(self): 启动实时转录 self.is_recording True # 启动录音线程 record_thread threading.Thread(targetself.record_audio) record_thread.daemon True record_thread.start() # 启动识别线程可以启动多个提高并发 transcribe_threads [] for i in range(2): # 两个识别线程 t threading.Thread(targetself.transcribe_worker) t.daemon True t.start() transcribe_threads.append(t) # 主线程显示结果 try: while True: try: text self.text_queue.get(timeout0.5) # 这里可以处理识别结果比如显示在界面上或发送到其他服务 print(f实时输出: {text}) except queue.Empty: continue except KeyboardInterrupt: print(\n停止转录) finally: self.is_recording False for t in transcribe_threads: t.join(timeout2) # 使用示例 if __name__ __main__: # 注意需要先安装pyaudio: pip install pyaudio transcriber RealtimeTranscriber(chunk_duration3) # 每3秒识别一次 transcriber.start()这个实现有几个关键点分段录音每3秒可调整录一段音频避免单次处理时间太长多线程处理录音和识别在不同的线程进行互不干扰队列管理用队列来传递音频数据和识别结果线程安全实时输出识别结果几乎实时显示有3秒左右的延迟实际测试中延迟主要来自网络传输和模型推理。在本地部署的情况下延迟可以控制在3-5秒对于很多语音助手场景已经够用了。2.3 场景三视频字幕自动生成做视频内容的朋友都知道加字幕是个体力活。用Qwen3-ASR可以自动从视频中提取音频并生成字幕文件。import subprocess import os from pathlib import Path import json class VideoSubtitleGenerator: def __init__(self, api_basehttp://localhost:8000/v1): self.api_base api_base def extract_audio(self, video_path, audio_path): 从视频中提取音频 cmd [ ffmpeg, -i, video_path, -vn, # 不要视频 -acodec, pcm_s16le, # 音频编码 -ar, 16000, # 采样率 -ac, 1, # 单声道 -y, # 覆盖输出文件 audio_path ] try: subprocess.run(cmd, checkTrue, capture_outputTrue) print(f音频提取完成: {audio_path}) return True except subprocess.CalledProcessError as e: print(f音频提取失败: {e.stderr.decode()}) return False def transcribe_with_timestamps(self, audio_path, segment_duration30): 带时间戳的转录 segment_duration: 分段时长秒用于粗略时间戳 # 先获取音频总时长 cmd [ ffprobe, -v, error, -show_entries, formatduration, -of, defaultnoprint_wrappers1:nokey1, audio_path ] try: result subprocess.run(cmd, capture_outputTrue, textTrue) total_duration float(result.stdout.strip()) except: total_duration 0 # 分段处理音频 segments [] current_time 0 while current_time total_duration: # 提取音频片段 segment_file ftemp_segment_{current_time}.wav cmd [ ffmpeg, -i, audio_path, -ss, str(current_time), # 开始时间 -t, str(min(segment_duration, total_duration - current_time)), # 时长 -y, segment_file ] subprocess.run(cmd, capture_outputTrue) # 识别这个片段 text self.transcribe_segment(segment_file) if text: segments.append({ start: current_time, end: min(current_time segment_duration, total_duration), text: text }) # 清理临时文件 os.remove(segment_file) current_time segment_duration return segments def transcribe_segment(self, audio_file): 识别单个音频片段 import requests import base64 try: # 读取音频文件并转换为base64 with open(audio_file, rb) as f: audio_data base64.b64encode(f.read()).decode(utf-8) payload { model: /root/ai-models/Qwen/Qwen3-ASR-1___7B, messages: [{ role: user, content: [{ type: audio_url, audio_url: { url: fdata:audio/wav;base64,{audio_data} } }] }] } response requests.post( f{self.api_base}/chat/completions, jsonpayload, timeout30 ) if response.status_code 200: result response.json() text result[choices][0][message][content] if asr_text in text: text text.split(asr_text)[1].split(/asr_text)[0] return text except Exception as e: print(f识别片段失败: {str(e)}) return None def generate_srt(self, segments, output_path): 生成SRT字幕文件 srt_content for i, segment in enumerate(segments, 1): # 转换时间格式 start_time self.format_time(segment[start]) end_time self.format_time(segment[end]) srt_content f{i}\n srt_content f{start_time} -- {end_time}\n srt_content f{segment[text]}\n\n with open(output_path, w, encodingutf-8) as f: f.write(srt_content) print(fSRT字幕已生成: {output_path}) return output_path def format_time(self, seconds): 将秒数转换为SRT时间格式 hours int(seconds // 3600) minutes int((seconds % 3600) // 60) secs seconds % 60 return f{hours:02d}:{minutes:02d}:{secs:06.3f}.replace(., ,) def process_video(self, video_path, output_dirsubtitles): 处理整个视频文件 # 创建输出目录 os.makedirs(output_dir, exist_okTrue) video_name Path(video_path).stem audio_path Path(output_dir) / f{video_name}.wav print(f处理视频: {video_path}) # 1. 提取音频 if not self.extract_audio(video_path, str(audio_path)): return False # 2. 分段识别 print(开始语音识别...) segments self.transcribe_with_timestamps(str(audio_path)) if not segments: print(识别失败无结果) return False # 3. 生成字幕文件 srt_path Path(output_dir) / f{video_name}.srt self.generate_srt(segments, str(srt_path)) # 4. 生成JSON格式的详细结果 json_path Path(output_dir) / f{video_name}.json result { video: video_path, audio: str(audio_path), subtitle: str(srt_path), segments: segments, total_duration: segments[-1][end] if segments else 0 } with open(json_path, w, encodingutf-8) as f: json.dump(result, f, ensure_asciiFalse, indent2) print(f处理完成) print(f- 音频文件: {audio_path}) print(f- 字幕文件: {srt_path}) print(f- 详细结果: {json_path}) return True # 使用示例 if __name__ __main__: # 注意需要先安装ffmpeg generator VideoSubtitleGenerator() # 处理单个视频 video_file /path/to/your/video.mp4 generator.process_video(video_file) # 或者批量处理 video_folder /path/to/your/videos output_base ./generated_subtitles for video in Path(video_folder).glob(*.mp4): output_dir Path(output_base) / video.stem generator.process_video(str(video), str(output_dir))这个脚本实现了完整的视频字幕生成流程音频提取用ffmpeg从视频中提取音频分段识别把长音频切成30秒一段分别识别时间戳对齐给每段文字加上开始和结束时间格式转换生成标准的SRT字幕文件几乎所有视频播放器都支持批量处理可以一次处理整个文件夹的视频我测试了一个10分钟的教学视频生成字幕大概用了8分钟准确率在90%左右。对于口播类内容这个准确率完全够用只需要稍微修改一下专业术语就行。3. 性能优化与问题排查用了一段时间后你可能会遇到一些性能问题或者想优化使用体验。这里分享几个实用的技巧。3.1 调整配置提升性能默认配置可能不适合所有场景根据你的需求调整这些参数修改GPU内存使用 如果遇到显存不足的问题可以调整模型使用的显存比例。修改容器里的这个文件# 进入容器 docker exec -it 容器名称 bash # 编辑启动脚本 vi /root/Qwen3-ASR-1.7B/scripts/start_asr.sh # 找到这行修改GPU_MEMORY的值 # 默认是0.8使用80%显存可以改成0.6或0.5 GPU_MEMORY0.6 # 重启服务 supervisorctl restart qwen3-asr-1.7b调整并发处理数 如果你需要同时处理很多音频可以调整vLLM的并发设置。在同一个启动脚本里找到这行# 增加--max-num-batched-tokens参数提高并发能力 --max-num-batched-tokens 4096 # 默认可能是2048可以适当增加选择适合的音频格式 不同音频格式的处理速度不一样一般来说WAV处理最快没有压缩但文件大MP3最常见压缩率高处理速度中等FLAC无损压缩质量好文件比WAV小M4A/AAC苹果设备常用压缩效率高如果追求速度建议用WAV格式。如果考虑存储空间用MP3或AAC。3.2 常见问题解决问题1服务启动失败报GPU错误# 查看详细错误信息 docker logs 你的容器名称 # 常见错误1CUDA版本不匹配 # 解决方法确保CUDA版本是11.7或12.0 nvidia-smi # 查看CUDA版本 # 常见错误2显存不足 # 解决方法按上面说的调整GPU_MEMORY参数或者换更大显存的显卡问题2识别速度慢可能的原因和解决方法音频文件太大长音频可以分段处理比如每5分钟切一段网络延迟如果API服务在远程服务器网络会影响速度。考虑本地部署GPU被其他程序占用用nvidia-smi查看GPU使用情况关掉不必要的程序音频质量差背景噪音大、音量太小都会影响识别速度和准确率问题3识别准确率不高提高准确率的方法预处理音频用工具降噪、标准化音量选择正确语言如果知道音频的语言在请求时指定比如language: Chinese分段处理长音频分段识别每段2-3分钟效果最好后处理对识别结果进行简单的文本校正比如纠正常见的同音字这里有个简单的后处理示例def post_process_text(text, correction_dictNone): 对识别结果进行后处理 correction_dict: 自定义校正词典比如 {语音助力: 语音助理} if correction_dict: for wrong, correct in correction_dict.items(): text text.replace(wrong, correct) # 简单的标点修正 import re text re.sub(r\s([。]), r\1, text) # 去掉标点前的空格 text re.sub(r([。])\s, r\1 , text) # 标点后加空格 return text # 使用示例 correction_dict { 语音助力: 语音助理, 智能加湿: 智能家居, 原宇宙: 元宇宙 } raw_text 这是一个语音助力系统的演示它属于智能加湿领域也涉及原宇宙应用。 processed post_process_text(raw_text, correction_dict) print(processed) # 输出这是一个语音助理系统的演示它属于智能家居领域也涉及元宇宙应用。4. 总结通过上面的介绍和示例你应该对Qwen3-ASR-1.7B有了比较全面的了解。我来总结几个关键点这个方案的优势很明显部署简单Docker一键部署不需要复杂的环境配置多语言支持30种语言22种方言覆盖大部分使用场景准确率不错1.7B的模型规模在中等硬件上就能达到商用级准确率使用灵活既有Web界面适合手动操作也有API接口方便集成成本可控相比商用语音识别服务自建的成本更低数据也更安全实际使用中的建议根据场景选择部署方式如果只是偶尔用用Web界面就够了。如果需要集成到自己的系统里用API更合适。注意音频质量清晰的音频能大幅提升识别准确率。如果源音频质量差可以考虑先用工具处理一下。合理分段对于长音频分段处理效果更好也更容易排查问题。结合后处理针对专业领域准备一个术语校正词典能显著提升专业内容的准确率。监控资源使用定期检查GPU显存和内存使用情况避免服务崩溃。下一步可以尝试的方向实时流式识别虽然本文实现了准实时识别但真正的流式识别延迟更低适合对话场景。说话人分离在会议录音场景如果能区分不同说话人整理起来更方便。情感分析结合在客服场景结合情感分析可以更好地理解客户情绪。多模态扩展结合视觉信息比如在视频会议中同时识别语音和唇形提升准确率。语音识别技术正在快速进步像Qwen3-ASR这样的开源模型让更多人能够低成本地用上先进技术。无论你是开发者想集成到自己的产品里还是普通用户想提高工作效率都值得尝试一下。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。