Tolan/backend/ai_service.py

"""
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: PCM 音频数据块（Base64 解码后的原始数据）
        """
        try:
            import time
            tts_start = time.time()
            
            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}")
            print(f"语速: {settings.tts_speech_rate}x (1.0=正常)")
            print(f"流式模式: 是 (PCM 24kHz)")
            
            # 使用流式 API
            def sync_streaming_tts():
                return dashscope.MultiModalConversation.call(
                    api_key=settings.dashscope_api_key,
                    model=settings.tts_model,
                    text=text,
                    voice=voice,
                    speech_rate=settings.tts_speech_rate,  # 语速调整
                    stream=True  # 🔥 启用流式
                )
            
            print("调用 DashScope TTS 流式 API...")
            
            # 在线程池中调用流式 API
            response = await asyncio.to_thread(sync_streaming_tts)
            
            chunk_count = 0
            first_chunk_time = None
            total_audio_size = 0
            
            # 流式接收音频数据
            for chunk in response:
                if hasattr(chunk, 'output') and hasattr(chunk.output, 'audio'):
                    audio = chunk.output.audio
                    
                    # 接收到音频数据块
                    if audio.data is not None:
                        # 记录首字节时间
                        if first_chunk_time is None:
                            first_chunk_time = time.time()
                            first_byte_latency = (first_chunk_time - tts_start) * 1000
                            print(f"⚡ 首字节延迟: {first_byte_latency:.1f}ms")
                        
                        # Base64 解码 PCM 数据
                        pcm_bytes = base64.b64decode(audio.data)
                        chunk_count += 1
                        total_audio_size += len(pcm_bytes)
                        
                        # 立即返回给调用方（实时传输）
                        yield pcm_bytes
                        await asyncio.sleep(0)  # 让出控制权
                    
                    # 检查是否结束
                    if chunk.output.finish_reason == "stop":
                        tts_end = time.time()
                        total_duration = (tts_end - tts_start) * 1000
                        
                        print(f"✅ TTS 流式完成:")
                        print(f"   - 总块数: {chunk_count}")
                        print(f"   - 音频大小: {total_audio_size} 字节")
                        print(f"   - 总耗时: {total_duration:.0f}ms")
                        if first_chunk_time:
                            print(f"   - 首字节延迟: {(first_chunk_time - tts_start) * 1000:.0f}ms")
                        
                        # 完整音频 URL（可选，用于备份）
                        if hasattr(audio, 'url') and audio.url:
                            print(f"   - 完整音频 URL: {audio.url}")
                        if hasattr(audio, 'expires_at') and audio.expires_at:
                            print(f"   - 过期时间: {audio.expires_at}")
                        
                        print(f"========== TTS 流式成功 ==========\n")
                        break
            
            if chunk_count == 0:
                raise Exception("TTS API 未返回任何音频数据")
                
        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:
        """根据情感选择音色"""
        # 情感到音色的映射（统一使用 Elias 男声）
        voice_map = {
            "neutral": "Elias",
            "happy": "Elias",
            "sad": "Elias",
            "excited": "Elias",
            "angry": "Elias",
            "surprised": "Elias",
            "thoughtful": "Elias",
            "encouraging": "Elias",
        }
        return voice_map.get(emotion, settings.tts_voice)


# 全局AI服务实例
ai_service = AIService()