""" AI服务：集成STT、LLM、TTS模型 """ import base64 import json import asyncio import tempfile import os from typing import AsyncGenerator, Optional import dashscope from openai import OpenAI from config import settings from models import EmotionType # 设置API Key dashscope.api_key = settings.dashscope_api_key class AIService: """AI服务管理器""" def __init__(self): import httpx import os # 临时清除代理环境变量（只在这个进程中） env_backup = {} proxy_vars = ['HTTP_PROXY', 'HTTPS_PROXY', 'http_proxy', 'https_proxy', 'ALL_PROXY', 'all_proxy'] for var in proxy_vars: if var in os.environ: env_backup[var] = os.environ[var] del os.environ[var] # 创建不使用代理的 HTTP 客户端 http_client = httpx.Client( proxies={}, # 明确设置为空字典，禁用所有代理 timeout=60.0, trust_env=False, # 不信任环境变量中的代理设置 ) self.openai_client = OpenAI( api_key=settings.dashscope_api_key, base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", http_client=http_client, # 使用自定义 HTTP 客户端 ) # 恢复环境变量（避免影响其他模块） for var, value in env_backup.items(): os.environ[var] = value # STT现在使用流式识别，在websocket_handler中直接处理 async def generate_response_stream( self, user_text: str ) -> AsyncGenerator[tuple[str, EmotionType], None]: """ 生成对话回复（流式） Args: user_text: 用户输入文本 Yields: (sentence, emotion): 句子片段和情感 """ try: print(f"\n========== LLM 流式开始 ==========") print(f"用户输入: {user_text}") # 构建System Prompt system_prompt = """You are a friendly English learning assistant. Please respond to the user in English to help them practice. Be encouraging and supportive. At the end of your response, on a new line, add ONLY the emotion tag in this exact format: [EMOTION:emotion_name] Available emotions: neutral, happy, sad, excited, angry, surprised, thoughtful, encouraging Example response: Great job! That's the correct pronunciation. Keep practicing and you'll get even better! [EMOTION:encouraging]""" print(f"调用 LLM 流式: {settings.llm_model}") # 调用LLM流式API response = self.openai_client.chat.completions.create( model=settings.llm_model, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_text}, ], temperature=0.7, max_tokens=500, stream=True, # 启用流式 ) buffer = "" full_response = "" sentence_count = 0 for chunk in response: if chunk.choices[0].delta.content: content = chunk.choices[0].delta.content buffer += content full_response += content # 检测句子边界（完整的句子） if self._is_complete_sentence(buffer): sentence = buffer.strip() sentence_count += 1 print(f"LLM 流式输出句子{sentence_count}: {sentence}") yield (sentence, "neutral") # 情感稍后统一提取 buffer = "" # 处理剩余内容和提取情感 emotion: EmotionType = "neutral" # 从完整响应中提取情感 if "[EMOTION:" in full_response: emotion_start = full_response.find("[EMOTION:") emotion_end = full_response.find("]", emotion_start) if emotion_end > emotion_start: emotion_str = full_response[emotion_start + 9:emotion_end].strip().lower() valid_emotions = ["neutral", "happy", "sad", "excited", "angry", "surprised", "thoughtful", "encouraging"] if emotion_str in valid_emotions: emotion = emotion_str # 移除情感标签部分 full_response = full_response[:emotion_start].strip() # 处理buffer中的剩余内容（已输出句子之后的部分） remaining = buffer.strip() if remaining and not remaining.startswith("[EMOTION"): sentence_count += 1 print(f"LLM 流式输出最后{sentence_count}: {remaining}") yield (remaining, emotion) elif not remaining and sentence_count == 0: # 如果一个句子都没输出，输出完整内容 if full_response: print(f"LLM 流式输出完整: {full_response}") yield (full_response, emotion) print(f"LLM 完整响应: {full_response}") print(f"提取情感: {emotion}") print(f"========== LLM 流式成功 ==========\n") except Exception as e: print(f"========== LLM 流式错误 ==========") print(f"错误: {e}") import traceback traceback.print_exc() print(f"========== LLM 流式错误结束 ==========\n") yield ("I'm sorry, I'm having trouble understanding. Could you please repeat that?", "neutral") def _is_complete_sentence(self, text: str) -> bool: """检测是否完整句子""" text = text.strip() # 必须以句子结束符结尾 if not text.endswith(('.', '!', '?', '。', '！', '？')): return False # 排除常见缩写（避免误判） words = text.split() if words: last_word = words[-1] # 常见英文缩写 common_abbrev = ['Mr.', 'Mrs.', 'Ms.', 'Dr.', 'Prof.', 'Sr.', 'Jr.', 'etc.', 'e.g.', 'i.e.', 'vs.'] if last_word in common_abbrev: return False # 至少5个字符（避免"Hi."这种过短的） if len(text) < 5: return False return True async def generate_response( self, user_text: str ) -> tuple[str, EmotionType]: """ 生成对话回复 Args: user_text: 用户输入文本 Returns: (response_text, emotion): 回复文本和情感 """ try: print(f"\n========== LLM 开始 ==========") print(f"用户输入: {user_text}") # 构建System Prompt，要求返回情感标签 system_prompt = """You are a friendly English learning assistant. Please respond to the user in English to help them practice. Be encouraging and supportive. At the end of your response, on a new line, add ONLY the emotion tag in this exact format: [EMOTION:emotion_name] Available emotions: neutral, happy, sad, excited, angry, surprised, thoughtful, encouraging Example response: Great job! That's the correct pronunciation. Keep practicing and you'll get even better! [EMOTION:encouraging]""" print(f"调用 LLM: {settings.llm_model}") # 调用LLM response = self.openai_client.chat.completions.create( model=settings.llm_model, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_text}, ], temperature=0.7, max_tokens=500, ) full_response = response.choices[0].message.content or "" print(f"LLM 原始响应: {full_response}") # 提取情感标签 emotion: EmotionType = "neutral" response_text = full_response # 查找情感标签 if "[EMOTION:" in full_response: emotion_start = full_response.find("[EMOTION:") emotion_end = full_response.find("]", emotion_start) if emotion_end > emotion_start: emotion_str = full_response[emotion_start + 9:emotion_end].strip().lower() # 验证情感类型 valid_emotions = ["neutral", "happy", "sad", "excited", "angry", "surprised", "thoughtful", "encouraging"] if emotion_str in valid_emotions: emotion = emotion_str print(f"提取到情感标签: {emotion}") # 移除情感标签（不发送给TTS） response_text = full_response[:emotion_start].strip() print(f"已移除情感标签，纯文本内容: {response_text}") # 如果没有文本内容，使用默认回复 if not response_text: response_text = "I understand. Please tell me more!" emotion = "neutral" print(f"✅ 将发送给TTS的文本: {response_text}") print(f"✅ 情感: {emotion}") print(f"========== LLM 成功 ==========\n") return (response_text, emotion) except Exception as e: print(f"========== LLM 错误 ==========") print(f"错误: {e}") import traceback traceback.print_exc() print(f"========== LLM 错误结束 ==========\n") return ("I'm sorry, I'm having trouble understanding. Could you please repeat that?", "neutral") async def text_to_speech_stream( self, text: str, emotion: EmotionType = "neutral" ) -> AsyncGenerator[bytes, None]: """ 文本转语音（流式） Args: text: 要合成的文本 emotion: 情感类型 Yields: audio_data: 音频数据块 """ try: print(f"\n========== TTS 开始 ==========") print(f"要合成的文本: {text}") print(f"情感: {emotion}") # 根据情感选择音色或参数 voice = self._get_voice_for_emotion(emotion) print(f"使用音色: {voice}") print(f"TTS模型: {settings.tts_model}") # 在线程池中执行同步调用，避免事件循环冲突 # 添加重试机制以应对网络不稳定 def sync_tts(): max_retries = 3 retry_delay = 1 # 秒 for attempt in range(max_retries): try: return dashscope.audio.qwen_tts.SpeechSynthesizer.call( model=settings.tts_model, text=text, voice=voice, ) except Exception as e: if attempt < max_retries - 1: print(f"TTS调用失败(尝试 {attempt + 1}/{max_retries}): {e}") print(f"等待{retry_delay}秒后重试...") import time time.sleep(retry_delay) retry_delay *= 2 # 指数退避 else: raise print("调用 DashScope TTS API...") # 调用TTS API response = await asyncio.to_thread(sync_tts) print(f"TTS API 响应类型: {type(response)}") # 获取音频数据 # response.output 包含 audio 字典，其中有 data、url、id、expires_at audio_bytes = None if hasattr(response, 'output'): output = response.output print(f"output 类型: {type(output)}") # 打印finish_reason用于调试 if isinstance(output, dict): print(f"finish_reason: {output.get('finish_reason', 'N/A')}") audio_info = output.get('audio', {}) elif hasattr(output, 'finish_reason'): print(f"finish_reason: {output.finish_reason}") audio_info = output.get('audio', {}) if isinstance(output, dict) else getattr(output, 'audio', {}) else: audio_info = {} # 检查是否有URL（新版API返回URL） if isinstance(audio_info, dict) and 'url' in audio_info and audio_info['url']: audio_url = audio_info['url'] print(f"从URL下载音频: {audio_url}") # 下载音频数据 import aiohttp async with aiohttp.ClientSession() as session: async with session.get(audio_url) as resp: if resp.status == 200: audio_bytes = await resp.read() print(f"音频下载成功，大小: {len(audio_bytes)} 字节") else: raise Exception(f"下载音频失败，HTTP状态码: {resp.status}") # 检查是否有直接的数据（旧版API） elif isinstance(audio_info, dict) and 'data' in audio_info and audio_info['data']: audio_data = audio_info['data'] if isinstance(audio_data, bytes): audio_bytes = audio_data print(f"音频数据(bytes): {len(audio_bytes)} 字节") elif isinstance(audio_data, str): audio_bytes = base64.b64decode(audio_data) print(f"音频数据(base64解码): {len(audio_bytes)} 字节") # 如果audio_info直接是bytes或str elif isinstance(audio_info, bytes): audio_bytes = audio_info print(f"音频数据(bytes): {len(audio_bytes)} 字节") elif isinstance(audio_info, str): audio_bytes = base64.b64decode(audio_info) print(f"音频数据(base64解码): {len(audio_bytes)} 字节") if audio_bytes: # 将音频数据分块返回 chunk_size = 3200 # 每块3200字节 chunk_count = (len(audio_bytes) + chunk_size - 1) // chunk_size print(f"将分为 {chunk_count} 个数据块发送") for i in range(0, len(audio_bytes), chunk_size): chunk = audio_bytes[i:i + chunk_size] yield chunk await asyncio.sleep(0) # 让出控制权 print(f"========== TTS 成功 ==========\n") else: error_msg = "TTS API未返回音频数据" print(f"错误: {error_msg}") print(f"========== TTS 失败 ==========\n") raise Exception(error_msg) except Exception as e: print(f"========== TTS 错误 ==========") print(f"错误: {e}") import traceback traceback.print_exc() print(f"========== TTS 错误结束 ==========\n") raise def _get_voice_for_emotion(self, emotion: EmotionType) -> str: """根据情感选择音色""" # 情感到音色的映射 voice_map = { "neutral": "Cherry", "happy": "Cherry", "sad": "Cherry", "excited": "Cherry", "angry": "Cherry", "surprised": "Cherry", "thoughtful": "Cherry", "encouraging": "Cherry", } return voice_map.get(emotion, settings.tts_voice) # 全局AI服务实例 ai_service = AIService()