Qwen3-ASR语音识别功能体验：API调用与Python客户端使用教程
·
Qwen3-ASR语音识别功能体验：API调用与Python客户端使用教程
刚部署好Qwen3-ASR语音识别服务，看着终端里显示“Running on public URL: http://0.0.0.0:7860”，是不是有点兴奋又有点迷茫？兴奋的是，这个支持30多种语言、22种中文方言的识别引擎终于跑起来了；迷茫的是，除了浏览器里那个简单的上传界面，我该怎么把它集成到自己的应用里？怎么用代码批量处理音频？怎么让它识别得更准、更快？
别担心，这篇文章就是为你准备的。我不讲复杂的模型原理，也不扯那些高大上的架构设计，就聚焦一件事：怎么用最简单、最直接的方式，让Qwen3-ASR为你干活。从最基本的API调用，到实用的Python客户端封装，再到提升识别准确率的小技巧，我会用最直白的语言，带你一步步把Qwen3-ASR变成你手边的语音识别工具。
1. 准备工作：确认服务状态与测试环境
在开始写代码之前，咱们得先确保服务是正常运行的。这就像开车前要检查油表一样，基础但重要。
1.1 快速检查服务状态
打开终端，执行下面这个命令：
curl -s http://localhost:7860/api/health
如果返回{"status":"OK"}，恭喜你，服务一切正常。如果返回Connection refused或者超时，那可能是服务没启动或者端口不对。
常见问题排查：
- 服务没启动：执行
/root/Qwen3-ASR-1.7B/start.sh重新启动 - 端口被占用：检查
7860端口是否被其他程序占用，可以用sudo lsof -i :7860查看 - 改了端口：如果你修改了默认端口，记得把上面的
localhost:7860换成你实际用的端口
1.2 准备测试音频
找一段测试用的音频文件，建议用.wav格式，因为这是最兼容的格式。如果你手头没有，可以用下面这个命令快速生成一段测试音频（需要安装ffmpeg）：
# 生成一段5秒的测试音频，内容是“你好，世界”
echo "你好，世界" | text2wave -o test.wav
或者更简单点，直接用手机录一段语音，保存为.wav格式。建议时长在10-30秒之间，太短了测试不出效果，太长了第一次调用可能等待时间比较久。
音频格式要求：
- 格式：WAV、MP3、FLAC、OGG等常见格式都支持
- 采样率：16000Hz或以上
- 声道：单声道或立体声都可以，系统会自动处理
- 大小：建议不超过50MB，大文件可以分段处理
2. 基础API调用：三种方法任你选
Qwen3-ASR提供了标准的HTTP API接口，这意味着你可以用几乎任何编程语言来调用它。下面我介绍三种最常用的方法，从最简单到最灵活。
2.1 方法一：用cURL快速测试（命令行党最爱）
如果你喜欢在命令行里操作，或者想快速验证服务是否正常，cURL是最直接的选择。
基本调用：
curl -X POST http://localhost:7860/api/predict \
-F "audio=@/path/to/your/audio.wav"
执行后，你会看到类似这样的返回：
{
"text": "你好，欢迎使用Qwen3-ASR语音识别服务",
"language": "zh",
"duration": 3.2,
"segments": [
{
"text": "你好，欢迎使用",
"start": 0.0,
"end": 1.5
},
{
"text": "Qwen3-ASR语音识别服务",
"start": 1.5,
"end": 3.2
}
]
}
带参数的调用：
如果你想指定语言或者调整其他参数，可以这样：
curl -X POST "http://localhost:7860/api/predict" \
-F "audio=@audio.wav" \
-F "language=zh" \
-F "task=transcribe" \
-F "vad_filter=true"
参数说明：
language：指定语言代码，比如zh（中文）、en（英文）、ja（日语）。如果不指定，系统会自动检测task：任务类型，transcribe是转写，translate是翻译（需要模型支持）vad_filter：是否启用语音活动检测，true可以过滤掉静音部分
2.2 方法二：用Python requests库（最常用）
对于大多数Python开发者来说，requests库是最熟悉的选择。代码简单直观，容易集成到现有项目中。
基础版本：
import requests
import json
def transcribe_audio(audio_path, server_url="http://localhost:7860"):
"""
调用Qwen3-ASR API进行语音识别
Args:
audio_path: 音频文件路径
server_url: 服务地址，默认localhost:7860
Returns:
dict: 识别结果，包含文本、语言、时间戳等信息
"""
url = f"{server_url}/api/predict"
try:
with open(audio_path, "rb") as audio_file:
files = {"audio": audio_file}
response = requests.post(url, files=files)
if response.status_code == 200:
return response.json()
else:
print(f"请求失败，状态码：{response.status_code}")
print(f"错误信息：{response.text}")
return None
except FileNotFoundError:
print(f"音频文件不存在：{audio_path}")
return None
except requests.exceptions.ConnectionError:
print(f"无法连接到服务，请检查服务是否运行在 {server_url}")
return None
# 使用示例
if __name__ == "__main__":
result = transcribe_audio("test.wav")
if result:
print(f"识别文本：{result['text']}")
print(f"检测语言：{result['language']}")
print(f"音频时长：{result['duration']}秒")
带参数的高级版本：
def transcribe_with_params(audio_path, language=None, task="transcribe", vad_filter=True):
"""
带参数的语音识别
Args:
audio_path: 音频文件路径
language: 指定语言，如'zh', 'en', 'ja'，None表示自动检测
task: 任务类型，'transcribe'或'translate'
vad_filter: 是否启用VAD过滤
Returns:
dict: 识别结果
"""
url = "http://localhost:7860/api/predict"
# 准备请求数据
files = {"audio": open(audio_path, "rb")}
data = {
"task": task,
"vad_filter": str(vad_filter).lower()
}
if language:
data["language"] = language
try:
response = requests.post(url, files=files, data=data)
files["audio"].close() # 记得关闭文件
if response.status_code == 200:
return response.json()
else:
print(f"识别失败：{response.text}")
return None
except Exception as e:
print(f"请求异常：{e}")
return None
# 使用示例：指定中文识别
result = transcribe_with_params("meeting.wav", language="zh", vad_filter=True)
2.3 方法三：用Python封装成客户端类（推荐用于项目）
如果你要在项目里频繁使用语音识别，我强烈建议封装成一个客户端类。这样代码更整洁，也更容易维护。
import requests
import json
from typing import Optional, Dict, Any
import os
class QwenASRClient:
"""Qwen3-ASR语音识别客户端"""
def __init__(self, base_url: str = "http://localhost:7860", timeout: int = 30):
"""
初始化客户端
Args:
base_url: 服务基础URL
timeout: 请求超时时间（秒）
"""
self.base_url = base_url.rstrip("/")
self.timeout = timeout
self.session = requests.Session()
def transcribe(
self,
audio_path: str,
language: Optional[str] = None,
task: str = "transcribe",
vad_filter: bool = True,
return_timestamps: bool = True
) -> Optional[Dict[str, Any]]:
"""
语音识别主方法
Args:
audio_path: 音频文件路径
language: 指定语言代码
task: 任务类型
vad_filter: 是否启用VAD过滤
return_timestamps: 是否返回时间戳
Returns:
识别结果字典，失败返回None
"""
if not os.path.exists(audio_path):
print(f"错误：音频文件不存在 - {audio_path}")
return None
url = f"{self.base_url}/api/predict"
# 准备请求参数
files = {"audio": open(audio_path, "rb")}
data = {
"task": task,
"vad_filter": str(vad_filter).lower(),
"return_timestamps": str(return_timestamps).lower()
}
if language:
data["language"] = language
try:
response = self.session.post(
url,
files=files,
data=data,
timeout=self.timeout
)
files["audio"].close()
if response.status_code == 200:
return response.json()
else:
print(f"API请求失败 [{response.status_code}]: {response.text}")
return None
except requests.exceptions.Timeout:
print(f"请求超时（{self.timeout}秒），请检查服务状态或增大超时时间")
return None
except requests.exceptions.ConnectionError:
print(f"无法连接到服务：{self.base_url}")
return None
except Exception as e:
print(f"未知错误：{e}")
return None
def batch_transcribe(
self,
audio_paths: list,
language: Optional[str] = None,
max_workers: int = 4
) -> Dict[str, Any]:
"""
批量语音识别
Args:
audio_paths: 音频文件路径列表
language: 指定语言代码
max_workers: 最大并发数
Returns:
字典，key为文件路径，value为识别结果
"""
from concurrent.futures import ThreadPoolExecutor, as_completed
results = {}
def process_file(path):
return path, self.transcribe(path, language=language)
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_path = {
executor.submit(process_file, path): path
for path in audio_paths
}
for future in as_completed(future_to_path):
path = future_to_path[future]
try:
_, result = future.result()
results[path] = result
print(f"已完成：{path}")
except Exception as e:
print(f"处理失败 {path}: {e}")
results[path] = None
return results
def get_service_info(self) -> Optional[Dict[str, Any]]:
"""获取服务信息"""
try:
response = self.session.get(
f"{self.base_url}/api/health",
timeout=5
)
if response.status_code == 200:
return response.json()
except:
pass
return None
# 使用示例
if __name__ == "__main__":
# 创建客户端实例
client = QwenASRClient(base_url="http://localhost:7860")
# 检查服务状态
info = client.get_service_info()
if info:
print(f"服务状态：{info.get('status', 'unknown')}")
# 单文件识别
result = client.transcribe(
audio_path="test.wav",
language="zh", # 指定中文
vad_filter=True
)
if result:
print(f"识别结果：{result['text']}")
print(f"分段信息：")
for seg in result.get('segments', []):
print(f" [{seg['start']:.1f}s - {seg['end']:.1f}s]: {seg['text']}")
# 批量识别
audio_files = ["audio1.wav", "audio2.wav", "audio3.wav"]
batch_results = client.batch_transcribe(audio_files, language="zh")
for file_path, res in batch_results.items():
if res:
print(f"{file_path}: {res['text'][:50]}...")
这个客户端类的好处很明显：
- 代码复用：一次封装，到处使用
- 错误处理：内置了完整的错误处理逻辑
- 批量处理：支持并发处理多个音频文件
- 易于扩展：可以很方便地添加新功能
3. 实战应用：几个真实场景的代码示例
光说不练假把式，下面我给出几个实际应用场景的完整代码示例，你可以直接复制使用。
3.1 场景一：会议录音转文字
假设你有一个会议录音文件，需要转换成文字记录，并且要区分不同说话人（虽然Qwen3-ASR本身不区分说话人，但我们可以用时间戳来辅助）。
import os
from datetime import datetime
from qwen_asr_client import QwenASRClient # 假设上面封装的类保存为qwen_asr_client.py
class MeetingTranscriber:
"""会议录音转文字工具"""
def __init__(self, client):
self.client = client
def transcribe_meeting(self, audio_path, output_format="txt"):
"""
转录会议录音
Args:
audio_path: 会议录音文件路径
output_format: 输出格式，支持'txt', 'json', 'srt'
"""
print(f"开始处理会议录音：{audio_path}")
print(f"文件大小：{os.path.getsize(audio_path) / 1024 / 1024:.2f} MB")
# 调用语音识别
result = self.client.transcribe(
audio_path,
language="zh", # 会议通常是中文
vad_filter=True, # 启用VAD过滤静音
return_timestamps=True
)
if not result:
print("识别失败")
return
# 提取基本信息
total_duration = result['duration']
text = result['text']
segments = result.get('segments', [])
print(f"识别完成！")
print(f"音频时长：{total_duration:.1f}秒")
print(f"识别文本长度：{len(text)}字符")
print(f"分段数量：{len(segments)}")
# 根据格式保存结果
if output_format == "txt":
self._save_as_txt(audio_path, text, segments)
elif output_format == "json":
self._save_as_json(audio_path, result)
elif output_format == "srt":
self._save_as_srt(audio_path, segments)
print(f"结果已保存到：{audio_path}.{output_format}")
def _save_as_txt(self, audio_path, text, segments):
"""保存为纯文本格式"""
output_path = f"{audio_path}.txt"
with open(output_path, "w", encoding="utf-8") as f:
f.write(f"会议录音转录结果\n")
f.write(f"文件：{os.path.basename(audio_path)}\n")
f.write(f"转录时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
f.write("=" * 50 + "\n\n")
f.write("完整文本：\n")
f.write(text + "\n\n")
f.write("分段文本（带时间戳）：\n")
for i, seg in enumerate(segments, 1):
start_time = self._format_time(seg['start'])
end_time = self._format_time(seg['end'])
f.write(f"[{start_time} - {end_time}] {seg['text']}\n")
def _save_as_json(self, audio_path, result):
"""保存为JSON格式"""
import json
output_path = f"{audio_path}.json"
with open(output_path, "w", encoding="utf-8") as f:
json.dump(result, f, ensure_ascii=False, indent=2)
def _save_as_srt(self, audio_path, segments):
"""保存为SRT字幕格式"""
output_path = f"{audio_path}.srt"
with open(output_path, "w", encoding="utf-8") as f:
for i, seg in enumerate(segments, 1):
start_time = self._format_srt_time(seg['start'])
end_time = self._format_srt_time(seg['end'])
f.write(f"{i}\n")
f.write(f"{start_time} --> {end_time}\n")
f.write(f"{seg['text']}\n\n")
def _format_time(self, seconds):
"""格式化时间为 MM:SS"""
minutes = int(seconds // 60)
secs = int(seconds % 60)
return f"{minutes:02d}:{secs:02d}"
def _format_srt_time(self, seconds):
"""格式化时间为 SRT 格式 HH:MM:SS,mmm"""
hours = int(seconds // 3600)
minutes = int((seconds % 3600) // 60)
secs = int(seconds % 60)
millis = int((seconds - int(seconds)) * 1000)
return f"{hours:02d}:{minutes:02d}:{secs:02d},{millis:03d}"
# 使用示例
if __name__ == "__main__":
# 创建客户端
client = QwenASRClient(base_url="http://localhost:7860")
# 创建转录器
transcriber = MeetingTranscriber(client)
# 转录会议录音
transcriber.transcribe_meeting(
audio_path="meeting_recording.wav",
output_format="txt" # 也可以选'json'或'srt'
)
3.2 场景二：多语言音频批量处理
如果你有多个不同语言的音频文件需要处理，这个脚本会很有用。
import glob
import json
from collections import defaultdict
class MultilingualAudioProcessor:
"""多语言音频批量处理器"""
def __init__(self, client):
self.client = client
self.language_map = {
'zh': '中文',
'en': '英文',
'ja': '日文',
'ko': '韩文',
'fr': '法文',
'de': '德文',
'es': '西班牙文',
# 可以继续添加其他语言
}
def process_directory(self, directory_path, output_dir="results"):
"""
处理目录下的所有音频文件
Args:
directory_path: 音频文件目录
output_dir: 输出目录
"""
# 创建输出目录
os.makedirs(output_dir, exist_ok=True)
# 查找所有音频文件
audio_extensions = ['*.wav', '*.mp3', '*.flac', '*.ogg', '*.m4a']
audio_files = []
for ext in audio_extensions:
audio_files.extend(glob.glob(os.path.join(directory_path, ext)))
print(f"找到 {len(audio_files)} 个音频文件")
# 按语言分类的结果
results_by_language = defaultdict(list)
# 处理每个文件
for i, audio_file in enumerate(audio_files, 1):
print(f"\n处理文件 {i}/{len(audio_files)}: {os.path.basename(audio_file)}")
# 先自动检测语言
auto_result = self.client.transcribe(
audio_file,
language=None, # 不指定语言，让模型自动检测
vad_filter=True
)
if not auto_result:
print(f" × 识别失败")
continue
detected_lang = auto_result.get('language', 'unknown')
lang_name = self.language_map.get(detected_lang, detected_lang)
print(f" ✓ 检测到语言: {lang_name} ({detected_lang})")
print(f" 识别文本: {auto_result['text'][:100]}...")
# 保存结果
result_data = {
'file': os.path.basename(audio_file),
'detected_language': detected_lang,
'text': auto_result['text'],
'duration': auto_result['duration'],
'segments': auto_result.get('segments', [])
}
# 按语言分类
results_by_language[detected_lang].append(result_data)
# 保存单个文件结果
self._save_single_result(audio_file, result_data, output_dir)
# 保存按语言分类的汇总结果
self._save_language_summary(results_by_language, output_dir)
print(f"\n处理完成！结果保存在: {output_dir}")
def _save_single_result(self, audio_file, result_data, output_dir):
"""保存单个文件的结果"""
base_name = os.path.splitext(os.path.basename(audio_file))[0]
output_file = os.path.join(output_dir, f"{base_name}.json")
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(result_data, f, ensure_ascii=False, indent=2)
def _save_language_summary(self, results_by_language, output_dir):
"""保存按语言分类的汇总结果"""
summary = {
'total_files': sum(len(files) for files in results_by_language.values()),
'languages': {}
}
for lang_code, files in results_by_language.items():
lang_name = self.language_map.get(lang_code, lang_code)
summary['languages'][lang_name] = {
'code': lang_code,
'file_count': len(files),
'total_duration': sum(f['duration'] for f in files),
'files': [f['file'] for f in files]
}
summary_file = os.path.join(output_dir, "language_summary.json")
with open(summary_file, 'w', encoding='utf-8') as f:
json.dump(summary, f, ensure_ascii=False, indent=2)
# 同时生成一个易读的文本报告
report_file = os.path.join(output_dir, "report.txt")
with open(report_file, 'w', encoding='utf-8') as f:
f.write("多语言音频处理报告\n")
f.write("=" * 50 + "\n\n")
f.write(f"总计处理文件数: {summary['total_files']}\n")
f.write(f"检测到的语言种类: {len(summary['languages'])}\n\n")
for lang_name, info in summary['languages'].items():
f.write(f"{lang_name} ({info['code']}):\n")
f.write(f" 文件数量: {info['file_count']}\n")
f.write(f" 总时长: {info['total_duration']:.1f}秒\n")
f.write(f" 文件列表: {', '.join(info['files'][:5])}")
if len(info['files']) > 5:
f.write(f" ...等{len(info['files'])}个文件")
f.write("\n\n")
# 使用示例
if __name__ == "__main__":
client = QwenASRClient()
processor = MultilingualAudioProcessor(client)
# 处理一个目录下的所有音频文件
processor.process_directory(
directory_path="/path/to/audio/files",
output_dir="./transcription_results"
)
3.3 场景三：实时音频流识别（进阶）
虽然Qwen3-ASR默认不支持真正的实时流式识别，但我们可以通过分段上传的方式模拟实时效果。这对于需要"准实时"识别的场景很有用。
import pyaudio
import wave
import threading
import queue
import time
from datetime import datetime
class StreamTranscriber:
"""流式转录器（模拟实时）"""
def __init__(self, client, chunk_duration=5):
"""
初始化流式转录器
Args:
client: QwenASRClient实例
chunk_duration: 每个音频块的时长（秒）
"""
self.client = client
self.chunk_duration = chunk_duration
self.audio_queue = queue.Queue()
self.results_queue = queue.Queue()
self.is_recording = False
# 音频参数
self.format = pyaudio.paInt16
self.channels = 1
self.rate = 16000
self.chunk_size = 1024
def start_recording(self):
"""开始录音"""
self.is_recording = True
self.audio_data = []
# 创建音频流
self.audio = pyaudio.PyAudio()
self.stream = self.audio.open(
format=self.format,
channels=self.channels,
rate=self.rate,
input=True,
frames_per_buffer=self.chunk_size
)
print(f"开始录音，采样率：{self.rate}Hz，单声道")
print("按 Ctrl+C 停止录音")
# 启动录音线程
self.record_thread = threading.Thread(target=self._record_audio)
self.record_thread.start()
# 启动处理线程
self.process_thread = threading.Thread(target=self._process_chunks)
self.process_thread.start()
# 启动结果输出线程
self.output_thread = threading.Thread(target=self._output_results)
self.output_thread.start()
def stop_recording(self):
"""停止录音"""
self.is_recording = False
if hasattr(self, 'stream'):
self.stream.stop_stream()
self.stream.close()
if hasattr(self, 'audio'):
self.audio.terminate()
# 等待线程结束
if hasattr(self, 'record_thread'):
self.record_thread.join()
# 处理剩余的音频数据
self._process_remaining_audio()
# 等待处理线程结束
self.audio_queue.put(None) # 发送结束信号
if hasattr(self, 'process_thread'):
self.process_thread.join()
# 等待输出线程结束
self.results_queue.put(None)
if hasattr(self, 'output_thread'):
self.output_thread.join()
print("\n录音已停止")
def _record_audio(self):
"""录音线程函数"""
try:
while self.is_recording:
data = self.stream.read(self.chunk_size, exception_on_overflow=False)
self.audio_data.append(data)
# 每积累够一个chunk的时长，就放入队列
current_duration = len(self.audio_data) * self.chunk_size / self.rate
if current_duration >= self.chunk_duration:
chunk = b''.join(self.audio_data)
self.audio_queue.put(chunk)
self.audio_data = [] # 清空，开始积累下一个chunk
except Exception as e:
print(f"录音错误：{e}")
def _process_chunks(self):
"""处理音频块的线程函数"""
while True:
chunk_data = self.audio_queue.get()
if chunk_data is None: # 结束信号
break
# 将音频数据保存为临时文件
temp_file = self._save_chunk_to_file(chunk_data)
if temp_file:
# 调用语音识别
result = self.client.transcribe(
temp_file,
language="zh", # 假设是中文
vad_filter=True
)
# 清理临时文件
os.remove(temp_file)
if result:
self.results_queue.put(result)
def _output_results(self):
"""输出结果的线程函数"""
while True:
result = self.results_queue.get()
if result is None: # 结束信号
break
current_time = datetime.now().strftime("%H:%M:%S")
text = result.get('text', '')
if text.strip(): # 只输出非空文本
print(f"[{current_time}] {text}")
def _save_chunk_to_file(self, audio_data):
"""将音频数据保存为临时WAV文件"""
import tempfile
try:
# 创建临时文件
temp_fd, temp_path = tempfile.mkstemp(suffix='.wav')
os.close(temp_fd)
# 写入WAV文件
with wave.open(temp_path, 'wb') as wf:
wf.setnchannels(self.channels)
wf.setsampwidth(self.audio.get_sample_size(self.format))
wf.setframerate(self.rate)
wf.writeframes(audio_data)
return temp_path
except Exception as e:
print(f"保存临时文件失败：{e}")
return None
def _process_remaining_audio(self):
"""处理最后剩余的音频数据"""
if self.audio_data:
chunk = b''.join(self.audio_data)
self.audio_queue.put(chunk)
# 使用示例
if __name__ == "__main__":
# 注意：这个示例需要安装pyaudio库
# pip install pyaudio
try:
import pyaudio
except ImportError:
print("请先安装pyaudio: pip install pyaudio")
exit(1)
# 创建客户端
client = QwenASRClient()
# 创建流式转录器（每5秒识别一次）
transcriber = StreamTranscriber(client, chunk_duration=5)
print("准备开始实时语音识别...")
print("将会每5秒输出一次识别结果")
print("-" * 50)
try:
# 开始录音和识别
transcriber.start_recording()
# 主线程等待用户输入
input("按回车键停止录音...\n")
except KeyboardInterrupt:
print("\n接收到中断信号")
finally:
# 停止录音
transcriber.stop_recording()
4. 性能优化与实用技巧
掌握了基本用法后，再来看看怎么让Qwen3-ASR跑得更快、更准、更稳定。
4.1 调整识别参数提升准确率
Qwen3-ASR支持一些参数调整，可以在调用API时指定：
# 优化后的识别参数
optimized_result = client.transcribe(
audio_path="important_audio.wav",
language="zh", # 明确指定语言可以提高准确率
vad_filter=True, # 启用语音活动检测，过滤静音
# 以下参数需要查看API文档确认是否支持
# beam_size=5, # 增大beam size可能提高准确率但会变慢
# temperature=0.0, # 降低temperature使输出更确定
# best_of=5, # 取多个结果中的最佳
)
实用建议：
- 明确指定语言：如果你知道音频的语言，一定要指定。自动检测虽然方便，但准确率不如明确指定
- 启用VAD过滤：对于有大量静音或背景噪声的音频，
vad_filter=True可以显著提升效果 - 预处理音频：识别前可以先对音频进行预处理，比如降噪、归一化音量等
4.2 处理长音频的策略
Qwen3-ASR对单次识别的音频长度有限制（具体限制取决于你的GPU显存）。对于超长音频，需要分段处理：
def transcribe_long_audio(audio_path, chunk_duration=300):
"""
处理长音频文件（分段识别）
Args:
audio_path: 音频文件路径
chunk_duration: 每段时长（秒），默认5分钟
"""
import librosa
# 加载音频
y, sr = librosa.load(audio_path, sr=16000)
duration = len(y) / sr
print(f"音频总时长：{duration:.1f}秒")
print(f"将分为 {int(duration / chunk_duration) + 1} 段处理")
results = []
# 分段处理
for i in range(0, len(y), int(chunk_duration * sr)):
chunk = y[i:i + int(chunk_duration * sr)]
if len(chunk) == 0:
continue
# 保存分段为临时文件
import soundfile as sf
import tempfile
with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as tmp:
temp_path = tmp.name
sf.write(temp_path, chunk, sr)
# 识别分段
start_time = i / sr
print(f"处理第 {len(results) + 1} 段 [{start_time:.1f}s开始]...")
result = client.transcribe(temp_path, language="zh")
if result:
# 调整时间戳
for seg in result.get('segments', []):
seg['start'] += start_time
seg['end'] += start_time
results.append(result)
# 删除临时文件
os.unlink(temp_path)
# 合并结果
full_text = " ".join([r['text'] for r in results])
# 合并分段
all_segments = []
for r in results:
all_segments.extend(r.get('segments', []))
return {
'text': full_text,
'duration': duration,
'segments': all_segments,
'chunks': len(results)
}
4.3 错误处理与重试机制
在生产环境中，网络波动或服务暂时不可用是常有的事。一个好的客户端应该有重试机制：
import time
from functools import wraps
def retry_on_failure(max_retries=3, delay=1):
"""重试装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
if attempt < max_retries - 1:
wait_time = delay * (2 ** attempt) # 指数退避
print(f"第{attempt + 1}次尝试失败，{wait_time}秒后重试...")
time.sleep(wait_time)
else:
print(f"所有{max_retries}次尝试均失败")
raise last_exception
return wrapper
return decorator
class RobustASRClient(QwenASRClient):
"""带重试机制的客户端"""
@retry_on_failure(max_retries=3, delay=2)
def transcribe_with_retry(self, audio_path, **kwargs):
"""带重试的识别方法"""
return self.transcribe(audio_path, **kwargs)
def transcribe_important(self, audio_path, **kwargs):
"""
重要音频的识别（带备份方案）
策略：
1. 先尝试高质量识别（指定语言，启用VAD）
2. 如果失败，尝试快速识别（不指定语言，禁用VAD）
3. 如果还失败，返回错误信息但继续执行
"""
# 第一次尝试：高质量模式
try:
result = self.transcribe_with_retry(
audio_path,
language=kwargs.get('language', 'zh'),
vad_filter=True,
return_timestamps=True
)
if result:
result['mode'] = 'high_quality'
return result
except Exception as e:
print(f"高质量模式失败：{e}")
# 第二次尝试：快速模式
try:
print("尝试快速模式...")
result = self.transcribe(
audio_path,
language=None, # 自动检测语言
vad_filter=False, # 禁用VAD加快速度
return_timestamps=False # 不返回时间戳
)
if result:
result['mode'] = 'fast'
return result
except Exception as e:
print(f"快速模式也失败：{e}")
# 所有尝试都失败
return {
'error': '所有识别尝试均失败',
'file': audio_path,
'text': '',
'mode': 'failed'
}
5. 总结：从调用到集成的完整路径
通过上面的内容，你应该已经掌握了Qwen3-ASR语音识别服务从基础调用到高级集成的完整技能。让我们回顾一下关键点：
第一，基础调用三选一：根据你的需求，可以选择最简单的cURL命令、最灵活的Python requests，或者最工程化的客户端类封装。对于大多数项目，我推荐使用客户端类的方式，代码更整洁，功能更完整。
第二，实战场景直接套用：无论是会议录音转写、多语言批量处理，还是准实时流式识别，我都提供了完整的代码示例。你可以直接复制使用，或者根据实际需求稍作修改。
第三，性能优化有技巧：通过调整识别参数、分段处理长音频、添加重试机制，你可以让Qwen3-ASR在各种环境下都稳定工作。记住，明确指定语言和启用VAD过滤是两个最有效的准确率提升手段。
第四，错误处理要周全：网络服务总有不可靠的时候。好的代码不仅要能处理成功的情况，更要能优雅地处理失败。重试机制、降级方案、详细日志，这些都是生产环境必备的。
现在，Qwen3-ASR已经不仅仅是浏览器里的一个上传界面了。你可以把它集成到你的会议系统、客服平台、内容生产流水线，或者任何需要语音转文字的地方。30多种语言、22种中文方言的识别能力，加上相对简单的API接口，让它成为了一个非常实用的工具。
最后提醒一点：虽然我提供了很多代码示例，但最好的学习方式还是动手实践。选一个你最需要的场景，从最简单的代码开始，一步步添加功能，遇到问题就回头看看这篇文章。用不了多久，你就能让Qwen3-ASR成为你工作中得力的语音识别助手。
获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
0
0- 0
已为社区贡献98条内容
所有评论(0)