012-kaopeilian/backend/app/services/ai/duo_practice_analysis_service.py

"""
双人对练分析服务

功能：
- 分析双人对练对话
- 生成双方评估报告
- 对话标注和建议
"""
import json
import logging
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional

from app.services.ai.ai_service import AIService
from app.services.ai.prompts.duo_practice_prompts import SYSTEM_PROMPT, USER_PROMPT

logger = logging.getLogger(__name__)


@dataclass
class UserEvaluation:
    """用户评估结果"""
    user_name: str
    role_name: str
    total_score: int
    dimensions: Dict[str, Dict[str, Any]]
    highlights: List[str]
    improvements: List[Dict[str, str]]


@dataclass
class DuoPracticeAnalysisResult:
    """双人对练分析结果"""
    # 整体评估
    interaction_quality: int = 0
    scene_restoration: int = 0
    overall_comment: str = ""
    
    # 用户A评估
    user_a_evaluation: Optional[UserEvaluation] = None
    
    # 用户B评估
    user_b_evaluation: Optional[UserEvaluation] = None
    
    # 对话标注
    dialogue_annotations: List[Dict[str, Any]] = field(default_factory=list)
    
    # AI 元数据
    raw_response: str = ""
    ai_provider: str = ""
    ai_model: str = ""
    ai_latency_ms: int = 0


