""" 双人对练分析服务 功能： - 分析双人对练对话 - 生成双方评估报告 - 对话标注和建议 """ 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 )