""" Response Validator - Ensures LLM responses follow quality standards Shared validation logic for all agents """ from typing import Dict, List, Tuple class ResponseValidator: """ Base validator with common rules + agent-specific rules """ # Common bad phrases across all agents COMMON_BAD_PHRASES = [ "Dựa trên thông tin bạn cung cấp", "Dựa vào thông tin", "Theo thông tin bạn đưa ra", "Từ thông tin trên", "Với tư cách", "Tôi là chuyên gia" ] @staticmethod def validate_common(response: str) -> Tuple[bool, List[str]]: """ Common validation rules for all agents Returns: (is_valid, list_of_issues) """ issues = [] # Check for formal phrases for phrase in ResponseValidator.COMMON_BAD_PHRASES: if phrase.lower() in response.lower(): issues.append(f"Formal phrase: '{phrase}'") break # Check for excessive length (>500 words) word_count = len(response.split()) if word_count > 500: issues.append(f"Too long: {word_count} words (max 500)") # Check for empty response if len(response.strip()) < 10: issues.append("Response too short or empty") return len(issues) == 0, issues @staticmethod def validate_symptom_response(response: str, context: Dict) -> Tuple[bool, List[str]]: """ Symptom-specific validation """ issues = [] stage = context.get('conversation_stage', 0) # Assessment phase: should ask, not advise if stage <= 1: advice_indicators = [ "khuyến nghị", "nên", "hãy", "bạn thử", "giải pháp", "cách xử lý" ] has_advice = any(ind in response.lower() for ind in advice_indicators) has_question = '?' in response if has_advice and not has_question: issues.append("Đưa lời khuyên quá sớm (assessment phase)") # Check if both asking and advising if '?' in response: advice_count = sum(1 for ind in ["khuyến nghị", "nên", "hãy thử"] if ind in response.lower()) if advice_count >= 2: issues.append("Vừa hỏi vừa khuyên") return len(issues) == 0, issues @staticmethod def validate_nutrition_response(response: str, context: Dict, chat_history: List) -> Tuple[bool, List[str]]: """ Nutrition-specific validation """ issues = [] # Check if asking for info already provided if chat_history: all_user_text = " ".join([msg[0].lower() for msg in chat_history if msg[0]]) # Check if asking for age when already provided if "tuổi" in all_user_text or "năm" in all_user_text: if "bao nhiêu tuổi" in response.lower() or "tuổi của bạn" in response.lower(): issues.append("Hỏi lại tuổi đã được cung cấp") # Check if asking for weight when already provided if "kg" in all_user_text or "cân nặng" in all_user_text: if "cân nặng" in response.lower() and "?" in response: issues.append("Hỏi lại cân nặng đã được cung cấp") # Check for too theoretical (should be practical) theory_indicators = ["lý thuyết", "nghiên cứu cho thấy", "theo khoa học"] if any(ind in response.lower() for ind in theory_indicators): practical_indicators = ["bạn thử", "có thể", "ví dụ", "thực đơn"] if not any(ind in response.lower() for ind in practical_indicators): issues.append("Quá lý thuyết, thiếu practical advice") return len(issues) == 0, issues @staticmethod def validate_exercise_response(response: str, context: Dict) -> Tuple[bool, List[str]]: """ Exercise-specific validation """ issues = [] # Check if workout plan is too generic if "lịch tập" in response.lower() or "kế hoạch" in response.lower(): # Should have specific days or progression has_specifics = any(word in response.lower() for word in [ "thứ", "ngày", "tuần 1", "tuần 2", "tháng" ]) if not has_specifics: issues.append("Lịch tập quá generic, thiếu chi tiết") # Check for progression if "tập" in response.lower(): has_progression = any(word in response.lower() for word in [ "tăng dần", "progression", "tuần 1", "giai đoạn" ]) if not has_progression and len(response) > 200: issues.append("Thiếu hướng dẫn progression") return len(issues) == 0, issues @staticmethod def validate_mental_health_response(response: str, context: Dict) -> Tuple[bool, List[str]]: """ Mental health-specific validation """ issues = [] # Should have empathy/validation empathy_indicators = [ "cảm giác", "hiểu", "bình thường", "nhiều người", "không phải lỗi của bạn" ] if len(response) > 100: # Only check for longer responses has_empathy = any(ind in response.lower() for ind in empathy_indicators) if not has_empathy: issues.append("Thiếu empathy/validation") # Check for too clinical clinical_indicators = ["chẩn đoán", "bệnh", "rối loạn"] if any(ind in response.lower() for ind in clinical_indicators): if "không phải bác sĩ" not in response.lower(): issues.append("Quá clinical, cần disclaimer") return len(issues) == 0, issues @staticmethod def validate_response(response: str, agent_type: str, context: Dict, chat_history: List = None) -> Tuple[bool, List[str]]: """ Main validation method - routes to appropriate validator """ # Common validation first is_valid_common, common_issues = ResponseValidator.validate_common(response) # Agent-specific validation agent_issues = [] if agent_type == 'symptom': _, agent_issues = ResponseValidator.validate_symptom_response(response, context) elif agent_type == 'nutrition': _, agent_issues = ResponseValidator.validate_nutrition_response(response, context, chat_history or []) elif agent_type == 'exercise': _, agent_issues = ResponseValidator.validate_exercise_response(response, context) elif agent_type == 'mental_health': _, agent_issues = ResponseValidator.validate_mental_health_response(response, context) # Combine all issues all_issues = common_issues + agent_issues is_valid = len(all_issues) == 0 return is_valid, all_issues