""" STT服务模块 - 基于 DashScope OmniRealtimeConversation API 参考官方demo实现，提供流式语音识别功能 """ import asyncio import time import logging import traceback from typing import Optional, Callable, Awaitable import dashscope from dashscope.audio.qwen_omni import OmniRealtimeConversation, OmniRealtimeCallback from dashscope.audio.qwen_omni.omni_realtime import MultiModality, TranscriptionParams from config import settings, get_dashscope_ws_url logger = logging.getLogger(__name__) # 🔥 配置 DashScope SDK 的日志级别，避免空错误日志 dashscope_logger = logging.getLogger('dashscope') dashscope_logger.setLevel(logging.WARNING) # 只记录 WARNING 及以上级别 class STTCallback(OmniRealtimeCallback): """STT识别回调处理器（参考官方demo实现）""" def __init__( self, request_id: str, on_text_update: Optional[Callable[[str, bool], Awaitable[None]]] = None, on_error: Optional[Callable[[str], Awaitable[None]]] = None, event_loop: Optional[asyncio.AbstractEventLoop] = None ): """ 初始化STT回调处理器 Args: request_id: 请求ID（用于日志标识） on_text_update: 文本更新回调 (text, is_final) -> None on_error: 错误回调 (error_msg) -> None event_loop: 事件循环（用于线程安全的异步调用） """ self.request_id = request_id self.on_text_update = on_text_update self.on_error = on_error self.event_loop = event_loop # 识别结果存储 self.current_text = "" self.is_final = False self.final_event = asyncio.Event() if event_loop else None # 时间统计 self.last_update_time = 0.0 self.first_text_time = None # 首个文本生成时间 self.session_id: Optional[str] = None # 语音检测统计 self.speech_detected = False # 是否检测到语音 self.speech_start_count = 0 # 语音开始次数 self.speech_stop_count = 0 # 语音停止次数 # 事件接收统计（用于诊断） self.event_count = 0 # 收到的事件总数 self.received_events = [] # 收到的事件类型列表（最多保存20个） # 事件处理器映射（参考官方demo） self.handlers = { 'session.created': self._handle_session_created, 'conversation.item.input_audio_transcription.completed': self._handle_final_text, 'conversation.item.input_audio_transcription.text': self._handle_stash_text, 'input_audio_buffer.speech_started': self._handle_speech_start, 'input_audio_buffer.speech_stopped': self._handle_speech_stop, 'error': self._handle_error, } def on_open(self): """连接打开回调""" logger.debug( f"[STT调试] request_id={self.request_id}, " f"WebSocket连接已打开" ) def on_close(self, code: int, msg: str): """连接关闭回调""" logger.debug( f"[STT调试] request_id={self.request_id}, " f"WebSocket连接已关闭, " f"关闭码={code}, " f"关闭消息={msg}" ) # 非正常关闭（1000是正常关闭码，None是主动关闭） if code is not None and code != 1000 and self.on_error: error_msg = f"STT连接异常关闭: {msg} (code: {code})" logger.error( f"[STT异常] request_id={self.request_id}, " f"错误类型=连接异常关闭, " f"关闭码={code}, " f"错误消息={msg}" ) self._async_call(self.on_error(error_msg)) def on_event(self, response: dict): """事件分发（参考官方demo）""" try: event_type = response.get('type') handler = self.handlers.get(event_type) # 🔥 更新事件接收统计 self.event_count += 1 if len(self.received_events) < 20: self.received_events.append(event_type) # 🔥 打印STT接口的原始返回日志（完整JSON） import json try: response_json = json.dumps(response, ensure_ascii=False, indent=2) logger.info( f"[STT原始返回] request_id={self.request_id}, " f"事件#{self.event_count}, " f"事件类型={event_type}\n" f"完整响应数据:\n{response_json}" ) except Exception as json_error: logger.warning( f"[STT原始返回] request_id={self.request_id}, " f"事件#{self.event_count}, " f"事件类型={event_type}, " f"JSON序列化失败: {json_error}, " f"原始响应={response}" ) # 调试日志：记录所有事件（高频事件简化输出） if event_type in ['input_audio_buffer.speech_started', 'input_audio_buffer.speech_stopped']: logger.debug( f"[STT调试] request_id={self.request_id}, " f"事件类型={event_type}" ) else: logger.debug( f"[STT调试] request_id={self.request_id}, " f"事件类型={event_type}, " f"事件数据={response}" ) if handler: handler(response) else: # 记录未处理的事件（排除常见的已知事件） known_events = [ 'session.created', 'session.updated', 'input_audio_buffer.committed', 'conversation.item.created' # 会话项创建事件 ] if event_type not in known_events: pass except Exception as e: logger.error( f"[STT异常] request_id={self.request_id}, " f"错误类型=事件处理异常, " f"异常类型={type(e).__name__}, " f"错误消息={str(e)}, " f"事件类型={response.get('type', 'unknown')}, " f"堆栈跟踪:\n{traceback.format_exc()}" ) # ========== 事件处理器（参考官方demo） ========== def _handle_session_created(self, response: dict): """处理会话创建事件""" session_id = response.get('session', {}).get('id', '') self.session_id = session_id logger.debug( f"[STT调试] request_id={self.request_id}, " f"会话已创建, session_id={session_id}" ) def _handle_final_text(self, response: dict): """处理最终识别结果（参考官方demo）""" text = response.get('transcript', '') logger.debug( f"[STT调试] request_id={self.request_id}, " f"收到最终识别结果, 文本长度={len(text)}, " f"文本内容={text[:50]}{'...' if len(text) > 50 else ''}" ) # 更新状态 self.current_text = text self.is_final = True self.last_update_time = time.time() # 触发最终结果事件 if self.final_event and self.event_loop: try: self.event_loop.call_soon_threadsafe(self.final_event.set) logger.debug( f"[STT调试] request_id={self.request_id}, " f"已触发最终结果事件" ) except Exception as e: logger.error( f"[STT异常] request_id={self.request_id}, " f"错误类型=事件触发异常, " f"异常类型={type(e).__name__}, " f"错误消息={str(e)}, " f"堆栈跟踪:\n{traceback.format_exc()}" ) # 回调通知 if self.on_text_update: self._async_call(self.on_text_update(text, True)) def _handle_stash_text(self, response: dict): """处理中间识别结果（参考官方demo）""" # 合并 text（已确认部分）和 stash（正在识别部分） confirmed_text = response.get('text', '') stash_text = response.get('stash', '') full_text = confirmed_text + stash_text # 更新状态 if full_text: self.current_text = full_text self.is_final = False current_time = time.time() self.last_update_time = current_time # 记录首个文本生成时间 if self.first_text_time is None and full_text: self.first_text_time = current_time logger.info( f"[STT] request_id={self.request_id}, " f"首个文本生成, 文本={full_text[:30]}{'...' if len(full_text) > 30 else ''}" ) # 调试日志：每5次中间结果记录一次（避免日志过多） if hasattr(self, '_stash_count'): self._stash_count += 1 else: self._stash_count = 1 if self._stash_count % 5 == 0: logger.debug( f"[STT调试] request_id={self.request_id}, " f"中间识别结果 #{self._stash_count}, " f"文本长度={len(full_text)}, " f"文本内容={full_text[:30]}{'...' if len(full_text) > 30 else ''}" ) # 回调通知 if self.on_text_update: self._async_call(self.on_text_update(full_text, False)) def _handle_speech_start(self, response: dict): """处理语音开始事件""" self.speech_detected = True self.speech_start_count += 1 logger.debug( f"[STT调试] request_id={self.request_id}, " f"检测到语音开始, speech_start_count={self.speech_start_count}" ) def _handle_speech_stop(self, response: dict): """处理语音停止事件""" self.speech_stop_count += 1 logger.debug( f"[STT调试] request_id={self.request_id}, " f"检测到语音停止, speech_stop_count={self.speech_stop_count}" ) def _handle_error(self, response: dict): """处理错误事件""" error = response.get('error', {}) error_msg = error.get('message', 'Unknown error') error_code = error.get('code', '') # 详细记录STT API错误 logger.error( f"[STT异常] request_id={self.request_id}, " f"错误类型=STT API错误, " f"错误码={error_code}, " f"错误消息={error_msg}, " f"完整响应={response}" ) # 回调通知 if self.on_error: self._async_call(self.on_error(f"{error_msg} (code: {error_code})")) # ========== 辅助方法 ========== def _async_call(self, coro): """线程安全的异步调用""" if self.event_loop: try: asyncio.run_coroutine_threadsafe(coro, self.event_loop) except Exception as e: logger.error( f"[STT异常] request_id={self.request_id}, " f"错误类型=异步调用异常, " f"异常类型={type(e).__name__}, " f"错误消息={str(e)}, " f"堆栈跟踪:\n{traceback.format_exc()}" ) def get_text(self) -> str: """获取当前识别文本""" return self.current_text def is_final_result(self) -> bool: """是否为最终结果""" return self.is_final class STTSession: """STT识别会话管理器（参考官方demo实现）""" def __init__( self, request_id: str, on_text_update: Optional[Callable[[str, bool], Awaitable[None]]] = None, on_error: Optional[Callable[[str], Awaitable[None]]] = None, event_loop: Optional[asyncio.AbstractEventLoop] = None ): """ 初始化STT会话 Args: request_id: 请求ID on_text_update: 文本更新回调 on_error: 错误回调 event_loop: 事件循环 """ self.request_id = request_id self.callback = STTCallback( request_id=request_id, on_text_update=on_text_update, on_error=on_error, event_loop=event_loop ) self.conversation: Optional[OmniRealtimeConversation] = None self.is_connected = False self.is_closed = False async def connect(self, max_retries: int = 2) -> bool: """ 连接STT服务并初始化会话（带重试机制） 🔥 每次请求都创建新的 WebSocket 会话，确保可以随时建立请求 不会因为连接复用而阻塞，每个 request_id 都有独立的会话 Args: max_retries: 最大重试次数（默认2次，总共尝试3次） Returns: 是否连接成功 """ last_error = None for attempt in range(max_retries + 1): try: start_time = time.time() # 设置API Key dashscope.api_key = settings.active_dashscope_api_key logger.debug( f"[STT调试] request_id={self.request_id}, " f"连接尝试 {attempt + 1}/{max_retries + 1}, " f"API Key已设置, 使用国际版={settings.use_dashscope_intl}" ) # 🔥 创建新的会话对象（每次请求独立创建，参考官方demo） def create_conversation(): ws_url = get_dashscope_ws_url() logger.debug( f"[STT调试] request_id={self.request_id}, " f"创建会话对象, URL={ws_url}" ) conversation = OmniRealtimeConversation( model=settings.STT_MODEL, url=ws_url, callback=self.callback ) # 注入自身引用（回调中可能需要） self.callback.conversation = conversation logger.debug( f"[STT调试] request_id={self.request_id}, " f"会话对象已创建" ) return conversation # 异步创建会话（每次都是新连接） create_start = time.time() self.conversation = await asyncio.to_thread(create_conversation) create_duration = (time.time() - create_start) * 1000 logger.debug( f"[STT调试] request_id={self.request_id}, " f"会话对象创建完成, 耗时={create_duration:.1f}ms" ) # 连接WebSocket（添加超时保护） logger.debug( f"[STT调试] request_id={self.request_id}, " f"开始连接WebSocket..." ) connect_start = time.time() try: await asyncio.wait_for( asyncio.to_thread(self.conversation.connect), timeout=10.0 # 10秒连接超时 ) connect_duration = (time.time() - connect_start) * 1000 logger.debug( f"[STT调试] request_id={self.request_id}, " f"WebSocket连接成功, 耗时={connect_duration:.1f}ms" ) except asyncio.TimeoutError: raise Exception("STT WebSocket连接超时（10秒）") # 配置转录参数（参考官方demo） # ⚠️ 重要：language参数是必需的！ # 官方支持: 'zh'(中文), 'en'(英文), 'ja'(日文), 'ko'(韩文) # 尝试支持: 'auto'(自动检测) - 需要测试是否有效 stt_language = getattr(settings, 'STT_LANGUAGE', 'zh') transcription_params = TranscriptionParams( language=stt_language, sample_rate=16000, input_audio_format="pcm" # corpus_text="" # 可选：辅助识别的语料 ) logger.debug( f"[STT调试] request_id={self.request_id}, " f"配置转录参数: language={stt_language}, " f"sample_rate=16000, format=pcm" ) # 更新会话配置（禁用VAD，用户通过按钮控制录音） logger.debug( f"[STT调试] request_id={self.request_id}, " f"开始更新会话配置..." ) config_start = time.time() try: await asyncio.wait_for( asyncio.to_thread( self.conversation.update_session, output_modalities=[MultiModality.TEXT], enable_input_audio_transcription=True, transcription_params=transcription_params, enable_turn_detection=False ), timeout=5.0 # 5秒配置超时 ) config_duration = (time.time() - config_start) * 1000 logger.debug( f"[STT调试] request_id={self.request_id}, " f"会话配置完成, 耗时={config_duration:.1f}ms" ) except asyncio.TimeoutError: raise Exception("STT会话配置超时（5秒）") self.is_connected = True connect_duration = (time.time() - start_time) * 1000 logger.info( f"[STT连接成功] request_id={self.request_id}, " f"尝试次数={attempt + 1}, " f"耗时={connect_duration:.1f}ms" ) return True except Exception as e: last_error = e connect_duration = (time.time() - start_time) * 1000 if 'start_time' in locals() else 0 # 详细记录错误 error_details = { "request_id": self.request_id, "错误类型": "STT连接失败", "异常类型": type(e).__name__, "错误消息": str(e), "尝试次数": f"{attempt + 1}/{max_retries + 1}", "耗时": f"{connect_duration:.1f}ms", "API Key": f"{settings.active_dashscope_api_key[:10]}... (已脱敏)", "使用国际版": settings.use_dashscope_intl, } # 如果是最后一次尝试，记录完整堆栈 if attempt == max_retries: error_details["堆栈跟踪"] = traceback.format_exc() logger.error(f"[STT异常] {error_details}") else: logger.warning(f"[STT连接重试] {error_details}") # 如果不是最后一次尝试，等待后重试 if attempt < max_retries: wait_time = (attempt + 1) * 0.5 # 递增等待：0.5s, 1s logger.info(f"[STT连接重试] request_id={self.request_id}, {wait_time:.1f}秒后重试...") await asyncio.sleep(wait_time) else: # 最后一次尝试失败，通知错误回调 if self.callback.on_error: await self.callback.on_error(f"连接失败（已重试{max_retries}次）: {str(e)}") return False async def send_audio(self, audio_data: bytes) -> bool: """ 发送音频数据（参考官方demo，带超时保护） Args: audio_data: PCM音频数据（原始bytes） Returns: 是否发送成功 """ if not self.is_connected or not self.conversation: logger.warning( f"[STT警告] request_id={self.request_id}, " f"连接状态异常: is_connected={self.is_connected}, " f"conversation={'存在' if self.conversation else '不存在'}" ) return False try: # 🔥 音频数据验证和统计（用于诊断问题） import struct try: # 将PCM字节数据转换为Int16数组 sample_count = len(audio_data) // 2 if sample_count > 0: samples = struct.unpack(f'<{sample_count}h', audio_data) max_amplitude = max(abs(s) for s in samples) if samples else 0 avg_amplitude = sum(abs(s) for s in samples) // len(samples) if samples else 0 zero_count = sum(1 for s in samples if s == 0) zero_ratio = zero_count / len(samples) if samples else 0.0 # 记录音频统计信息（仅第一个chunk和异常情况） if not hasattr(self, '_audio_chunk_count'): self._audio_chunk_count = 0 self._audio_chunk_count += 1 # 第一个chunk或异常情况（全零或振幅过低）时记录详细信息 if self._audio_chunk_count == 1 or max_amplitude < 100 or zero_ratio > 0.9: logger.info( f"[STT音频统计] request_id={self.request_id}, " f"chunk={self._audio_chunk_count}, " f"长度={len(audio_data)} bytes, " f"采样数={sample_count}, " f"最大振幅={max_amplitude}, " f"平均振幅={avg_amplitude}, " f"零值比例={zero_ratio:.2%}, " f"是否静音={'是' if max_amplitude < 100 or zero_ratio > 0.9 else '否'}" ) except Exception as stats_error: logger.warning( f"[STT音频统计] request_id={self.request_id}, " f"统计计算失败: {stats_error}" ) # Base64编码（API要求） import base64 audio_b64 = base64.b64encode(audio_data).decode('ascii') # 使用INFO级别记录音频准备发送（便于排查问题） logger.info( f"[STT] request_id={self.request_id}, " f"准备发送音频: 原始长度={len(audio_data)} bytes, " f"Base64长度={len(audio_b64)} chars" ) # 发送音频帧（参考官方demo，添加超时保护） send_start = time.time() try: await asyncio.wait_for( asyncio.to_thread(self.conversation.append_audio, audio_b64), timeout=5.0 # 5秒发送超时 ) send_duration = (time.time() - send_start) * 1000 # 使用INFO级别记录音频发送成功（便于排查问题） logger.info( f"[STT] request_id={self.request_id}, " f"音频发送成功, 原始长度={len(audio_data)} bytes, " f"耗时={send_duration:.1f}ms" ) except asyncio.TimeoutError: raise Exception("发送音频数据超时（5秒）") return True except Exception as e: # 标记连接可能已断开 error_str = str(e).lower() if 'timeout' in error_str or 'closed' in error_str or 'disconnect' in error_str: self.is_connected = False logger.error( f"[STT异常] request_id={self.request_id}, " f"错误类型=连接可能已断开, " f"异常类型={type(e).__name__}, " f"错误消息={str(e)}, " f"音频数据长度={len(audio_data)} bytes" ) else: logger.error( f"[STT异常] request_id={self.request_id}, " f"错误类型=发送音频失败, " f"异常类型={type(e).__name__}, " f"错误消息={str(e)}, " f"音频数据长度={len(audio_data)} bytes, " f"连接状态=is_connected={self.is_connected}, " f"堆栈跟踪:\n{traceback.format_exc()}" ) if self.callback.on_error: await self.callback.on_error(f"发送音频失败: {e}") return False async def commit_audio(self) -> bool: """ 提交音频以触发识别（禁用VAD时需要手动调用，带超时保护） Returns: 是否提交成功 """ if not self.is_connected or not self.conversation: logger.warning( f"[STT警告] request_id={self.request_id}, " f"连接状态异常，无法提交音频: " f"is_connected={self.is_connected}, " f"conversation={'存在' if self.conversation else '不存在'}" ) return False try: logger.debug( f"[STT调试] request_id={self.request_id}, " f"开始提交音频..." ) # 添加超时保护 commit_start = time.time() try: await asyncio.wait_for( asyncio.to_thread(self.conversation.commit), timeout=5.0 # 5秒提交超时 ) commit_duration = (time.time() - commit_start) * 1000 logger.info( f"[STT] request_id={self.request_id}, " f"音频提交成功, 耗时={commit_duration:.1f}ms" ) except asyncio.TimeoutError: raise Exception("提交音频超时（5秒）") return True except Exception as e: # 标记连接可能已断开 error_str = str(e).lower() if 'timeout' in error_str or 'closed' in error_str or 'disconnect' in error_str: self.is_connected = False logger.error( f"[STT异常] request_id={self.request_id}, " f"错误类型=提交音频失败, " f"异常类型={type(e).__name__}, " f"错误消息={str(e)}, " f"连接状态=is_connected={self.is_connected}, " f"堆栈跟踪:\n{traceback.format_exc()}" ) return False async def wait_for_final_result(self, timeout: float = 3.0) -> Optional[str]: """ 等待最终识别结果 Args: timeout: 超时时间（秒） Returns: 识别文本，超时或失败返回None """ logger.debug( f"[STT调试] request_id={self.request_id}, " f"开始等待最终识别结果, 超时时间={timeout}秒" ) if not self.callback.final_event: logger.debug( f"[STT调试] request_id={self.request_id}, " f"未设置事件循环，直接返回当前文本" ) return self.callback.get_text() wait_start = time.time() try: await asyncio.wait_for(self.callback.final_event.wait(), timeout=timeout) wait_duration = (time.time() - wait_start) * 1000 result_text = self.callback.get_text() logger.info( f"[STT] request_id={self.request_id}, " f"收到最终识别结果, 等待耗时={wait_duration:.1f}ms, " f"文本长度={len(result_text)}, " f"文本内容={result_text[:50]}{'...' if len(result_text) > 50 else ''}" ) return result_text except asyncio.TimeoutError: # 超时后，检查是否有中间结果和语音检测状态 wait_duration = (time.time() - wait_start) * 1000 current_text = self.callback.get_text() speech_detected = self.callback.speech_detected speech_start_count = self.callback.speech_start_count speech_stop_count = self.callback.speech_stop_count # 🔥 输出事件接收统计（用于诊断） event_count = self.callback.event_count received_events = self.callback.received_events logger.warning( f"[STT警告] request_id={self.request_id}, " f"错误类型=等待识别结果超时, " f"超时时间={timeout}秒, " f"等待耗时={wait_duration:.1f}ms, " f"当前识别文本={current_text}, " f"语音检测状态=speech_detected={speech_detected}, " f"speech_start_count={speech_start_count}, " f"speech_stop_count={speech_stop_count}, " f"收到事件总数={event_count}, " f"收到的事件类型={received_events}" ) return current_text except Exception as e: logger.error( f"[STT异常] request_id={self.request_id}, " f"错误类型=等待识别结果异常, " f"异常类型={type(e).__name__}, " f"错误消息={str(e)}, " f"超时时间={timeout}秒, " f"堆栈跟踪:\n{traceback.format_exc()}" ) return None async def close(self): """关闭STT会话（参考官方demo）""" if self.is_closed: logger.debug( f"[STT调试] request_id={self.request_id}, " f"会话已关闭，跳过" ) return logger.debug( f"[STT调试] request_id={self.request_id}, " f"开始关闭STT会话, " f"连接状态=is_connected={self.is_connected}, " f"conversation={'存在' if self.conversation else '不存在'}" ) try: if self.conversation: # 异步关闭连接 close_start = time.time() await asyncio.to_thread(self.conversation.close) close_duration = (time.time() - close_start) * 1000 logger.debug( f"[STT调试] request_id={self.request_id}, " f"会话关闭完成, 耗时={close_duration:.1f}ms" ) self.is_connected = False self.is_closed = True except Exception as e: logger.error( f"[STT异常] request_id={self.request_id}, " f"错误类型=关闭会话异常, " f"异常类型={type(e).__name__}, " f"错误消息={str(e)}, " f"连接状态=is_connected={self.is_connected}, is_closed={self.is_closed}, " f"堆栈跟踪:\n{traceback.format_exc()}" ) def get_current_text(self) -> str: """获取当前识别文本""" return self.callback.get_text() def get_last_update_time(self) -> float: """获取最后更新时间""" return self.callback.last_update_time def get_first_text_time(self) -> Optional[float]: """获取首个文本生成时间""" return self.callback.first_text_time class STTService: """STT服务管理器（单例）""" _instance = None def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance def __init__(self): """初始化STT服务""" if not hasattr(self, '_initialized'): self._initialized = True # 设置API Key dashscope.api_key = settings.active_dashscope_api_key # 国际版endpoint配置 if settings.use_dashscope_intl: from config import get_dashscope_domain dashscope_domain = get_dashscope_domain() dashscope.base_http_api_url = f"https://{dashscope_domain}/api/v1" dashscope.base_websocket_api_url = f"wss://{dashscope_domain}/api-ws/v1" async def create_session( self, request_id: str, on_text_update: Optional[Callable[[str, bool], Awaitable[None]]] = None, on_error: Optional[Callable[[str], Awaitable[None]]] = None, event_loop: Optional[asyncio.AbstractEventLoop] = None ) -> STTSession: """ 创建STT识别会话 Args: request_id: 请求ID on_text_update: 文本更新回调 on_error: 错误回调 event_loop: 事件循环 Returns: STT会话对象 """ session = STTSession( request_id=request_id, on_text_update=on_text_update, on_error=on_error, event_loop=event_loop ) # 自动连接 success = await session.connect() return session # 全局STT服务实例 stt_service = STTService()