Qwen3-ASR语音识别API调用指南:Python/cURL示例

1. 引言:语音识别的新选择

语音识别技术正在改变我们与设备交互的方式,从智能助手到会议转录,从客服系统到内容创作,语音转文字的需求无处不在。Qwen3-ASR作为新一代多语言语音识别服务,基于强大的Qwen3-ASR-1.7B模型构建,支持30多种语言和22种中文方言识别,为开发者提供了高质量的语音转文字解决方案。

本文将手把手教你如何通过Python和cURL两种方式调用Qwen3-ASR的API接口,无论你是刚入门的开发者还是经验丰富的工程师,都能快速上手使用这项强大的语音识别服务。

2. 环境准备与快速部署

2.1 系统要求检查

在开始调用API之前,确保你的系统满足以下基本要求:

  • Python 3.6+ 环境(推荐3.8+)
  • 网络连接正常,能够访问部署Qwen3-ASR服务的服务器
  • 音频文件准备(支持wav、mp3等常见格式)

2.2 服务部署确认

Qwen3-ASR服务通常部署在服务器端,默认服务地址为 http://<server-ip>:7860。如果你是系统管理员,可以通过以下方式启动服务:

# 快速启动服务
/root/Qwen3-ASR-1.7B/start.sh

# 或者使用systemd服务(生产环境推荐)
sudo systemctl start qwen3-asr

确保服务正常运行后,我们就可以开始调用API了。

3. Python客户端调用详解

3.1 基础环境配置

首先确保你的Python环境已安装requests库,这是调用HTTP API的基础:

pip install requests

3.2 最简单的语音识别调用

下面是一个完整的Python示例,展示如何调用Qwen3-ASR服务进行语音识别:

import requests
import json

# 服务地址 - 根据实际部署修改
service_url = "http://localhost:7860"

# 音频文件路径
audio_file_path = "path/to/your/audio.wav"

def transcribe_audio(audio_path):
    """
    调用Qwen3-ASR进行语音转文字
    """
    try:
        # 打开音频文件
        with open(audio_path, "rb") as audio_file:
            # 构建请求
            files = {"audio": audio_file}
            
            # 发送POST请求
            response = requests.post(
                f"{service_url}/api/predict", 
                files=files
            )
            
            # 检查响应状态
            if response.status_code == 200:
                result = response.json()
                return result
            else:
                print(f"请求失败,状态码: {response.status_code}")
                return None
                
    except Exception as e:
        print(f"发生错误: {str(e)}")
        return None

# 调用示例
if __name__ == "__main__":
    result = transcribe_audio(audio_file_path)
    if result:
        print("识别结果:", json.dumps(result, indent=2, ensure_ascii=False))
    else:
        print("识别失败")

3.3 高级功能调用示例

Qwen3-ASR支持更多高级参数,下面是一个增强版的调用示例:

import requests

def advanced_transcribe(audio_path, language=None, context=None):
    """
    高级语音识别调用,支持语言指定和上下文
    """
    service_url = "http://localhost:7860"
    
    # 构建请求数据
    files = {"audio": open(audio_path, "rb")}
    data = {}
    
    if language:
        data["language"] = language  # 可选参数:指定语言
    
    if context:
        data["context"] = context    # 可选参数:添加上下文信息
    
    try:
        response = requests.post(
            f"{service_url}/api/predict",
            files=files,
            data=data
        )
        
        if response.status_code == 200:
            return response.json()
        else:
            print(f"错误: {response.status_code} - {response.text}")
            return None
            
    except requests.exceptions.RequestException as e:
        print(f"网络错误: {str(e)}")
        return None
    finally:
        files["audio"].close()

# 使用示例
result = advanced_transcribe(
    audio_path="meeting_recording.wav",
    language="zh",  # 指定中文
    context="技术会议,讨论AI和机器学习"  # 提供上下文
)

4. cURL命令行调用指南

4.1 基础cURL调用

如果你习惯使用命令行工具,cURL是一个简单高效的选择:

# 最基本的调用方式
curl -X POST http://localhost:7860/api/predict \
  -F "audio=@audio.wav"

4.2 带参数的高级cURL调用

# 指定语言和上下文的调用
curl -X POST http://localhost:7860/api/predict \
  -F "audio=@audio.wav" \
  -F "language=zh" \
  -F "context=商务会议记录"

4.3 处理响应结果

cURL默认输出响应内容,你可以使用以下方式美化输出:

# 格式化JSON输出(需要jq工具)
curl -X POST http://localhost:7860/api/predict \
  -F "audio=@audio.wav" | jq .

# 或者保存到文件
curl -X POST http://localhost:7860/api/predict \
  -F "audio=@audio.wav" -o result.json

5. 实战案例与最佳实践

5.1 批量处理音频文件

在实际项目中,我们经常需要处理多个音频文件:

import os
import glob
import requests
from concurrent.futures import ThreadPoolExecutor

