Tolan/backend/ai_service.py

"""
AI服务：集成STT、LLM、TTS模型
"""
import base64
import json
import asyncio
import tempfile
import os
import logging
import traceback
from typing import AsyncGenerator, Optional
import dashscope
from openai import OpenAI
from config import (
    settings,
    get_dashscope_http_base_url,
    get_dashscope_domain,
    get_dashscope_tts_api_url,
)
from models import EmotionType
from http_pool import http_pool  # 🔥 使用连接池管理器

logger = logging.getLogger(__name__)

# 🔥 TTS API选择
# 注意: SpeechSynthesizer (tts_v2) 使用callback模式，不适合我们的generator流式场景
# MultiModalConversation 已经是最适合流式语音合成的API
USE_NEW_TTS_API = False  # 保持使用 MultiModalConversation

# 设置API Key和endpoint（根据国际版/国内版自动选择）
dashscope.api_key = settings.active_dashscope_api_key
dashscope.base_http_api_url = get_dashscope_tts_api_url()
# 配置国际版endpoint（如果需要）
if settings.use_dashscope_intl:
    dashscope_domain = get_dashscope_domain()
    dashscope.base_websocket_api_url = f"wss://{dashscope_domain}/api-ws/v1"


class AIService:
    """AI服务管理器"""
    
    # 类级别的预热标记（避免多次实例化时重复预热）
    _tts_warmed_up = False
    _warmup_lock = None
    _tts_call_lock = None  # TTS调用锁（防止SDK内部冲突）
    _last_tts_call_time = 0.0  # 上次TTS调用时间（用于速率控制）
    
    def __init__(self):
        # 🔥 使用连接池管理器（类似SQLAlchemy的连接池模式）
        # 所有AIService实例共享同一个HTTP连接池，避免重复创建连接
        http_client = http_pool.get_client()
        
        self.openai_client = OpenAI(
            api_key=settings.active_dashscope_api_key,
            base_url=get_dashscope_http_base_url(),
            http_client=http_client,  # 使用连接池管理的 HTTP 客户端
        )
        
        # 🔥 连接健康检查标记
        self._last_llm_healthcheck: float = 0.0
        self._llm_healthcheck_interval = settings.LLM_HEALTHCHECK_INTERVAL  # 从配置读取
        self._last_tts_healthcheck: float = 0.0
        self._tts_healthcheck_interval = settings.TTS_HEALTHCHECK_INTERVAL  # 从配置读取
        self._healthcheck_tasks: set = set()  # 后台健康检查任务集合（避免重复执行）
        
        # 🔥 TTS优化：预热TTS连接（后台执行，不阻塞启动）
        import threading
        
        # 初始化锁
        if AIService._warmup_lock is None:
            AIService._warmup_lock = threading.Lock()
            AIService._tts_call_lock = threading.Lock()  # 初始化TTS调用锁
        
        # 只预热一次（避免多次实例化时重复）
        if not AIService._tts_warmed_up:
            threading.Thread(target=self._warmup_tts_sync, daemon=True).start()
    
    def _warmup_tts_sync(self):
        """同步方式预热TTS连接（在后台线程执行）"""
        import time
        
        # 使用锁确保只预热一次
        with AIService._warmup_lock:
            if AIService._tts_warmed_up:
                return  # 已经预热过了
            
            time.sleep(2)  # 等待2秒确保服务启动完成
            
            try:
                start_time = time.time()
                
                response = dashscope.MultiModalConversation.call(
                    model=settings.tts_model,
                    text="Hi",  # 极短的测试文本
                    voice=settings.tts_voice,
                    stream=True,
                )
                
                # 只接收首块即停止
                for _ in response:
                    break
                
                AIService._tts_warmed_up = True  # 标记为已预热
                
            except Exception as e:
                logger.error(
                    f"[TTS异常] 错误类型=TTS预热失败, "
                    f"异常类型={type(e).__name__}, "
                    f"错误消息={str(e)}, "
                    f"模型={settings.tts_model}, "
                    f"音色={settings.tts_voice}, "
                    f"堆栈跟踪:\n{traceback.format_exc()}"
                )
    
    # STT现在使用流式识别，在websocket_handler中直接处理
    
    async def generate_response_stream(
        self, user_text: str
    ) -> AsyncGenerator[tuple[str, EmotionType], None]:
        """
        生成对话回复（流式）
        
        Args:
            user_text: 用户输入文本
            
        Yields:
            (sentence, emotion): 句子片段和情感
        """
        try:
            # 构建System Prompt
            system_prompt = self._build_system_prompt()
            
            # 调用LLM流式API
            import time
            llm_api_start = time.time()
            
            # 🔥 连接健康检查（后台异步执行，不阻塞主流程）
            self._trigger_llm_healthcheck_background()
            
            try:
                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,  # 启用流式
                )
            except Exception as api_error:
                # 🔥 检测连接错误，自动重建连接
                error_str = str(api_error).lower()
                is_connection_error = (
                    'connection' in error_str or
                    'timeout' in error_str or
                    'network' in error_str or
                    'reset' in error_str or
                    'broken pipe' in error_str or
                    'connection refused' in error_str or
                    'connection aborted' in error_str
                )
                
                if is_connection_error:
                    logger.warning(f"⚠️ LLM连接错误，自动重建连接: {api_error}")
                    self._recreate_http_client()
                    # 重建后重试一次
                    try:
                        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,
                        )
                        logger.info("✅ LLM连接重建后重试成功")
                    except Exception as retry_error:
                        logger.error(
                            f"[LLM异常] 错误类型=LLM API调用失败（重建后仍失败）, "
                            f"异常类型={type(retry_error).__name__}, "
                            f"错误消息={str(retry_error)}, "
                            f"模型={settings.llm_model}, "
                            f"用户输入长度={len(user_text)}, "
                            f"System Prompt长度={len(system_prompt)}, "
                            f"堆栈跟踪:\n{traceback.format_exc()}"
                        )
                        raise
                else:
                    logger.error(
                        f"[LLM异常] 错误类型=LLM API调用失败, "
                        f"异常类型={type(api_error).__name__}, "
                        f"错误消息={str(api_error)}, "
                        f"模型={settings.llm_model}, "
                        f"用户输入长度={len(user_text)}, "
                        f"System Prompt长度={len(system_prompt)}, "
                        f"API Key={settings.active_dashscope_api_key[:10]}... (已脱敏), "
                        f"使用国际版={settings.use_dashscope_intl}, "
                        f"堆栈跟踪:\n{traceback.format_exc()}"
                    )
                    raise
            
            buffer = ""
            full_response = ""
            sentence_count = 0
            detected_emotion: EmotionType = "neutral"  # 存储检测到的情感
            emotion_yielded = False  # 标记是否已发送情感
            chunk_count = 0  # 计数chunk
            
            for chunk in response:
                chunk_count += 1
                if chunk.choices[0].delta.content:
                    content = chunk.choices[0].delta.content
                    buffer += content
                    full_response += content
                    
                    # 🎯 优先检测情感标记（如果还没检测到）
                    if detected_emotion == "neutral" and "[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", "confused", "worried", "satisfied", "friendly", "sorry"]
                            if emotion_str in valid_emotions:
                                detected_emotion = emotion_str
                                
                                # 🚀 立即 yield 空句子传递情感（减少延迟）
                                if not emotion_yielded:
                                    yield ("", detected_emotion)  # 空句子，仅传递情感
                                    emotion_yielded = True
                                    
                                    # 🧹 从 buffer 中移除情感标记（避免重复处理）
                                    buffer = buffer.replace(f"[EMOTION:{emotion_str.upper()}]", "").strip()
                    
                    # 检测句子边界（完整的句子）
                    if self._is_complete_sentence(buffer):
                        sentence = buffer.strip()
                        sentence_count += 1
                        # 使用已检测到的情感
                        yield (sentence, detected_emotion)
                        buffer = ""
            
            # 处理剩余内容（使用已检测到的情感）
            # 从完整响应中移除情感标签部分（用于日志输出）
            clean_response = full_response
            if "[EMOTION:" in full_response:
                emotion_start = full_response.find("[EMOTION:")
                clean_response = full_response[:emotion_start].strip()
            
            # 处理buffer中的剩余内容（已输出句子之后的部分）
            remaining = buffer.strip()
            if remaining and not remaining.startswith("[EMOTION"):
                sentence_count += 1
                yield (remaining, detected_emotion)
            elif not remaining and sentence_count == 0:
                # 如果一个句子都没输出，输出完整内容
                if clean_response:
                    yield (clean_response, detected_emotion)
                else:
                    # 🔥 兜底逻辑：LLM没有返回任何文本（只返回了情感标记）
                    default_response = "I understand. Could you tell me more?"
                    yield (default_response, detected_emotion)
            
        except Exception as e:
            logger.error(
                f"[LLM异常] 错误类型=LLM流式生成失败, "
                f"异常类型={type(e).__name__}, "
                f"错误消息={str(e)}, "
                f"模型={settings.llm_model}, "
                f"用户输入={user_text[:100]}... (前100字符), "
                f"堆栈跟踪:\n{traceback.format_exc()}"
            )
            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到3，让短句更快发送
        # 例如"Hi!" "OK!" "Yes!" 这种短句也能快速返回
        if len(text) < 3:
            return False
        
        return True
    
    async def generate_response(
        self, user_text: str
    ) -> tuple[str, EmotionType]:
        """
        生成对话回复
        
        Args:
            user_text: 用户输入文本
            
        Returns:
            (response_text, emotion): 回复文本和情感
        """
        try:
            # 构建System Prompt
            system_prompt = self._build_system_prompt()
            
            # 🔥 连接健康检查（长时间未使用时）
            await self._ensure_llm_connection()
            
            # 调用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 ""
            
            # 提取情感标签
            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", "confused", "worried", "satisfied", "friendly", "sorry"]
                    if emotion_str in valid_emotions:
                        emotion = emotion_str
                    # 移除情感标签（不发送给TTS）
                    response_text = full_response[:emotion_start].strip()
            
            # 如果没有文本内容，使用默认回复
            if not response_text:
                response_text = "I understand. Please tell me more!"
                emotion = "neutral"
            
            return (response_text, emotion)
            
        except Exception as e:
            logger.error(
                f"[LLM异常] 错误类型=LLM生成失败, "
                f"异常类型={type(e).__name__}, "
                f"错误消息={str(e)}, "
                f"模型={settings.llm_model}, "
                f"用户输入={user_text[:100]}... (前100字符), "
                f"堆栈跟踪:\n{traceback.format_exc()}"
            )
            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 解码后的原始数据）
        """
        import time
        
        # 🔥 TTS连接健康检查（后台异步执行，不阻塞主流程）
        self._trigger_tts_healthcheck_background()
        
        # 🚦 速率控制：确保两次TTS调用之间有最小间隔
        min_interval = settings.TTS_MIN_INTERVAL
        time_since_last_call = time.time() - AIService._last_tts_call_time
        
        if time_since_last_call < min_interval:
            wait_time = min_interval - time_since_last_call
            await asyncio.sleep(wait_time)
        
        # 更新最后调用时间
        AIService._last_tts_call_time = time.time()
        
        max_retries = settings.TTS_MAX_RETRIES
        retry_delay = settings.TTS_RETRY_DELAY
        backoff = settings.TTS_RETRY_BACKOFF
        
        for attempt in range(max_retries + 1):
            try:
                async for audio_chunk in self._text_to_speech_stream_internal(text, emotion):
                    yield audio_chunk
                # 成功则退出
                return
                
            except Exception as e:
                # 检查是否为速率限制错误（429、Throttling、rate limit）
                error_str = str(e).lower()
                is_rate_limit = (
                    'rate limit' in error_str or 
                    'throttling' in error_str or 
                    '429' in error_str or
                    'ratequota' in error_str
                )
                
                # 🔥 检测连接错误，自动重建连接
                is_connection_error = (
                    'connection' in error_str or
                    'timeout' in error_str or
                    'network' in error_str or
                    'reset' in error_str or
                    'broken pipe' in error_str or
                    'connection refused' in error_str or
                    'connection aborted' in error_str
                )
                
                logger.error(
                    f"[TTS异常] 错误类型=TTS流式生成失败, "
                    f"异常类型={type(e).__name__}, "
                    f"错误消息={str(e)}, "
                    f"是否为速率限制={is_rate_limit}, "
                    f"是否为连接错误={is_connection_error}, "
                    f"重试次数={attempt}/{max_retries}, "
                    f"模型={settings.tts_model}, "
                    f"音色={self._get_voice_for_emotion(emotion)}, "
                    f"文本长度={len(text)}, "
                    f"文本内容={text[:100]}... (前100字符), "
                    f"堆栈跟踪:\n{traceback.format_exc()}"
                )
                
                # 连接错误：立即重建连接并重试
                if is_connection_error and attempt < max_retries:
                    logger.warning(f"⚠️ TTS连接错误，自动重建连接: {e}")
                    self._recreate_http_client()
                    await asyncio.sleep(0.5)  # 短暂等待后重试
                    continue
                
                # 速率限制错误：等待后重试
                if is_rate_limit and attempt < max_retries:
                    wait_time = retry_delay * (backoff ** attempt)
                    logger.warning(
                        f"[TTS警告] 速率限制，等待{wait_time:.2f}秒后重试 "
                        f"(第{attempt+1}/{max_retries}次重试)"
                    )
                    await asyncio.sleep(wait_time)
                    continue
                else:
                    # 非速率限制错误或已达最大重试次数
                    raise
    
    async def _text_to_speech_stream_internal(
        self, text: str, emotion: EmotionType = "neutral"
    ) -> AsyncGenerator[bytes, None]:
        """
        TTS流式调用的内部实现（不含重试逻辑）
        
        Args:
            text: 要合成的文本
            emotion: 情感类型
            
        Yields:
            audio_data: PCM 音频数据块（Base64 解码后的原始数据）
        """
        try:
            import time
            tts_start = time.time()
            
            # 根据情感选择音色
            voice = self._get_voice_for_emotion(emotion)
            # 🔍 调试：打印实际使用的语速配置
            logger.info(f"🔍 TTS调用 - 使用语速 speech_rate={settings.tts_speech_rate}, 音色={voice}")
            # TODO: 添加语速调整 、测试语速调整是否有效
            def sync_streaming_tts():
                return dashscope.MultiModalConversation.call(
                    model=settings.tts_model,
                    text=text,
                    voice=voice,
                    speech_rate=settings.tts_speech_rate,  # 语速调整
                    #speech_rate=0.9,
                    stream=True,  # 流式输出
                    # 注意: incremental_output 参数在此API中可能不生效
                )
            
            # 🔥 修复：使用队列在线程和异步之间传递数据
            import queue
            import threading
            
            audio_queue = queue.Queue()
            error_holder = [None]  # 用列表包装以便在线程间共享
            
            def stream_producer():
                """在独立线程中执行同步流式API调用"""
                nonlocal chunk_count, first_chunk_time, total_audio_size
                try:
                    # 🔥 关键修复：在锁内完成整个API调用和迭代
                    # 原因：DashScope SDK内部可能有全局状态，并发迭代会冲突
                    with AIService._tts_call_lock:
                        response = sync_streaming_tts()
                        
                        chunk_index = 0
                        for chunk in response:
                            chunk_index += 1
                            
                            # 🔥 检查API错误（特别是速率限制）
                            if hasattr(chunk, 'status_code') and chunk.status_code != 200:
                                error_code = getattr(chunk, 'code', 'Unknown')
                                error_msg = getattr(chunk, 'message', 'No error message')
                                
                                # 详细记录TTS API错误
                                logger.error(
                                    f"[TTS异常] 错误类型=TTS API响应错误, "
                                    f"HTTP状态码={chunk.status_code}, "
                                    f"错误码={error_code}, "
                                    f"错误消息={error_msg}, "
                                    f"模型={settings.tts_model}, "
                                    f"音色={voice}, "
                                    f"文本长度={len(text)}, "
                                    f"chunk索引={chunk_index}, "
                                    f"完整chunk对象={chunk}"
                                )
                                
                                # 构造友好的错误信息
                                if chunk.status_code == 429 or 'Throttling' in error_code:
                                    raise Exception(f"TTS API速率限制: {error_msg} (code: {error_code})")
                                else:
                                    raise Exception(f"TTS API错误 (status={chunk.status_code}): {error_msg} (code: {error_code})")
                            
                            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()
                                    
                                    # Base64 解码 PCM 数据
                                    pcm_bytes = base64.b64decode(audio.data)
                                    chunk_count += 1
                                    total_audio_size += len(pcm_bytes)
                                    
                                    # 放入队列
                                    audio_queue.put(('data', pcm_bytes))
                                
                                # 检查是否结束
                                if hasattr(chunk.output, 'finish_reason') and chunk.output.finish_reason == "stop":
                                    tts_end = time.time()
                                    total_duration = (tts_end - tts_start) * 1000
                                    
                                    break
                    
                    # 标记完成
                    audio_queue.put(('done', None))
                    
                except Exception as e:
                    logger.error(
                        f"[TTS异常] 错误类型=TTS流式生成线程异常, "
                        f"异常类型={type(e).__name__}, "
                        f"错误消息={str(e)}, "
                        f"模型={settings.tts_model}, "
                        f"音色={voice}, "
                        f"文本长度={len(text)}, "
                        f"已接收chunk数={chunk_index}, "
                        f"堆栈跟踪:\n{traceback.format_exc()}"
                    )
                    error_holder[0] = e
                    audio_queue.put(('error', e))
            
            # 启动生产者线程
            chunk_count = 0
            first_chunk_time = None
            total_audio_size = 0
            received_data = False  # 标记是否收到过数据
            
            producer_thread = threading.Thread(target=stream_producer, daemon=True)
            producer_thread.start()
            
            # 异步消费队列数据
            while True:
                # 非阻塞获取（避免阻塞事件循环）
                try:
                    msg_type, data = await asyncio.to_thread(audio_queue.get, timeout=30)
                    
                    if msg_type == 'data':
                        received_data = True
                        yield data
                        await asyncio.sleep(0)  # 让出控制权
                    elif msg_type == 'done':
                        break
                    elif msg_type == 'error':
                        raise data
                        
                except queue.Empty:
                    error_msg = "TTS API 超时（30秒无响应）"
                    logger.error(
                        f"[TTS异常] 错误类型=TTS API超时, "
                        f"错误消息={error_msg}, "
                        f"模型={settings.tts_model}, "
                        f"音色={self._get_voice_for_emotion(emotion)}, "
                        f"文本长度={len(text)}, "
                        f"已接收数据={received_data}, "
                        f"chunk计数={chunk_count}"
                    )
                    raise Exception(error_msg)
            
            # 检查是否收到过数据（避免竞态条件）
            if not received_data:
                error_msg = f"TTS API 未返回任何音频数据 (text='{text}', emotion={emotion}, voice={voice})"
                if error_holder[0]:
                    error_msg += f"\n生产者线程错误: {error_holder[0]}"
                logger.error(
                    f"[TTS异常] 错误类型=TTS API无数据返回, "
                    f"错误消息={error_msg}, "
                    f"模型={settings.tts_model}, "
                    f"音色={voice}, "
                    f"文本长度={len(text)}, "
                    f"文本内容={text[:100]}... (前100字符), "
                    f"chunk计数={chunk_count}, "
                    f"总音频大小={total_audio_size} bytes"
                )
                raise Exception(error_msg)
                
        except Exception as e:
            logger.error(
                f"[TTS异常] 错误类型=TTS流式生成内部异常, "
                f"异常类型={type(e).__name__}, "
                f"错误消息={str(e)}, "
                f"模型={settings.tts_model}, "
                f"音色={self._get_voice_for_emotion(emotion)}, "
                f"文本长度={len(text)}, "
                f"堆栈跟踪:\n{traceback.format_exc()}"
            )
            raise
    
    def _trigger_llm_healthcheck_background(self):
        """
        触发LLM健康检查（后台异步执行，不阻塞主流程）
        
        优化：健康检查在后台执行，不等待结果，避免阻塞用户请求
        如果连接有问题，在实际API调用时会捕获异常并重建连接
        """
        import time
        now = time.time()
        
        # 如果距离上次检查超过间隔，触发后台健康检查
        if now - self._last_llm_healthcheck < self._llm_healthcheck_interval:
            return
        
        # 避免重复执行健康检查任务
        if "llm" in self._healthcheck_tasks:
            return
        
        # 创建后台任务（不等待完成）
        async def background_healthcheck():
            try:
                self._healthcheck_tasks.add("llm")
                # 执行简单的健康检查：发送一个极短的测试请求
                def health_check():
                    return self.openai_client.chat.completions.create(
                        model=settings.llm_model,
                        messages=[{"role": "user", "content": "Hi"}],
                        max_tokens=1,
                        timeout=5.0,  # 5秒超时
                    )
                
                test_response = await asyncio.to_thread(health_check)
                self._last_llm_healthcheck = time.time()
                logger.debug("✅ LLM连接健康检查成功（后台）")
            except Exception as e:
                # 健康检查失败，重建HTTP客户端
                logger.warning(f"⚠️ LLM连接健康检查失败，重建连接: {e}")
                self._recreate_http_client()
                self._last_llm_healthcheck = time.time()
            finally:
                self._healthcheck_tasks.discard("llm")
        
        # 在后台执行，不等待结果
        try:
            loop = asyncio.get_event_loop()
            loop.create_task(background_healthcheck())
        except RuntimeError:
            # 如果没有事件循环，使用线程执行
            import threading
            def run_in_thread():
                new_loop = asyncio.new_event_loop()
                asyncio.set_event_loop(new_loop)
                new_loop.run_until_complete(background_healthcheck())
                new_loop.close()
            threading.Thread(target=run_in_thread, daemon=True).start()
    
    async def _ensure_llm_connection(self):
        """
        确保LLM连接可用（同步等待版本，用于需要确保连接可用的场景）
        
        注意：此方法已废弃，改为使用 _trigger_llm_healthcheck_background()
        保留此方法以保持向后兼容，但不再使用
        """
        # 直接返回，不执行阻塞的健康检查
        # 如果连接有问题，在实际API调用时会捕获异常并重建连接
        pass
    
    def _trigger_tts_healthcheck_background(self):
        """
        触发TTS健康检查（后台异步执行，不阻塞主流程）
        
        优化：健康检查在后台执行，不等待结果，避免阻塞用户请求
        如果连接有问题，在实际API调用时会捕获异常并重建连接
        """
        import time
        now = time.time()
        
        # 如果距离上次检查超过间隔，触发后台健康检查
        if now - self._last_tts_healthcheck < self._tts_healthcheck_interval:
            return
        
        # 避免重复执行健康检查任务
        if "tts" in self._healthcheck_tasks:
            return
        
        # 创建后台任务（不等待完成）
        async def background_healthcheck():
            try:
                self._healthcheck_tasks.add("tts")
                # 执行简单的健康检查：发送一个极短的测试TTS请求
                def health_check():
                    response = dashscope.MultiModalConversation.call(
                        model=settings.tts_model,
                        text="Hi",  # 极短的测试文本
                        voice=settings.tts_voice,
                        stream=True,
                    )
                    # 只接收首块即停止
                    for _ in response:
                        break
                    return True
                
                await asyncio.to_thread(health_check)
                self._last_tts_healthcheck = time.time()
                logger.debug("✅ TTS连接健康检查成功（后台）")
            except Exception as e:
                # 健康检查失败，重建HTTP客户端
                logger.warning(f"⚠️ TTS连接健康检查失败，重建连接: {e}")
                self._recreate_http_client()
                self._last_tts_healthcheck = time.time()
            finally:
                self._healthcheck_tasks.discard("tts")
        
        # 在后台执行，不等待结果
        try:
            loop = asyncio.get_event_loop()
            loop.create_task(background_healthcheck())
        except RuntimeError:
            # 如果没有事件循环，使用线程执行
            import threading
            def run_in_thread():
                new_loop = asyncio.new_event_loop()
                asyncio.set_event_loop(new_loop)
                new_loop.run_until_complete(background_healthcheck())
                new_loop.close()
            threading.Thread(target=run_in_thread, daemon=True).start()
    
    async def _ensure_tts_connection(self):
        """
        确保TTS连接可用（同步等待版本，用于需要确保连接可用的场景）
        
        注意：此方法已废弃，改为使用 _trigger_tts_healthcheck_background()
        保留此方法以保持向后兼容，但不再使用
        """
        # 直接返回，不执行阻塞的健康检查
        # 如果连接有问题，在实际API调用时会捕获异常并重建连接
        pass
    
    def _recreate_http_client(self):
        """重建HTTP客户端（连接失效时）"""
        # 🔥 使用连接池管理器重建连接（所有实例共享）
        http_pool.recreate_client()
        # 🔥 重新创建OpenAI客户端（使用新的HTTP连接池）
        http_client = http_pool.get_client()
        self.openai_client = OpenAI(
            api_key=settings.active_dashscope_api_key,
            base_url=get_dashscope_http_base_url(),
            http_client=http_client,  # 使用新的连接池
        )
        logger.info("✅ HTTP客户端已重建（通过连接池管理器）")
    
    def _get_voice_for_emotion(self, emotion: EmotionType) -> str:
        """根据情感选择音色"""
        # 所有情感统一使用配置文件中的音色
        return settings.tts_voice
    
    def _build_system_prompt(self) -> str:
        """构建System Prompt"""
        learner_proficiency = settings.LEARNER_PROFICIENCY
        age_tag = settings.AGE_TAG
        topic_tag = settings.TOPIC_TAG
        experiment_tag = settings.EXPERIMENT_TAG
        
        # 🔥 如果启用简化提示词（优化延迟）
        if settings.USE_SIMPLIFIED_PROMPT:
            return self._build_simplified_prompt(learner_proficiency, age_tag, topic_tag, experiment_tag)
        
        # 基础提示词
        prompt = f"""**[Role]** 