class DuoPracticeAnalysisService:
    """
    双人对练分析服务
    
    使用示例：
    ```python
    service = DuoPracticeAnalysisService()
    result = await service.analyze(
        scene_name="销售场景",
        scene_background="客户咨询产品",
        role_a_name="销售顾问",
        role_b_name="顾客",
        user_a_name="张三",
        user_b_name="李四",
        dialogue_history=dialogue_list,
        duration_seconds=300,
        total_turns=20
    )
    ```
    """
    
    MODULE_CODE = "duo_practice_analysis"
    
    async def analyze(
        self,
        scene_name: str,
        scene_background: str,
        role_a_name: str,
        role_b_name: str,
        role_a_description: str,
        role_b_description: str,
        user_a_name: str,
        user_b_name: str,
        dialogue_history: List[Dict[str, Any]],
        duration_seconds: int,
        total_turns: int,
        db: Any = None
    ) -> DuoPracticeAnalysisResult:
        """
        分析双人对练
        
        Args:
            scene_name: 场景名称
            scene_background: 场景背景
            role_a_name: 角色A名称
            role_b_name: 角色B名称
            role_a_description: 角色A描述
            role_b_description: 角色B描述
            user_a_name: 用户A名称
            user_b_name: 用户B名称
            dialogue_history: 对话历史列表
            duration_seconds: 对练时长（秒）
            total_turns: 总对话轮次
            db: 数据库会话
            
        Returns:
            DuoPracticeAnalysisResult: 分析结果
        """
        try:
            logger.info(f"开始双人对练分析: {scene_name}, 轮次={total_turns}")
            
            # 格式化对话历史
            dialogue_text = self._format_dialogue_history(dialogue_history)
            
            # 创建 AI 服务
            ai_service = AIService(module_code=self.MODULE_CODE, db_session=db)
            
            # 构建用户提示词
            user_prompt = USER_PROMPT.format(
                scene_name=scene_name,
                scene_background=scene_background or "未设置",
                role_a_name=role_a_name,
                role_b_name=role_b_name,
                role_a_description=role_a_description or f"扮演{role_a_name}角色",
                role_b_description=role_b_description or f"扮演{role_b_name}角色",
                user_a_name=user_a_name,
                user_b_name=user_b_name,
                dialogue_history=dialogue_text,
                duration_seconds=duration_seconds,
                total_turns=total_turns
            )
            
            # 调用 AI
            messages = [
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": user_prompt}
            ]
            
            ai_response = await ai_service.chat(
                messages=messages,
                model="gemini-3-flash-preview",  # 使用快速模型
                temperature=0.3,
                prompt_name="duo_practice_analysis"
            )
            
            logger.info(f"AI 分析完成: provider={ai_response.provider}, latency={ai_response.latency_ms}ms")
            
            # 解析 AI 输出
            result = self._parse_analysis_result(
                ai_response.content,
                user_a_name=user_a_name,
                user_b_name=user_b_name,
                role_a_name=role_a_name,
                role_b_name=role_b_name
            )
            
            # 填充 AI 元数据
            result.raw_response = ai_response.content
            result.ai_provider = ai_response.provider
            result.ai_model = ai_response.model
            result.ai_latency_ms = ai_response.latency_ms
            
            return result
            
        except Exception as e:
            logger.error(f"双人对练分析失败: {e}", exc_info=True)
            # 返回空结果
            return DuoPracticeAnalysisResult(
                overall_comment=f"分析失败: {str(e)}"
            )
    
    def _format_dialogue_history(self, dialogues: List[Dict[str, Any]]) -> str:
        """格式化对话历史"""
        lines = []
        for d in dialogues:
            speaker = d.get("role_name") or d.get("speaker", "未知")
            content = d.get("content", "")
            seq = d.get("sequence", 0)
            lines.append(f"[{seq}] {speaker}：{content}")
        return "\n".join(lines)
    
    def _parse_analysis_result(
        self,
        ai_output: str,
        user_a_name: str,
        user_b_name: str,
        role_a_name: str,
        role_b_name: str
    ) -> DuoPracticeAnalysisResult:
        """解析 AI 输出"""
        result = DuoPracticeAnalysisResult()
        
        try:
            # 尝试提取 JSON
            json_str = ai_output
            
            # 如果输出包含 markdown 代码块，提取其中的 JSON
            if "```json" in ai_output:
                start = ai_output.find("```json") + 7
                end = ai_output.find("```", start)
                json_str = ai_output[start:end].strip()
            elif "```" in ai_output:
                start = ai_output.find("```") + 3
                end = ai_output.find("```", start)
                json_str = ai_output[start:end].strip()
            
            data = json.loads(json_str)
            
            # 解析整体评估
            overall = data.get("overall_evaluation", {})
            result.interaction_quality = overall.get("interaction_quality", 0)
            result.scene_restoration = overall.get("scene_restoration", 0)
            result.overall_comment = overall.get("overall_comment", "")
            
            # 解析用户A评估
            user_a_data = data.get("user_a_evaluation", {})
            if user_a_data:
                result.user_a_evaluation = UserEvaluation(
                    user_name=user_a_data.get("user_name", user_a_name),
                    role_name=user_a_data.get("role_name", role_a_name),
                    total_score=user_a_data.get("total_score", 0),
                    dimensions=user_a_data.get("dimensions", {}),
                    highlights=user_a_data.get("highlights", []),
                    improvements=user_a_data.get("improvements", [])
                )
            
            # 解析用户B评估
            user_b_data = data.get("user_b_evaluation", {})
            if user_b_data:
                result.user_b_evaluation = UserEvaluation(
                    user_name=user_b_data.get("user_name", user_b_name),
                    role_name=user_b_data.get("role_name", role_b_name),
                    total_score=user_b_data.get("total_score", 0),
                    dimensions=user_b_data.get("dimensions", {}),
                    highlights=user_b_data.get("highlights", []),
                    improvements=user_b_data.get("improvements", [])
                )
            
            # 解析对话标注
            result.dialogue_annotations = data.get("dialogue_annotations", [])
            
        except json.JSONDecodeError as e:
            logger.warning(f"JSON 解析失败: {e}")
            result.overall_comment = "AI 输出格式异常，请重试"
        except Exception as e:
            logger.error(f"解析分析结果失败: {e}")
            result.overall_comment = f"解析失败: {str(e)}"
        
        return result
    
    def result_to_dict(self, result: DuoPracticeAnalysisResult) -> Dict[str, Any]:
        """将结果转换为字典（用于 API 响应）"""
        return {
            "overall_evaluation": {
                "interaction_quality": result.interaction_quality,
                "scene_restoration": result.scene_restoration,
                "overall_comment": result.overall_comment
            },
            "user_a_evaluation": {
                "user_name": result.user_a_evaluation.user_name,
                "role_name": result.user_a_evaluation.role_name,
                "total_score": result.user_a_evaluation.total_score,
                "dimensions": result.user_a_evaluation.dimensions,
                "highlights": result.user_a_evaluation.highlights,
                "improvements": result.user_a_evaluation.improvements
            } if result.user_a_evaluation else None,
            "user_b_evaluation": {
                "user_name": result.user_b_evaluation.user_name,
                "role_name": result.user_b_evaluation.role_name,
                "total_score": result.user_b_evaluation.total_score,
                "dimensions": result.user_b_evaluation.dimensions,
                "highlights": result.user_b_evaluation.highlights,
                "improvements": result.user_b_evaluation.improvements
            } if result.user_b_evaluation else None,
            "dialogue_annotations": result.dialogue_annotations,
            "ai_metadata": {
                "provider": result.ai_provider,
                "model": result.ai_model,
                "latency_ms": result.ai_latency_ms
            }
        }


# ==================== 全局实例 ====================

duo_practice_analysis_service = DuoPracticeAnalysisService()


# ==================== 便捷函数 ====================

async def analyze_duo_practice(
    scene_name: str,
    scene_background: str,
    role_a_name: str,
    role_b_name: str,
    role_a_description: str,
    role_b_description: str,
    user_a_name: str,
    user_b_name: str,
    dialogue_history: List[Dict[str, Any]],
    duration_seconds: int,
    total_turns: int,
    db: Any = None
) -> DuoPracticeAnalysisResult:
    """便捷函数：分析双人对练"""
    return await duo_practice_analysis_service.analyze(
        scene_name=scene_name,
        scene_background=scene_background,
        role_a_name=role_a_name,
        role_b_name=role_b_name,
        role_a_description=role_a_description,
        role_b_description=role_b_description,
        user_a_name=user_a_name,
        user_b_name=user_b_name,
        dialogue_history=dialogue_history,
        duration_seconds=duration_seconds,
        total_turns=total_turns,
        db=db
    )