def batch_transcribe(audio_dir, output_dir="results"):
    """
    批量处理目录中的所有音频文件
    """
    # 创建输出目录
    os.makedirs(output_dir, exist_ok=True)
    
    # 获取所有音频文件
    audio_files = glob.glob(os.path.join(audio_dir, "*.wav")) + \
                 glob.glob(os.path.join(audio_dir, "*.mp3"))
    
    def process_file(audio_path):
        try:
            result = transcribe_audio(audio_path)
            if result:
                # 保存结果
                base_name = os.path.basename(audio_path)
                output_file = os.path.join(output_dir, f"{base_name}.json")
                with open(output_file, "w", encoding="utf-8") as f:
                    json.dump(result, f, ensure_ascii=False, indent=2)
                print(f"处理完成: {base_name}")
            return True
        except Exception as e:
            print(f"处理失败 {audio_path}: {str(e)}")
            return False
    
    # 使用线程池并行处理
    with ThreadPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(process_file, audio_files))
    
    success_count = sum(results)
    print(f"批量处理完成,成功: {success_count}/{len(audio_files)}")

# 使用示例
batch_transcribe("audio_files", "transcription_results")

5.2 实时音频流处理

对于实时应用场景,你可以这样处理音频流:

import pyaudio
import wave
import requests
from io import BytesIO

def record_and_transcribe(duration=5, sample_rate=16000):
    """
    录制音频并实时识别
    """
    # 录制音频
    chunk = 1024
    format = pyaudio.paInt16
    channels = 1
    
    p = pyaudio.PyAudio()
    
    stream = p.open(format=format,
                    channels=channels,
                    rate=sample_rate,
                    input=True,
                    frames_per_buffer=chunk)
    
    print("开始录音...")
    frames = []
    
    for i in range(0, int(sample_rate / chunk * duration)):
        data = stream.read(chunk)
        frames.append(data)
    
    print("录音结束")
    
    stream.stop_stream()
    stream.close()
    p.terminate()
    
    # 保存为内存中的WAV文件
    wav_buffer = BytesIO()
    with wave.open(wav_buffer, 'wb') as wf:
        wf.setnchannels(channels)
        wf.setsampwidth(p.get_sample_size(format))
        wf.setframerate(sample_rate)
        wf.writeframes(b''.join(frames))
    
    # 重置缓冲区位置并发送识别请求
    wav_buffer.seek(0)
    files = {'audio': ('recording.wav', wav_buffer, 'audio/wav')}
    
    response = requests.post(
        "http://localhost:7860/api/predict",
        files=files
    )
    
    if response.status_code == 200:
        return response.json()
    else:
        return None

6. 常见问题与解决方案

6.1 连接问题排查

如果遇到连接问题,可以按照以下步骤排查:

import requests

def check_service_status(url):
    """
    检查服务状态
    """
    try:
        response = requests.get(f"{url}/", timeout=5)
        return response.status_code == 200
    except:
        return False

# 检查服务是否可用
if check_service_status("http://localhost:7860"):
    print("服务正常运行")
else:
    print("服务不可用,请检查:")
    print("1. 服务是否启动:ps aux | grep qwen-asr")
    print("2. 端口是否监听:netstat -tlnp | grep 7860")
    print("3. 防火墙设置:sudo ufw status")

6.2 音频格式处理建议

Qwen3-ASR对音频格式有一定要求,以下是一些处理建议:

from pydub import AudioSegment
import io

def convert_audio_format(input_path, output_path, target_format="wav", sample_rate=16000):
    """
    转换音频格式为适合识别的格式
    """
    audio = AudioSegment.from_file(input_path)
    audio = audio.set_frame_rate(sample_rate).set_channels(1)
    audio.export(output_path, format=target_format)
    return output_path

def prepare_audio_for_api(audio_path):
    """
    准备音频文件用于API调用
    """
    # 如果是非WAV格式,先进行转换
    if not audio_path.lower().endswith('.wav'):
        converted_path = audio_path + '.converted.wav'
        convert_audio_format(audio_path, converted_path)
        return converted_path
    return audio_path

7. 总结

通过本文的详细指南,你应该已经掌握了使用Python和cURL调用Qwen3-ASR语音识别API的方法。无论是简单的单文件识别还是复杂的批量处理,Qwen3-ASR都能提供高质量的多语言语音转文字服务。

关键要点回顾

  • Python调用使用requests库,简单易用
  • cURL命令适合快速测试和脚本调用
  • 批量处理和实时流处理可以满足不同场景需求
  • 正确的音频格式和参数设置能提升识别准确率

在实际项目中,建议先进行小规模测试,确保服务稳定性和识别质量满足需求。对于生产环境,考虑添加重试机制、错误处理和性能监控,以构建健壮的语音识别应用。


获取更多AI镜像

想探索更多AI镜像和应用场景?访问 CSDN星图镜像广场,提供丰富的预置镜像,覆盖大模型推理、图像生成、视频生成、模型微调等多个领域,支持一键部署。

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