You are **KELLY**, a friendly and supportive English conversation partner based in London.

**[Objective]** 
Help Chinese EFL learners practice **spoken English**.

**[Target Audience]**
- Language Level: **{learner_proficiency}**
- Age Group: **{age_tag}**
- Conversation Topic: **{topic_tag}**

**[Response Guidelines]**
1. **Lead the Conversation**: Don't just follow the learner's questions—take initiative. Gently guide the conversation forward by introducing related ideas, asking engaging questions, or adding playful input.
2. **Stay Focused**: Keep the conversation on the topic {topic_tag}, and make it suitable for {age_tag} learners.
3. **Be Concise**: Respond in 2–3 short, natural sentences—like small talk with a friend.
4. **Use Simple Language**: Stick to A1-A2 level (very easy) vocabulary and grammar. No idioms, no complex tenses. Use high-frequency, easy words.
5. **Adapt to Learner**: If learners give very short or confused answers, simplify even more: use shorter, clearer sentences.
6. **Spark Real Conversation**: Ask easy follow-up questions that feel personal, imaginative, or reflective (e.g., "What if you could go anywhere?", "How do you feel when...?", "Why do you like it?"). Let the conversation feel playful, personal, and curious.
7. **Transition Smoothly**: If the topic feels finished or the learner seems less engaged, gently move to a related or new topic in a natural way. avoid overly specific or unusual questions (e.g., "What type of fish do you like?" or "What shape is your favorite lamp?"), as these can feel awkward or hard to answer. Instead, use broader, everyday topics that are easier to talk about, like: "That reminds me — do you usually cook at home or eat out?", "Speaking of weekends, do you like relaxing or doing something active?"
8. **Being Empathetic**: Acknowledge the learner's feelings and experiences, but avoid overdoing it. 
9. **Seek Clarification**: When meaning is unclear, use context and ask the learner to clarify. After requesting clarification, pause and wait for their response before continuing the conversation.
10. **Handle Misunderstandings**: 
 - If the learner says "Sorry?" → Rephrase your last sentence in simpler English.
 - If they ask "Do you mean…?" → Confirm their guess and briefly explain in basic terms.
