Qwen3-ASR-1.7B实战教程流式输入适配开发基于FastAPI1. 引言如果你正在寻找一个能离线运行、支持多语言、识别速度还很快的语音转文字工具那么Qwen3-ASR-1.7B绝对值得你花时间了解一下。这个模型来自阿里通义千问团队有17亿参数支持中文、英文、日语、韩语、粤语等多种语言还能自动检测语言类型。最吸引人的是它完全可以在本地离线运行不需要连接任何外部服务数据安全有保障。官方提供的镜像已经很好用了通过Web界面就能上传音频文件进行转写。但如果你想让这个模型更“智能”一点比如支持实时语音输入、边说话边转写那就需要做一些开发工作了。今天我就来分享一个实战项目如何基于FastAPI为Qwen3-ASR-1.7B开发流式输入适配功能。我会带你从零开始一步步实现一个能实时处理语音流的服务。2. 为什么需要流式输入2.1 文件上传 vs 流式输入先说说两者的区别这样你就能明白为什么流式输入在某些场景下是刚需。文件上传模式当前镜像的默认方式你需要先录好音频文件上传整个文件到服务器服务器一次性处理完整个文件返回完整的转写结果这种方式适合事后转写比如会议录音整理、采访内容转录。但它有个明显的缺点必须等录音结束才能开始处理。流式输入模式我们要实现的功能一边录音一边发送数据服务器实时处理收到的音频片段边说话边看到转写结果延迟可以控制在几百毫秒内想象一下这些场景实时语音助手你说完一句话它马上就能理解并回应在线会议字幕参会者说话的同时字幕就显示出来了语音笔记应用你边说它边记说完笔记也整理好了客服系统客户说话时系统实时转写并分析意图这些场景都需要“实时”处理不能等录音结束。2.2 技术挑战从文件处理切换到流式处理技术上需要解决几个问题音频分块如何把连续的语音流切成合适的小块实时处理如何保证每个小块都能快速处理上下文衔接如何让前后音频块的转写结果连贯资源管理如何避免内存泄漏和资源浪费别担心这些问题我们都会一一解决。3. 环境准备与快速部署3.1 获取镜像并启动首先你需要部署Qwen3-ASR-1.7B的基础镜像。如果你已经部署过了可以直接跳到下一节。# 在镜像市场找到这个镜像 镜像名ins-asr-1.7b-v1 适用底座insbase-cuda124-pt250-dual-v7 # 部署后通过SSH连接到实例 ssh root你的实例IP启动服务# 进入容器后执行启动命令 bash /root/start_asr_1.7b.sh等待1-2分钟服务就启动了。你可以通过浏览器访问http://你的实例IP:7860测试基础功能是否正常。3.2 检查API接口基础镜像已经提供了FastAPI后端服务运行在7861端口。我们先测试一下现有的APIimport requests import json # 测试现有的文件上传接口 url http://localhost:7861/asr files {file: open(test.wav, rb)} data {language: zh} response requests.post(url, filesfiles, datadata) print(response.json())如果返回类似下面的结果说明API服务正常{ language: Chinese, text: 你好这是一个测试音频 }3.3 安装额外依赖我们需要安装一些额外的Python包来支持流式处理pip install websockets pydub sounddevicewebsockets用于WebSocket通信实现双向实时数据传输pydub音频处理库方便操作音频数据sounddevice如果需要从麦克风实时采集音频客户端用4. 流式输入的核心设计4.1 整体架构我们先来看看要实现的功能架构客户端浏览器/App ↓ WebSocket连接 服务端FastAPI WebSocket ↓ 实时音频流 音频缓冲区 ↓ 按固定时长切片 ASR处理队列 ↓ 调用Qwen3-ASR模型 实时转写结果 ↓ 通过WebSocket返回 客户端显示4.2 关键技术点音频分块策略固定时长分块每500ms处理一次VAD语音活动检测分块只在有声音时处理重叠分块前后块有重叠避免切分单词实时处理优化异步处理使用FastAPI的异步支持批处理优化积累一定量数据再处理结果缓存缓存部分结果减少重复计算上下文管理维护会话状态合并相邻块的转写结果处理说话人停顿和继续5. 实现流式ASR服务5.1 创建WebSocket端点首先我们在FastAPI中添加WebSocket支持。创建一个新的文件stream_asr.py# stream_asr.py import asyncio import json import numpy as np from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.middleware.cors import CORSMiddleware import torch import torchaudio from typing import List, Dict, Any import logging # 设置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) app FastAPI(titleQwen3-ASR Stream API) # 允许跨域 app.add_middleware( CORSMiddleware, allow_origins[*], allow_credentialsTrue, allow_methods[*], allow_headers[*], ) # 音频缓冲区管理类 class AudioBuffer: def __init__(self, sample_rate16000): self.sample_rate sample_rate self.buffer [] self.max_buffer_size sample_rate * 10 # 最多缓存10秒音频 def add_audio(self, audio_data: np.ndarray): 添加音频数据到缓冲区 self.buffer.extend(audio_data.tolist()) # 限制缓冲区大小 if len(self.buffer) self.max_buffer_size: self.buffer self.buffer[-self.max_buffer_size:] def get_chunk(self, chunk_duration_ms500): 获取指定时长的音频块 chunk_samples int(self.sample_rate * chunk_duration_ms / 1000) if len(self.buffer) chunk_samples: chunk np.array(self.buffer[:chunk_samples], dtypenp.float32) self.buffer self.buffer[chunk_samples:] return chunk return None def clear(self): 清空缓冲区 self.buffer []5.2 集成Qwen3-ASR模型接下来我们需要加载Qwen3-ASR模型。这里要注意官方镜像已经预装了模型我们直接调用即可# 继续在 stream_asr.py 中添加 class StreamASRService: def __init__(self): self.model None self.processor None self.is_loaded False async def load_model(self): 异步加载模型 if self.is_loaded: return try: # 导入qwen-asr from qwen_asr import QwenASR logger.info(开始加载Qwen3-ASR模型...) # 模型路径根据实际镜像中的路径调整 model_path /root/.cache/modelscope/hub/qwen/Qwen3-ASR-1.7B # 初始化模型 self.model QwenASR( model_pathmodel_path, devicecuda if torch.cuda.is_available() else cpu ) self.is_loaded True logger.info(模型加载完成) except Exception as e: logger.error(f模型加载失败: {e}) raise async def transcribe_stream(self, audio_chunk: np.ndarray, language: str auto): 转写音频块 if not self.is_loaded: await self.load_model() try: # 确保音频是单声道、16kHz if len(audio_chunk.shape) 1: audio_chunk audio_chunk.mean(axis0) # 转换为torch tensor audio_tensor torch.FloatTensor(audio_chunk).unsqueeze(0) # 调用模型进行转写 result self.model.transcribe( audioaudio_tensor, languagelanguage ) return { text: result[text], language: result.get(language, language), confidence: result.get(confidence, 1.0) } except Exception as e: logger.error(f转写失败: {e}) return {text: , language: language, error: str(e)} # 创建全局服务实例 asr_service StreamASRService()5.3 实现WebSocket处理逻辑现在实现核心的WebSocket处理# 继续在 stream_asr.py 中添加 app.websocket(/ws/asr) async def websocket_asr(websocket: WebSocket): WebSocket端点处理实时音频流 await websocket.accept() # 每个连接独立的缓冲区 audio_buffer AudioBuffer(sample_rate16000) language auto # 默认自动检测语言 try: # 接收初始配置 config_data await websocket.receive_json() language config_data.get(language, auto) logger.info(f新连接建立语言设置: {language}) # 预加载模型 await asr_service.load_model() while True: # 接收音频数据Base64编码或二进制 message await websocket.receive() if bytes in message: # 二进制音频数据 audio_bytes message[bytes] # 将字节数据转换为numpy数组 # 这里假设客户端发送的是16位PCM数据 audio_data np.frombuffer(audio_bytes, dtypenp.int16).astype(np.float32) / 32768.0 # 添加到缓冲区 audio_buffer.add_audio(audio_data) # 处理音频块每500ms处理一次 while True: chunk audio_buffer.get_chunk(chunk_duration_ms500) if chunk is None: break # 异步转写 result await asr_service.transcribe_stream(chunk, language) # 发送结果给客户端 await websocket.send_json(result) elif text in message: # 文本消息可能是控制命令 text_msg message[text] if text_msg reset: audio_buffer.clear() await websocket.send_json({status: buffer_cleared}) elif text_msg.startswith(language:): language text_msg.split(:)[1] await websocket.send_json({status: flanguage_changed_to_{language}}) except WebSocketDisconnect: logger.info(客户端断开连接) except Exception as e: logger.error(fWebSocket处理错误: {e}) await websocket.close(code1011)5.4 添加辅助API端点为了方便测试和管理我们再添加一些REST API端点# 继续在 stream_asr.py 中添加 app.get(/health) async def health_check(): 健康检查 return { status: healthy, model_loaded: asr_service.is_loaded, service: qwen3-asr-stream } app.post(/stream/transcribe) async def transcribe_audio_file(file: UploadFile, language: str auto): 文件流式转写兼容原有接口 # 读取音频文件 audio_bytes await file.read() # 使用pydub处理音频 from pydub import AudioSegment import io # 将字节数据转换为AudioSegment audio AudioSegment.from_file(io.BytesIO(audio_bytes)) # 转换为单声道、16kHz audio audio.set_channels(1).set_frame_rate(16000) # 转换为numpy数组 samples np.array(audio.get_array_of_samples()).astype(np.float32) / 32768.0 # 分块处理 chunk_size 16000 // 2 # 500ms的样本数 results [] for i in range(0, len(samples), chunk_size): chunk samples[i:i chunk_size] if len(chunk) chunk_size // 2: # 最后一块太小就跳过 break result await asr_service.transcribe_stream(chunk, language) results.append(result) # 合并结果 full_text .join([r[text] for r in results if r[text]]) return { language: results[0][language] if results else language, text: full_text, chunks: len(results) } if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port7862)6. 客户端实现示例服务端准备好了我们还需要一个客户端来测试。这里提供一个简单的Web客户端示例!-- stream_client.html -- !DOCTYPE html html head titleQwen3-ASR 流式测试客户端/title style body { font-family: Arial, sans-serif; max-width: 800px; margin: 0 auto; padding: 20px; } .container { display: flex; flex-direction: column; gap: 20px; } .control-panel { display: flex; gap: 10px; align-items: center; } button { padding: 10px 20px; font-size: 16px; cursor: pointer; } #startBtn { background-color: #4CAF50; color: white; border: none; } #stopBtn { background-color: #f44336; color: white; border: none; } #result { border: 1px solid #ddd; padding: 20px; min-height: 200px; font-size: 18px; line-height: 1.6; white-space: pre-wrap; background-color: #f9f9f9; } .status { color: #666; font-style: italic; } .transcript { margin: 10px 0; padding: 10px; background: white; border-left: 4px solid #4CAF50; } /style /head body div classcontainer h1Qwen3-ASR 流式语音识别测试/h1 div classcontrol-panel select idlanguageSelect option valueauto自动检测/option option valuezh中文/option option valueen英文/option option valueja日语/option option valueko韩语/option /select button idstartBtn开始录音/button button idstopBtn disabled停止录音/button button idclearBtn清空结果/button div classstatus idstatus准备就绪/div /div div h3实时转写结果/h3 div idresult等待语音输入.../div /div div h3识别日志/h3 div idlog/div /div /div script let mediaRecorder null; let audioChunks []; let ws null; let isRecording false; // WebSocket连接 function connectWebSocket() { const language document.getElementById(languageSelect).value; const wsUrl ws://${window.location.hostname}:7862/ws/asr; ws new WebSocket(wsUrl); ws.onopen () { console.log(WebSocket连接已建立); ws.send(JSON.stringify({ language })); updateStatus(已连接可以开始录音); }; ws.onmessage (event) { const data JSON.parse(event.data); console.log(收到转写结果:, data); if (data.text data.text.trim()) { addTranscript(data.text, data.language); } if (data.error) { addLog(错误: ${data.error}); } }; ws.onerror (error) { console.error(WebSocket错误:, error); updateStatus(连接错误); }; ws.onclose () { console.log(WebSocket连接已关闭); updateStatus(连接已断开); }; } // 开始录音 async function startRecording() { try { const stream await navigator.mediaDevices.getUserMedia({ audio: true }); mediaRecorder new MediaRecorder(stream, { mimeType: audio/webm;codecsopus }); audioChunks []; mediaRecorder.ondataavailable (event) { if (event.data.size 0) { audioChunks.push(event.data); // 将音频数据发送到服务器 if (ws ws.readyState WebSocket.OPEN) { // 这里需要将webm转换为wav格式 // 简化处理直接发送原始数据实际项目中需要转换 ws.send(event.data); } } }; // 每500ms发送一次数据 mediaRecorder.start(500); isRecording true; updateStatus(录音中...); document.getElementById(startBtn).disabled true; document.getElementById(stopBtn).disabled false; } catch (error) { console.error(录音失败:, error); updateStatus(录音失败: error.message); } } // 停止录音 function stopRecording() { if (mediaRecorder isRecording) { mediaRecorder.stop(); mediaRecorder.stream.getTracks().forEach(track track.stop()); isRecording false; updateStatus(录音已停止); document.getElementById(startBtn).disabled false; document.getElementById(stopBtn).disabled true; } } // 添加转写结果 function addTranscript(text, language) { const resultDiv document.getElementById(result); const transcriptDiv document.createElement(div); transcriptDiv.className transcript; transcriptDiv.innerHTML strong[${language}]/strong ${text}; resultDiv.appendChild(transcriptDiv); resultDiv.scrollTop resultDiv.scrollHeight; // 添加到日志 addLog(识别到 ${language}: ${text}); } // 添加日志 function addLog(message) { const logDiv document.getElementById(log); const logEntry document.createElement(div); logEntry.textContent [${new Date().toLocaleTimeString()}] ${message}; logDiv.appendChild(logEntry); logDiv.scrollTop logDiv.scrollHeight; } // 更新状态 function updateStatus(message) { document.getElementById(status).textContent message; } // 清空结果 function clearResults() { document.getElementById(result).innerHTML 等待语音输入...; document.getElementById(log).innerHTML ; } // 事件监听 document.getElementById(startBtn).addEventListener(click, () { if (!ws || ws.readyState ! WebSocket.OPEN) { connectWebSocket(); } startRecording(); }); document.getElementById(stopBtn).addEventListener(click, stopRecording); document.getElementById(clearBtn).addEventListener(click, clearResults); // 页面加载时初始化 window.addEventListener(load, () { updateStatus(点击开始录音按钮进行测试); }); /script /body /html7. 部署与测试7.1 启动流式ASR服务将上面的代码保存后启动新的服务# 在Qwen3-ASR实例中 cd /root python stream_asr.py服务会运行在7862端口。你可以同时保留原有的7861端口服务两者互不干扰。7.2 配置Nginx反向代理可选如果你希望通过同一个域名访问可以配置Nginx# /etc/nginx/sites-available/stream_asr server { listen 80; server_name your-domain.com; location / { proxy_pass http://localhost:7860; # Gradio界面 proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; } location /api/ { proxy_pass http://localhost:7861/; # 原有FastAPI proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; } location /stream/ { proxy_pass http://localhost:7862/; # 新的流式服务 proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; proxy_set_header Host $host; } }7.3 测试流式功能现在可以测试我们的流式ASR服务了测试WebSocket连接import asyncio import websockets import json import numpy as np async def test_stream(): uri ws://localhost:7862/ws/asr async with websockets.connect(uri) as websocket: # 发送配置 await websocket.send(json.dumps({language: zh})) # 模拟发送音频数据 for i in range(10): # 生成测试音频1秒的静音 audio_data np.zeros(16000, dtypenp.float32) audio_bytes audio_data.tobytes() await websocket.send(audio_bytes) # 接收结果 response await websocket.recv() result json.loads(response) print(fChunk {i1}: {result}) await asyncio.sleep(1) asyncio.run(test_stream())测试文件流式转写import requests url http://localhost:7862/stream/transcribe files {file: open(test.wav, rb)} data {language: zh} response requests.post(url, filesfiles, datadata) print(response.json())8. 性能优化与生产建议8.1 性能优化技巧流式ASR对性能要求比较高这里分享几个优化建议1. 音频预处理优化async def optimize_audio_processing(audio_chunk): 优化音频处理流程 # 使用GPU加速的音频处理 if torch.cuda.is_available(): audio_tensor audio_tensor.cuda() # 批量处理积累多个chunk一起处理 if len(self.pending_chunks) self.batch_size: self.pending_chunks.append(audio_chunk) return None # 合并处理 batch_audio torch.cat(self.pending_chunks, dim0) results await self.batch_transcribe(batch_audio) self.pending_chunks [] return results2. 连接管理优化class ConnectionManager: 管理WebSocket连接 def __init__(self): self.active_connections: List[WebSocket] [] async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) def disconnect(self, websocket: WebSocket): self.active_connections.remove(websocket) async def broadcast(self, message: str): 广播消息给所有连接 for connection in self.active_connections: try: await connection.send_text(message) except: self.disconnect(connection)3. 内存管理import gc class MemoryAwareASR: 内存感知的ASR服务 def __init__(self, max_memory_mb1024): self.max_memory max_memory_mb * 1024 * 1024 async def check_memory(self): 检查内存使用情况 import psutil process psutil.Process() memory_usage process.memory_info().rss if memory_usage self.max_memory * 0.8: # 使用超过80% logger.warning(内存使用过高触发垃圾回收) gc.collect() torch.cuda.empty_cache() if torch.cuda.is_available() else None return False return True8.2 生产环境部署建议1. 使用Docker容器化FROM python:3.11-slim WORKDIR /app # 安装依赖 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制代码 COPY . . # 暴露端口 EXPOSE 7862 # 启动命令 CMD [python, stream_asr.py]2. 使用Supervisor管理进程[program:stream_asr] command/usr/local/bin/python /app/stream_asr.py directory/app autostarttrue autorestarttrue stderr_logfile/var/log/stream_asr.err.log stdout_logfile/var/log/stream_asr.out.log3. 监控与日志import logging from prometheus_client import Counter, Histogram, start_http_server # 定义监控指标 REQUEST_COUNT Counter(asr_requests_total, Total ASR requests) REQUEST_LATENCY Histogram(asr_request_latency_seconds, ASR request latency) app.websocket(/ws/asr) async def websocket_asr(websocket: WebSocket): REQUEST_COUNT.inc() with REQUEST_LATENCY.time(): # 处理逻辑 pass9. 常见问题与解决方案9.1 音频同步问题问题客户端发送的音频和服务端处理的速度不一致导致延迟累积。解决方案class AdaptiveBuffer: 自适应缓冲区 def __init__(self, target_latency_ms500): self.target_latency target_latency_ms self.buffer [] self.processing_times [] def adjust_chunk_size(self): 根据处理时间调整chunk大小 if len(self.processing_times) 10: return self.target_latency avg_time sum(self.processing_times[-10:]) / 10 # 如果平均处理时间超过目标延迟减小chunk大小 if avg_time self.target_latency * 1.2: return max(100, self.target_latency * 0.8) # 最小100ms elif avg_time self.target_latency * 0.8: return min(2000, self.target_latency * 1.2) # 最大2000ms return self.target_latency9.2 网络中断处理问题网络不稳定导致连接中断需要重连机制。解决方案// 客户端重连逻辑 class WebSocketClient { constructor(url) { this.url url; this.ws null; this.reconnectAttempts 0; this.maxReconnectAttempts 5; this.connect(); } connect() { this.ws new WebSocket(this.url); this.ws.onopen () { console.log(连接成功); this.reconnectAttempts 0; }; this.ws.onclose () { console.log(连接断开尝试重连...); this.reconnect(); }; this.ws.onerror (error) { console.error(连接错误:, error); }; } reconnect() { if (this.reconnectAttempts this.maxReconnectAttempts) { this.reconnectAttempts; const delay Math.min(1000 * Math.pow(2, this.reconnectAttempts), 30000); setTimeout(() { console.log(第${this.reconnectAttempts}次重连...); this.connect(); }, delay); } } }9.3 多语言切换延迟问题切换语言时模型需要重新加载导致延迟。解决方案class MultiLanguageASR: 多语言ASR管理器 def __init__(self): self.models {} # language - model self.current_language None async def get_model(self, language): 获取指定语言的模型 if language not in self.models: # 异步加载新语言模型 model await self.load_language_model(language) self.models[language] model return self.models[language] async def switch_language(self, new_language): 切换语言 if new_language ! self.current_language: model await self.get_model(new_language) self.current_model model self.current_language new_language10. 总结通过这篇文章我们完成了一个完整的Qwen3-ASR-1.7B流式输入适配开发。从基础的文件上传模式升级到了支持实时语音流的WebSocket服务。10.1 关键收获理解了流式处理的必要性实时语音识别在很多场景下是刚需不能等到录音结束再处理。掌握了核心技术点WebSocket实时通信音频流的分块处理异步模型调用上下文状态管理实现了完整解决方案服务端FastAPI WebSocket Qwen3-ASR客户端Web Audio API 实时显示部署Docker 监控 日志学到了优化技巧自适应缓冲区管理连接重连机制内存和性能监控10.2 下一步建议如果你想让这个系统更完善可以考虑添加说话人分离识别不同说话人的声音实现实时翻译识别后立即翻译成其他语言集成语音唤醒只有听到特定关键词才开始识别添加情感分析识别说话人的情绪状态优化移动端体验开发专门的移动端App10.3 实际应用场景这个流式ASR系统可以用于在线会议系统实时生成会议字幕语音助手实现真正的实时对话教育平台实时语音评测和反馈客服系统实时分析客户情绪和意图无障碍应用为听障人士提供实时字幕最重要的是整个系统完全可以在本地离线运行数据安全有保障响应速度也很快。Qwen3-ASR-1.7B的识别准确率在多语言场景下表现不错加上流式处理能力完全可以满足大多数实时语音识别的需求。希望这个教程对你有帮助。如果你在实现过程中遇到问题或者有更好的优化建议欢迎交流讨论。记住技术是为解决问题服务的选择最适合你场景的方案才是最重要的。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
Qwen3-ASR-1.7B实战教程：流式输入适配开发（基于FastAPI）
Qwen3-ASR-1.7B实战教程流式输入适配开发基于FastAPI1. 引言如果你正在寻找一个能离线运行、支持多语言、识别速度还很快的语音转文字工具那么Qwen3-ASR-1.7B绝对值得你花时间了解一下。这个模型来自阿里通义千问团队有17亿参数支持中文、英文、日语、韩语、粤语等多种语言还能自动检测语言类型。最吸引人的是它完全可以在本地离线运行不需要连接任何外部服务数据安全有保障。官方提供的镜像已经很好用了通过Web界面就能上传音频文件进行转写。但如果你想让这个模型更“智能”一点比如支持实时语音输入、边说话边转写那就需要做一些开发工作了。今天我就来分享一个实战项目如何基于FastAPI为Qwen3-ASR-1.7B开发流式输入适配功能。我会带你从零开始一步步实现一个能实时处理语音流的服务。2. 为什么需要流式输入2.1 文件上传 vs 流式输入先说说两者的区别这样你就能明白为什么流式输入在某些场景下是刚需。文件上传模式当前镜像的默认方式你需要先录好音频文件上传整个文件到服务器服务器一次性处理完整个文件返回完整的转写结果这种方式适合事后转写比如会议录音整理、采访内容转录。但它有个明显的缺点必须等录音结束才能开始处理。流式输入模式我们要实现的功能一边录音一边发送数据服务器实时处理收到的音频片段边说话边看到转写结果延迟可以控制在几百毫秒内想象一下这些场景实时语音助手你说完一句话它马上就能理解并回应在线会议字幕参会者说话的同时字幕就显示出来了语音笔记应用你边说它边记说完笔记也整理好了客服系统客户说话时系统实时转写并分析意图这些场景都需要“实时”处理不能等录音结束。2.2 技术挑战从文件处理切换到流式处理技术上需要解决几个问题音频分块如何把连续的语音流切成合适的小块实时处理如何保证每个小块都能快速处理上下文衔接如何让前后音频块的转写结果连贯资源管理如何避免内存泄漏和资源浪费别担心这些问题我们都会一一解决。3. 环境准备与快速部署3.1 获取镜像并启动首先你需要部署Qwen3-ASR-1.7B的基础镜像。如果你已经部署过了可以直接跳到下一节。# 在镜像市场找到这个镜像 镜像名ins-asr-1.7b-v1 适用底座insbase-cuda124-pt250-dual-v7 # 部署后通过SSH连接到实例 ssh root你的实例IP启动服务# 进入容器后执行启动命令 bash /root/start_asr_1.7b.sh等待1-2分钟服务就启动了。你可以通过浏览器访问http://你的实例IP:7860测试基础功能是否正常。3.2 检查API接口基础镜像已经提供了FastAPI后端服务运行在7861端口。我们先测试一下现有的APIimport requests import json # 测试现有的文件上传接口 url http://localhost:7861/asr files {file: open(test.wav, rb)} data {language: zh} response requests.post(url, filesfiles, datadata) print(response.json())如果返回类似下面的结果说明API服务正常{ language: Chinese, text: 你好这是一个测试音频 }3.3 安装额外依赖我们需要安装一些额外的Python包来支持流式处理pip install websockets pydub sounddevicewebsockets用于WebSocket通信实现双向实时数据传输pydub音频处理库方便操作音频数据sounddevice如果需要从麦克风实时采集音频客户端用4. 流式输入的核心设计4.1 整体架构我们先来看看要实现的功能架构客户端浏览器/App ↓ WebSocket连接 服务端FastAPI WebSocket ↓ 实时音频流 音频缓冲区 ↓ 按固定时长切片 ASR处理队列 ↓ 调用Qwen3-ASR模型 实时转写结果 ↓ 通过WebSocket返回 客户端显示4.2 关键技术点音频分块策略固定时长分块每500ms处理一次VAD语音活动检测分块只在有声音时处理重叠分块前后块有重叠避免切分单词实时处理优化异步处理使用FastAPI的异步支持批处理优化积累一定量数据再处理结果缓存缓存部分结果减少重复计算上下文管理维护会话状态合并相邻块的转写结果处理说话人停顿和继续5. 实现流式ASR服务5.1 创建WebSocket端点首先我们在FastAPI中添加WebSocket支持。创建一个新的文件stream_asr.py# stream_asr.py import asyncio import json import numpy as np from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.middleware.cors import CORSMiddleware import torch import torchaudio from typing import List, Dict, Any import logging # 设置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) app FastAPI(titleQwen3-ASR Stream API) # 允许跨域 app.add_middleware( CORSMiddleware, allow_origins[*], allow_credentialsTrue, allow_methods[*], allow_headers[*], ) # 音频缓冲区管理类 class AudioBuffer: def __init__(self, sample_rate16000): self.sample_rate sample_rate self.buffer [] self.max_buffer_size sample_rate * 10 # 最多缓存10秒音频 def add_audio(self, audio_data: np.ndarray): 添加音频数据到缓冲区 self.buffer.extend(audio_data.tolist()) # 限制缓冲区大小 if len(self.buffer) self.max_buffer_size: self.buffer self.buffer[-self.max_buffer_size:] def get_chunk(self, chunk_duration_ms500): 获取指定时长的音频块 chunk_samples int(self.sample_rate * chunk_duration_ms / 1000) if len(self.buffer) chunk_samples: chunk np.array(self.buffer[:chunk_samples], dtypenp.float32) self.buffer self.buffer[chunk_samples:] return chunk return None def clear(self): 清空缓冲区 self.buffer []5.2 集成Qwen3-ASR模型接下来我们需要加载Qwen3-ASR模型。这里要注意官方镜像已经预装了模型我们直接调用即可# 继续在 stream_asr.py 中添加 class StreamASRService: def __init__(self): self.model None self.processor None self.is_loaded False async def load_model(self): 异步加载模型 if self.is_loaded: return try: # 导入qwen-asr from qwen_asr import QwenASR logger.info(开始加载Qwen3-ASR模型...) # 模型路径根据实际镜像中的路径调整 model_path /root/.cache/modelscope/hub/qwen/Qwen3-ASR-1.7B # 初始化模型 self.model QwenASR( model_pathmodel_path, devicecuda if torch.cuda.is_available() else cpu ) self.is_loaded True logger.info(模型加载完成) except Exception as e: logger.error(f模型加载失败: {e}) raise async def transcribe_stream(self, audio_chunk: np.ndarray, language: str auto): 转写音频块 if not self.is_loaded: await self.load_model() try: # 确保音频是单声道、16kHz if len(audio_chunk.shape) 1: audio_chunk audio_chunk.mean(axis0) # 转换为torch tensor audio_tensor torch.FloatTensor(audio_chunk).unsqueeze(0) # 调用模型进行转写 result self.model.transcribe( audioaudio_tensor, languagelanguage ) return { text: result[text], language: result.get(language, language), confidence: result.get(confidence, 1.0) } except Exception as e: logger.error(f转写失败: {e}) return {text: , language: language, error: str(e)} # 创建全局服务实例 asr_service StreamASRService()5.3 实现WebSocket处理逻辑现在实现核心的WebSocket处理# 继续在 stream_asr.py 中添加 app.websocket(/ws/asr) async def websocket_asr(websocket: WebSocket): WebSocket端点处理实时音频流 await websocket.accept() # 每个连接独立的缓冲区 audio_buffer AudioBuffer(sample_rate16000) language auto # 默认自动检测语言 try: # 接收初始配置 config_data await websocket.receive_json() language config_data.get(language, auto) logger.info(f新连接建立语言设置: {language}) # 预加载模型 await asr_service.load_model() while True: # 接收音频数据Base64编码或二进制 message await websocket.receive() if bytes in message: # 二进制音频数据 audio_bytes message[bytes] # 将字节数据转换为numpy数组 # 这里假设客户端发送的是16位PCM数据 audio_data np.frombuffer(audio_bytes, dtypenp.int16).astype(np.float32) / 32768.0 # 添加到缓冲区 audio_buffer.add_audio(audio_data) # 处理音频块每500ms处理一次 while True: chunk audio_buffer.get_chunk(chunk_duration_ms500) if chunk is None: break # 异步转写 result await asr_service.transcribe_stream(chunk, language) # 发送结果给客户端 await websocket.send_json(result) elif text in message: # 文本消息可能是控制命令 text_msg message[text] if text_msg reset: audio_buffer.clear() await websocket.send_json({status: buffer_cleared}) elif text_msg.startswith(language:): language text_msg.split(:)[1] await websocket.send_json({status: flanguage_changed_to_{language}}) except WebSocketDisconnect: logger.info(客户端断开连接) except Exception as e: logger.error(fWebSocket处理错误: {e}) await websocket.close(code1011)5.4 添加辅助API端点为了方便测试和管理我们再添加一些REST API端点# 继续在 stream_asr.py 中添加 app.get(/health) async def health_check(): 健康检查 return { status: healthy, model_loaded: asr_service.is_loaded, service: qwen3-asr-stream } app.post(/stream/transcribe) async def transcribe_audio_file(file: UploadFile, language: str auto): 文件流式转写兼容原有接口 # 读取音频文件 audio_bytes await file.read() # 使用pydub处理音频 from pydub import AudioSegment import io # 将字节数据转换为AudioSegment audio AudioSegment.from_file(io.BytesIO(audio_bytes)) # 转换为单声道、16kHz audio audio.set_channels(1).set_frame_rate(16000) # 转换为numpy数组 samples np.array(audio.get_array_of_samples()).astype(np.float32) / 32768.0 # 分块处理 chunk_size 16000 // 2 # 500ms的样本数 results [] for i in range(0, len(samples), chunk_size): chunk samples[i:i chunk_size] if len(chunk) chunk_size // 2: # 最后一块太小就跳过 break result await asr_service.transcribe_stream(chunk, language) results.append(result) # 合并结果 full_text .join([r[text] for r in results if r[text]]) return { language: results[0][language] if results else language, text: full_text, chunks: len(results) } if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port7862)6. 客户端实现示例服务端准备好了我们还需要一个客户端来测试。这里提供一个简单的Web客户端示例!-- stream_client.html -- !DOCTYPE html html head titleQwen3-ASR 流式测试客户端/title style body { font-family: Arial, sans-serif; max-width: 800px; margin: 0 auto; padding: 20px; } .container { display: flex; flex-direction: column; gap: 20px; } .control-panel { display: flex; gap: 10px; align-items: center; } button { padding: 10px 20px; font-size: 16px; cursor: pointer; } #startBtn { background-color: #4CAF50; color: white; border: none; } #stopBtn { background-color: #f44336; color: white; border: none; } #result { border: 1px solid #ddd; padding: 20px; min-height: 200px; font-size: 18px; line-height: 1.6; white-space: pre-wrap; background-color: #f9f9f9; } .status { color: #666; font-style: italic; } .transcript { margin: 10px 0; padding: 10px; background: white; border-left: 4px solid #4CAF50; } /style /head body div classcontainer h1Qwen3-ASR 流式语音识别测试/h1 div classcontrol-panel select idlanguageSelect option valueauto自动检测/option option valuezh中文/option option valueen英文/option option valueja日语/option option valueko韩语/option /select button idstartBtn开始录音/button button idstopBtn disabled停止录音/button button idclearBtn清空结果/button div classstatus idstatus准备就绪/div /div div h3实时转写结果/h3 div idresult等待语音输入.../div /div div h3识别日志/h3 div idlog/div /div /div script let mediaRecorder null; let audioChunks []; let ws null; let isRecording false; // WebSocket连接 function connectWebSocket() { const language document.getElementById(languageSelect).value; const wsUrl ws://${window.location.hostname}:7862/ws/asr; ws new WebSocket(wsUrl); ws.onopen () { console.log(WebSocket连接已建立); ws.send(JSON.stringify({ language })); updateStatus(已连接可以开始录音); }; ws.onmessage (event) { const data JSON.parse(event.data); console.log(收到转写结果:, data); if (data.text data.text.trim()) { addTranscript(data.text, data.language); } if (data.error) { addLog(错误: ${data.error}); } }; ws.onerror (error) { console.error(WebSocket错误:, error); updateStatus(连接错误); }; ws.onclose () { console.log(WebSocket连接已关闭); updateStatus(连接已断开); }; } // 开始录音 async function startRecording() { try { const stream await navigator.mediaDevices.getUserMedia({ audio: true }); mediaRecorder new MediaRecorder(stream, { mimeType: audio/webm;codecsopus }); audioChunks []; mediaRecorder.ondataavailable (event) { if (event.data.size 0) { audioChunks.push(event.data); // 将音频数据发送到服务器 if (ws ws.readyState WebSocket.OPEN) { // 这里需要将webm转换为wav格式 // 简化处理直接发送原始数据实际项目中需要转换 ws.send(event.data); } } }; // 每500ms发送一次数据 mediaRecorder.start(500); isRecording true; updateStatus(录音中...); document.getElementById(startBtn).disabled true; document.getElementById(stopBtn).disabled false; } catch (error) { console.error(录音失败:, error); updateStatus(录音失败: error.message); } } // 停止录音 function stopRecording() { if (mediaRecorder isRecording) { mediaRecorder.stop(); mediaRecorder.stream.getTracks().forEach(track track.stop()); isRecording false; updateStatus(录音已停止); document.getElementById(startBtn).disabled false; document.getElementById(stopBtn).disabled true; } } // 添加转写结果 function addTranscript(text, language) { const resultDiv document.getElementById(result); const transcriptDiv document.createElement(div); transcriptDiv.className transcript; transcriptDiv.innerHTML strong[${language}]/strong ${text}; resultDiv.appendChild(transcriptDiv); resultDiv.scrollTop resultDiv.scrollHeight; // 添加到日志 addLog(识别到 ${language}: ${text}); } // 添加日志 function addLog(message) { const logDiv document.getElementById(log); const logEntry document.createElement(div); logEntry.textContent [${new Date().toLocaleTimeString()}] ${message}; logDiv.appendChild(logEntry); logDiv.scrollTop logDiv.scrollHeight; } // 更新状态 function updateStatus(message) { document.getElementById(status).textContent message; } // 清空结果 function clearResults() { document.getElementById(result).innerHTML 等待语音输入...; document.getElementById(log).innerHTML ; } // 事件监听 document.getElementById(startBtn).addEventListener(click, () { if (!ws || ws.readyState ! WebSocket.OPEN) { connectWebSocket(); } startRecording(); }); document.getElementById(stopBtn).addEventListener(click, stopRecording); document.getElementById(clearBtn).addEventListener(click, clearResults); // 页面加载时初始化 window.addEventListener(load, () { updateStatus(点击开始录音按钮进行测试); }); /script /body /html7. 部署与测试7.1 启动流式ASR服务将上面的代码保存后启动新的服务# 在Qwen3-ASR实例中 cd /root python stream_asr.py服务会运行在7862端口。你可以同时保留原有的7861端口服务两者互不干扰。7.2 配置Nginx反向代理可选如果你希望通过同一个域名访问可以配置Nginx# /etc/nginx/sites-available/stream_asr server { listen 80; server_name your-domain.com; location / { proxy_pass http://localhost:7860; # Gradio界面 proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; } location /api/ { proxy_pass http://localhost:7861/; # 原有FastAPI proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; } location /stream/ { proxy_pass http://localhost:7862/; # 新的流式服务 proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; proxy_set_header Host $host; } }7.3 测试流式功能现在可以测试我们的流式ASR服务了测试WebSocket连接import asyncio import websockets import json import numpy as np async def test_stream(): uri ws://localhost:7862/ws/asr async with websockets.connect(uri) as websocket: # 发送配置 await websocket.send(json.dumps({language: zh})) # 模拟发送音频数据 for i in range(10): # 生成测试音频1秒的静音 audio_data np.zeros(16000, dtypenp.float32) audio_bytes audio_data.tobytes() await websocket.send(audio_bytes) # 接收结果 response await websocket.recv() result json.loads(response) print(fChunk {i1}: {result}) await asyncio.sleep(1) asyncio.run(test_stream())测试文件流式转写import requests url http://localhost:7862/stream/transcribe files {file: open(test.wav, rb)} data {language: zh} response requests.post(url, filesfiles, datadata) print(response.json())8. 性能优化与生产建议8.1 性能优化技巧流式ASR对性能要求比较高这里分享几个优化建议1. 音频预处理优化async def optimize_audio_processing(audio_chunk): 优化音频处理流程 # 使用GPU加速的音频处理 if torch.cuda.is_available(): audio_tensor audio_tensor.cuda() # 批量处理积累多个chunk一起处理 if len(self.pending_chunks) self.batch_size: self.pending_chunks.append(audio_chunk) return None # 合并处理 batch_audio torch.cat(self.pending_chunks, dim0) results await self.batch_transcribe(batch_audio) self.pending_chunks [] return results2. 连接管理优化class ConnectionManager: 管理WebSocket连接 def __init__(self): self.active_connections: List[WebSocket] [] async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) def disconnect(self, websocket: WebSocket): self.active_connections.remove(websocket) async def broadcast(self, message: str): 广播消息给所有连接 for connection in self.active_connections: try: await connection.send_text(message) except: self.disconnect(connection)3. 内存管理import gc class MemoryAwareASR: 内存感知的ASR服务 def __init__(self, max_memory_mb1024): self.max_memory max_memory_mb * 1024 * 1024 async def check_memory(self): 检查内存使用情况 import psutil process psutil.Process() memory_usage process.memory_info().rss if memory_usage self.max_memory * 0.8: # 使用超过80% logger.warning(内存使用过高触发垃圾回收) gc.collect() torch.cuda.empty_cache() if torch.cuda.is_available() else None return False return True8.2 生产环境部署建议1. 使用Docker容器化FROM python:3.11-slim WORKDIR /app # 安装依赖 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制代码 COPY . . # 暴露端口 EXPOSE 7862 # 启动命令 CMD [python, stream_asr.py]2. 使用Supervisor管理进程[program:stream_asr] command/usr/local/bin/python /app/stream_asr.py directory/app autostarttrue autorestarttrue stderr_logfile/var/log/stream_asr.err.log stdout_logfile/var/log/stream_asr.out.log3. 监控与日志import logging from prometheus_client import Counter, Histogram, start_http_server # 定义监控指标 REQUEST_COUNT Counter(asr_requests_total, Total ASR requests) REQUEST_LATENCY Histogram(asr_request_latency_seconds, ASR request latency) app.websocket(/ws/asr) async def websocket_asr(websocket: WebSocket): REQUEST_COUNT.inc() with REQUEST_LATENCY.time(): # 处理逻辑 pass9. 常见问题与解决方案9.1 音频同步问题问题客户端发送的音频和服务端处理的速度不一致导致延迟累积。解决方案class AdaptiveBuffer: 自适应缓冲区 def __init__(self, target_latency_ms500): self.target_latency target_latency_ms self.buffer [] self.processing_times [] def adjust_chunk_size(self): 根据处理时间调整chunk大小 if len(self.processing_times) 10: return self.target_latency avg_time sum(self.processing_times[-10:]) / 10 # 如果平均处理时间超过目标延迟减小chunk大小 if avg_time self.target_latency * 1.2: return max(100, self.target_latency * 0.8) # 最小100ms elif avg_time self.target_latency * 0.8: return min(2000, self.target_latency * 1.2) # 最大2000ms return self.target_latency9.2 网络中断处理问题网络不稳定导致连接中断需要重连机制。解决方案// 客户端重连逻辑 class WebSocketClient { constructor(url) { this.url url; this.ws null; this.reconnectAttempts 0; this.maxReconnectAttempts 5; this.connect(); } connect() { this.ws new WebSocket(this.url); this.ws.onopen () { console.log(连接成功); this.reconnectAttempts 0; }; this.ws.onclose () { console.log(连接断开尝试重连...); this.reconnect(); }; this.ws.onerror (error) { console.error(连接错误:, error); }; } reconnect() { if (this.reconnectAttempts this.maxReconnectAttempts) { this.reconnectAttempts; const delay Math.min(1000 * Math.pow(2, this.reconnectAttempts), 30000); setTimeout(() { console.log(第${this.reconnectAttempts}次重连...); this.connect(); }, delay); } } }9.3 多语言切换延迟问题切换语言时模型需要重新加载导致延迟。解决方案class MultiLanguageASR: 多语言ASR管理器 def __init__(self): self.models {} # language - model self.current_language None async def get_model(self, language): 获取指定语言的模型 if language not in self.models: # 异步加载新语言模型 model await self.load_language_model(language) self.models[language] model return self.models[language] async def switch_language(self, new_language): 切换语言 if new_language ! self.current_language: model await self.get_model(new_language) self.current_model model self.current_language new_language10. 总结通过这篇文章我们完成了一个完整的Qwen3-ASR-1.7B流式输入适配开发。从基础的文件上传模式升级到了支持实时语音流的WebSocket服务。10.1 关键收获理解了流式处理的必要性实时语音识别在很多场景下是刚需不能等到录音结束再处理。掌握了核心技术点WebSocket实时通信音频流的分块处理异步模型调用上下文状态管理实现了完整解决方案服务端FastAPI WebSocket Qwen3-ASR客户端Web Audio API 实时显示部署Docker 监控 日志学到了优化技巧自适应缓冲区管理连接重连机制内存和性能监控10.2 下一步建议如果你想让这个系统更完善可以考虑添加说话人分离识别不同说话人的声音实现实时翻译识别后立即翻译成其他语言集成语音唤醒只有听到特定关键词才开始识别添加情感分析识别说话人的情绪状态优化移动端体验开发专门的移动端App10.3 实际应用场景这个流式ASR系统可以用于在线会议系统实时生成会议字幕语音助手实现真正的实时对话教育平台实时语音评测和反馈客服系统实时分析客户情绪和意图无障碍应用为听障人士提供实时字幕最重要的是整个系统完全可以在本地离线运行数据安全有保障响应速度也很快。Qwen3-ASR-1.7B的识别准确率在多语言场景下表现不错加上流式处理能力完全可以满足大多数实时语音识别的需求。希望这个教程对你有帮助。如果你在实现过程中遇到问题或者有更好的优化建议欢迎交流讨论。记住技术是为解决问题服务的选择最适合你场景的方案才是最重要的。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
