# Qwen3-ASR-0.6B API开发手册：Python/JS调用转录服务完整示例 ## 1. 语音识别服务快速了解 Qwen3-ASR-0.6B是一个轻量级高性能语音识别模型，只有6亿参数却拥有出色的识别能力。它基于Qwen3-Omni基座和自研AuT语音编码器技术打造，专门为多语种识别和高效部署而设计。 这个模型最大的特点是既保持了高精度，又实现了低延迟和高并发处理，非常适合在边缘设备或云端部署。无论你是需要处理大量语音数据的企业应用，还是对响应速度有要求的实时应用，都能找到合适的应用场景。 服务提供了两种访问方式：直观的Web界面供非技术人员使用，以及完整的API接口供开发者集成。Web界面运行在8080端口，API服务则在8000端口提供服务。 ## 2. 核心功能特性 ### 2.1 多语言支持能力 Qwen3-ASR-0.6B最令人印象深刻的是其多语言支持能力，总共支持52种语言变体： **30种主流语言**包括：中文、英文、粤语、阿拉伯语、德语、法语、西班牙语、葡萄牙语、印尼语、意大利语、韩语、俄语、泰语、越南语、日语、土耳其语、印地语、马来语等常用语言。 **22种中文方言**覆盖了全国各地：包括安徽话、东北话、福建话、甘肃话、贵州话、河北话、河南话、湖北话、湖南话、江西话、宁夏话、山东话、陕西话、山西话、四川话、天津话、云南话、浙江话，以及吴语、闽南话等方言。 ### 2.2 音频格式与性能 模型支持常见的音频格式，让你不用花费时间在格式转换上： - **支持格式**：wav、mp3、m4a、flac、ogg - **文件大小**：最大支持100MB的音频文件 - **处理性能**：使用GPU加速，采用bfloat16精度，在保证质量的同时提升处理速度 这种格式兼容性意味着你可以直接上传手机录音、会议记录、播客音频等各种来源的文件，无需预先处理。 ## 3. 环境准备与快速测试 在开始编写代码之前，我们先确保服务正常运行并测试基本功能。 ### 3.1 服务健康检查 首先检查服务状态，使用简单的curl命令： ```bash curl http://你的服务器IP:8080/api/health ``` 正常响应应该类似这样： ```json { "status": "healthy", "model_loaded": true, "gpu_available": true, "gpu_memory": { "allocated": 1.46, "cached": 1.76 } } ``` 这个响应告诉你：服务正常运行、模型加载成功、GPU可用，并显示了当前GPU内存使用情况。 ### 3.2 测试文件转录 用实际音频文件测试转录功能： ```bash curl -X POST http://你的服务器IP:8080/api/transcribe \ -F "audio_file=@你的音频文件.mp3" \ -F "language=Chinese" ``` 如果一切正常，你会获得JSON格式的转录结果，包含识别出的文字内容。 ## 4. Python调用完整示例 现在我们来学习如何用Python代码调用语音识别服务。Python是数据处理和AI应用的首选语言，集成起来非常方便。 ### 4.1 安装必要依赖 首先确保安装了需要的Python库： ```bash pip install requests python-dotenv ``` requests库用于发送HTTP请求，python-dotenv用于管理环境变量（如服务器地址、API密钥等）。 ### 4.2 基础调用代码 创建一个简单的Python脚本实现语音转录： ```python import requests import json from pathlib import Path class QwenASRClient: def __init__(self, base_url="http://localhost:8080"): self.base_url = base_url def transcribe_file(self, file_path, language=None): """上传本地文件进行转录""" url = f"{self.base_url}/api/transcribe" with open(file_path, 'rb') as audio_file: files = {'audio_file': audio_file} data = {'language': language} if language else {} response = requests.post(url, files=files, data=data) response.raise_for_status() return response.json() def transcribe_url(self, audio_url, language=None): """通过URL转录在线音频""" url = f"{self.base_url}/api/transcribe_url" payload = {'audio_url': audio_url} if language: payload['language'] = language response = requests.post(url, json=payload) response.raise_for_status() return response.json() # 使用示例 if __name__ == "__main__": client = QwenASRClient("http://你的服务器IP:8080") # 转录本地文件 try: result = client.transcribe_file("meeting_recording.mp3", "Chinese") print("转录结果:", result['text']) except Exception as e: print(f"转录失败: {e}") ``` ### 4.3 高级功能实现 对于生产环境，我们需要更健壮的代码处理各种情况： ```python import requests from requests.exceptions import RequestException import time import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class RobustASRClient: def __init__(self, base_url, max_retries=3, timeout=30): self.base_url = base_url self.max_retries = max_retries self.timeout = timeout self.session = requests.Session() def transcribe_with_retry(self, file_path, language=None): """带重试机制的转录功能""" for attempt in range(self.max_retries): try: with open(file_path, 'rb') as f: files = {'audio_file': f} data = {'language': language} if language else {} response = self.session.post( f"{self.base_url}/api/transcribe", files=files, data=data, timeout=self.timeout ) response.raise_for_status() result = response.json() logger.info(f"转录成功: {file_path}") return result except RequestException as e: logger.warning(f"尝试 {attempt + 1} 失败: {e}") if attempt == self.max_retries - 1: raise time.sleep(2 ** attempt) # 指数退避 def batch_transcribe(self, file_paths, language=None): """批量处理多个音频文件""" results = [] for file_path in file_paths: try: result = self.transcribe_with_retry(file_path, language) results.append({ 'file': file_path, 'success': True, 'result': result }) except Exception as e: results.append({ 'file': file_path, 'success': False, 'error': str(e) }) return results # 使用示例 client = RobustASRClient("http://你的服务器IP:8080") # 批量处理多个文件 results = client.batch_transcribe([ "meeting1.mp3", "interview2.wav", "presentation3.m4a" ], "Chinese") for result in results: if result['success']: print(f"{result['file']}: {result['result']['text'][:100]}...") else: print(f"{result['file']} 失败: {result['error']}") ``` ## 5. JavaScript调用完整示例 对于前端应用或Node.js服务，JavaScript调用同样重要。以下是完整的JS实现方案。 ### 5.1 浏览器端调用 在浏览器中直接调用语音识别API： ```html <!DOCTYPE html> <html> <head> <title>语音识别示例</title> </head> <body> <input type="file" id="audioFile" accept=".mp3,.wav,.m4a,.flac,.ogg"> <button onclick="transcribeAudio()">开始转录</button> <div id="result"></div> <script> async function transcribeAudio() { const fileInput = document.getElementById('audioFile'); const resultDiv = document.getElementById('result'); if (!fileInput.files.length) { resultDiv.innerHTML = '请选择音频文件'; return; } const formData = new FormData(); formData.append('audio_file', fileInput.files[0]); formData.append('language', 'Chinese'); try { resultDiv.innerHTML = '处理中...'; const response = await fetch('http://你的服务器IP:8080/api/transcribe', { method: 'POST', body: formData }); if (!response.ok) { throw new Error(`HTTP错误: ${response.status}`); } const data = await response.json(); resultDiv.innerHTML = ` <h3>转录结果:</h3> <p>${data.text}</p> <p><small>处理时间: ${data.processing_time}s</small></p> `; } catch (error) { resultDiv.innerHTML = `错误: ${error.message}`; } } </script> </body> </html> ``` ### 5.2 Node.js服务端调用 在Node.js环境中调用语音识别服务： ```javascript const axios = require('axios'); const FormData = require('form-data'); const fs = require('fs'); class NodeASRClient { constructor(baseUrl = 'http://localhost:8080') { this.baseUrl = baseUrl; this.client = axios.create({ baseURL: baseUrl, timeout: 30000 }); } async transcribeFile(filePath, language = null) { try { const formData = new FormData(); formData.append('audio_file', fs.createReadStream(filePath)); if (language) { formData.append('language', language); } const response = await this.client.post('/api/transcribe', formData, { headers: formData.getHeaders() }); return response.data; } catch (error) { console.error('转录失败:', error.response?.data || error.message); throw error; } } async transcribeUrl(audioUrl, language = null) { try { const payload = { audio_url: audioUrl }; if (language) { payload.language = language; } const response = await this.client.post('/api/transcribe_url', payload); return response.data; } catch (error) { console.error('URL转录失败:', error.response?.data || error.message); throw error; } } } // 使用示例 async function main() { const client = new NodeASRClient('http://你的服务器IP:8080'); try { // 转录本地文件 const result = await client.transcribeFile('audio.mp3', 'Chinese'); console.log('转录结果:', result.text); // 或者转录在线音频 const urlResult = await client.transcribeUrl( 'https://example.com/audio.mp3', 'Chinese' ); console.log('在线音频结果:', urlResult.text); } catch (error) { console.error('处理失败:', error.message); } } // 如果是直接运行这个文件 if (require.main === module) { main(); } module.exports = NodeASRClient; ``` ### 5.3 前端框架集成示例 如果你使用React、Vue等现代前端框架，可以这样集成： ```javascript // React Hook示例 import { useState } from 'react'; function useSpeechRecognition(apiUrl) { const [isProcessing, setIsProcessing] = useState(false); const [result, setResult] = useState(null); const [error, setError] = useState(null); const transcribe = async (audioFile, language = null) => { setIsProcessing(true); setError(null); const formData = new FormData(); formData.append('audio_file', audioFile); if (language) { formData.append('language', language); } try { const response = await fetch(`${apiUrl}/api/transcribe`, { method: 'POST', body: formData }); if (!response.ok) { throw new Error(`请求失败: ${response.status}`); } const data = await response.json(); setResult(data); return data; } catch (err) { setError(err.message); throw err; } finally { setIsProcessing(false); } }; return { transcribe, isProcessing, result, error }; } // 在React组件中使用 function TranscriptionComponent() { const { transcribe, isProcessing, result, error } = useSpeechRecognition('http://你的服务器IP:8080'); const handleFileUpload = async (event) => { const file = event.target.files[0]; if (file) { try { await transcribe(file, 'Chinese'); } catch (err) { console.error('转录错误:', err); } } }; return ( <div> <input type="file" onChange={handleFileUpload} accept=".mp3,.wav,.m4a,.flac,.ogg" disabled={isProcessing} /> {isProcessing && <p>处理中...</p>} {error && ( <p style={{ color: 'red' }}>错误: {error}</p> )} {result && ( <div> <h3>转录结果:</h3> <p>{result.text}</p> </div> )} </div> ); } ``` ## 6. 实战技巧与最佳实践 在实际项目中应用语音识别服务时，这些技巧能帮你避免很多坑。 ### 6.1 错误处理与重试机制 网络请求总会遇到各种问题，良好的错误处理是必须的： ```python def safe_transcribe(client, file_path, max_attempts=3): """安全的转录函数，包含完善的错误处理""" for attempt in range(max_attempts): try: return client.transcribe_file(file_path, "Chinese") except requests.exceptions.ConnectionError: print(f"连接失败，尝试 {attempt + 1}/{max_attempts}") time.sleep(2 ** attempt) # 指数退避 except requests.exceptions.Timeout: print(f"请求超时，尝试 {attempt + 1}/{max_attempts}") time.sleep(2 ** attempt) except requests.exceptions.HTTPError as e: if e.response.status_code >= 500: print(f"服务器错误，尝试 {attempt + 1}/{max_attempts}") time.sleep(2 ** attempt) else: raise # 4xx错误直接抛出，不需要重试 raise Exception("所有重试尝试都失败了") # 使用示例 try: result = safe_transcribe(client, "important_meeting.mp3") print("转录成功:", result['text']) except Exception as e: print("最终失败:", e) ``` ### 6.2 性能优化建议 处理大量音频时，这些优化技巧能显著提升效率： ```python from concurrent.futures import ThreadPoolExecutor import os def process_audio_directory(directory_path, output_dir, language="Chinese"): """批量处理目录中的所有音频文件""" os.makedirs(output_dir, exist_ok=True) audio_files = [ f for f in os.listdir(directory_path) if f.lower().endswith(('.mp3', '.wav', '.m4a', '.flac', '.ogg')) ] def process_file(filename): try: file_path = os.path.join(directory_path, filename) result = client.transcribe_file(file_path, language) # 保存结果到文本文件 output_path = os.path.join(output_dir, f"{os.path.splitext(filename)[0]}.txt") with open(output_path, 'w', encoding='utf-8') as f: f.write(result['text']) return filename, True, None except Exception as e: return filename, False, str(e) # 使用线程池并行处理 with ThreadPoolExecutor(max_workers=4) as executor: # 根据服务器性能调整 results = list(executor.map(process_file, audio_files)) # 输出处理结果统计 successful = sum(1 for _, success, _ in results if success) print(f"处理完成: {successful}/{len(audio_files)} 成功") for filename, success, error in results: if not success: print(f"失败: {filename} - {error}") # 使用示例 process_audio_directory("audio_files", "transcription_results") ``` ## 7. 总结与后续步骤 通过本文的完整示例，你应该已经掌握了如何使用Python和JavaScript调用Qwen3-ASR-0.6B语音识别服务。从基础的单文件转录到高级的批量处理、错误处理和性能优化，这些代码可以直接用在你的项目中。 **关键要点回顾**： - 服务支持RESTful API调用，接口简单易用 - 多语言支持是最大亮点，覆盖52种语言变体 - 提供了文件上传和URL转录两种方式 - 完善的错误处理和重试机制对生产环境很重要 **下一步学习建议**： 1. 尝试处理更长的音频文件，测试服务的稳定性 2. 实验不同的语言设置，体验多语言识别能力 3. 在实际项目中集成，根据具体需求调整代码 4. 监控服务性能，优化并发处理参数 记住在实际部署时，要将示例中的"你的服务器IP"替换为实际的服务地址，并根据网络环境调整超时和重试参数。语音识别服务可以应用于会议记录、客服系统、内容转录、语音助手等多个场景，期待看到你的创新应用。 --- > **获取更多AI镜像** > > 想探索更多AI镜像和应用场景？访问 [CSDN星图镜像广场](https://ai.csdn.net/?utm_source=mirror_blog_end)，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。