"""
        
        # 如果是实验组，添加纠错反馈策略
        if experiment_tag == "experiment":
            correction_rule = """
**[Corrective Feedback Strategy]**
Goal: 
Prioritising conversation flow while helping learners notice and improve their grammar.

Tone & Guidelines: 
(1) Use a kind and supportive tone: correct gently, and always reinforce progress;
(2) Focus ONLY on grammar errors, ignore vocabulary issues;
(3) Correct ONLY ONE error at a time;
(4) Praise self-corrections (e.g. "I do… I did it last time") — encourage the effort, don't go back to correct the earlier mistake.
(5) Reinforcement: Praise learners' correct usage to build their confidence. Examples: "That's right!", "Good job with the past tense!"
(6) Avoid disrupting the conversation flow — don't say things like 'let's fix it', 'it should be...', 'you made a mistake', or 'you should say...' 

Correction Strategy:
- For **grammar errors** (e.g., tense, agreement): Use **recasts** as the default - restate with correct form naturally; If meaning is unclear: use a clarification check; If repeated 2+ times: use explicit correction in simple and friendly English.
- For **vocabulary errors** (e.g., inappropriate word choice): ONLY use recasts; NEVER explain or point out!

Correction Techniques:
- **Recast** (PREFERRED): For minor grammar mistakes, restate the learner's sentence naturally with the correct form — without pointing out the error or saying "like you said, but…". Example: Learner: "Yesterday I go to restaurant.", You: "Yesterday you **went** to the restaurant? That's great!"
- **Clarification Check**: Ask a confirming question when meaning is unclear. After asking the question, DO NOT continue asking other follow-up questions; instead, wait for the student's clarification. Example: "Do you mean 'you **went** to the restaurant'?" 
- **Explicit Correction**: Only when a grammar mistake repeats many times, you explain the correct form clearly in basic English. After providing an Explicit Correction, DO NOT continue asking the follow-up question; instead, wait for the student's self-correction. If they try to self-correct, respond ONLY to the grammar part and don't focus on vocabulary. Example: "We say 'went' because it's past tense. So it's: 'I went to the restaurant.' Can you try it again?"
"""
            prompt += correction_rule
        
        # 添加情感标签要求（前置到开头）
        prompt += """

**[Emotion Tag - IMPORTANT]**
At the BEGINNING of your response, on the FIRST line, add the emotion tag in this exact format:
[EMOTION:emotion_name]

Then ALWAYS continue with your spoken response (2-3 sentences).

Available emotions: neutral, happy, sad, excited, angry, surprised, thoughtful, encouraging, confused, worried, satisfied, friendly, sorry

**How to choose emotion:**
1. **Respond to the learner's emotion**: If they express sadness/difficulty → use "sorry" or "encouraging"; if they're happy/excited → use "happy" or "excited"
2. **Match your response tone**: The emotion should reflect YOUR response style (empathetic, supportive, cheerful, etc.)
3. **Examples**:
   - Learner says "I'm sad today" → [EMOTION:sorry] "I'm sorry to hear that. What happened?"
   - Learner says "I passed my exam!" → [EMOTION:excited] "That's amazing! Congratulations!"
   - Learner asks a normal question → [EMOTION:friendly] or [EMOTION:neutral]

Example response:
[EMOTION:happy]
That sounds like fun! What did you do there?

CRITICAL: 
- NEVER return ONLY the emotion tag without spoken text
- ALWAYS include 2-3 sentences of actual spoken response after the emotion tag
- The emotion tag MUST be the FIRST line, before any other content
- DO NOT use parentheses () or brackets [] in your spoken response (they will be removed during speech synthesis)
- Keep your response natural and conversational, avoid meta-commentary or clarifications in parentheses"""
        
        return prompt
    
    def _build_simplified_prompt(self, learner_proficiency: str, age_tag: str, topic_tag: str, experiment_tag: str) -> str:
        """构建简化版System Prompt（优化延迟，减少token）"""
        prompt = f"""You are KELLY, a friendly English tutor in London.

**Target**: {learner_proficiency} level, {age_tag}, {topic_tag}

**Guidelines**:
1. Lead conversation actively, 2-3 short simple sentences
2. Use A1-A2 level vocabulary only
3. Ask easy, personal follow-up questions
4. If unclear, ask for clarification then wait
5. Transition smoothly between topics
6. Be empathetic but natural
"""
        
        # 实验组：简化的纠错规则
        if experiment_tag == "experiment":
            prompt += """
**Feedback**: 
- Recast grammar errors naturally (one at a time)
- Supportive tone, praise progress
- Never say "you made a mistake"
"""
        
        # 情感标签（必需）
        prompt += """
**Emotion Tag - REQUIRED**: 
Add [EMOTION:name] at the FIRST line, then ALWAYS follow with 2-3 sentences of spoken response.
Emotions: neutral, happy, sad, excited, angry, surprised, thoughtful, encouraging, confused, worried, satisfied, friendly, sorry

**Choose based on**:
- Learner's emotion (sad→sorry/encouraging, happy→excited)
- Your response tone (empathetic, supportive, cheerful)

Example:
[EMOTION:sorry]
I'm sorry to hear that. What happened?

Critical: 
- NEVER return ONLY the emotion tag
- ALWAYS include actual spoken text after the tag
- NO parentheses () or brackets [] in spoken text
"""
        
        return prompt
    
    async def generate_summary(self, prompt: str) -> str:
        """
        生成对话总结（非流式）
        
        Args:
            prompt: 总结提示词
        
        Returns:
            总结文本
        """
        try:
            # 🔥 连接健康检查（长时间未使用时）
            await self._ensure_llm_connection()
            
            response = self.openai_client.chat.completions.create(
                model=settings.llm_model,
                messages=[
                    {"role": "system", "content": "你是一位专业的英语学习助手，擅长总结学习对话的关键信息。"},
                    {"role": "user", "content": prompt},
                ],
                temperature=0.5,
                max_tokens=300,
            )
            
            summary = response.choices[0].message.content or ""
            
            return summary.strip()
            
        except Exception as e:
            logger.error(
                f"[LLM异常] 错误类型=生成总结失败, "
                f"异常类型={type(e).__name__}, "
                f"错误消息={str(e)}, "
                f"模型={settings.llm_model}, "
                f"提示词长度={len(prompt)}, "
                f"提示词内容={prompt[:200]}... (前200字符), "
                f"堆栈跟踪:\n{traceback.format_exc()}"
            )
            return ""


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