""" Exercise Agent - Specialized agent for exercise and fitness advice """ from config.settings import client, MODEL from modules.exercise.exercise import generate_exercise_plan from health_data import HealthContext from fitness_tracking import FitnessTracker from rag.rag_integration import get_rag_integration from agents.core.base_agent import BaseAgent from typing import Dict, Any, List, Optional from datetime import datetime import re class ExerciseAgent(BaseAgent): def __init__(self, memory=None): super().__init__(memory) self.health_context = None self.fitness_tracker = None self.rag = get_rag_integration() # Configure handoff triggers for exercise agent self.handoff_triggers = { 'nutrition_agent': ['ăn gì', 'thực đơn', 'calo', 'dinh dưỡng', 'giảm cân nhanh', 'tăng cân'], 'symptom_agent': ['đau', 'chấn thương', 'bị thương', 'sưng', 'viêm'], 'mental_health_agent': ['stress', 'lo âu', 'không có động lực', 'chán'], 'general_health_agent': ['khám', 'bác sĩ', 'xét nghiệm'] } self.system_prompt = """Bạn là huấn luyện viên cá nhân chuyên nghiệp, nhiệt huyết và động viên. 💪 CHUYÊN MÔN: - Tạo kế hoạch tập luyện cá nhân hóa - Tư vấn bài tập phù hợp với thể trạng, mục tiêu - Hướng dẫn kỹ thuật tập an toàn - Tư vấn tập cho người có bệnh nền - Lịch tập gym, tập tại nhà, cardio, yoga... 🎯 CÁCH TƯ VẤN: 1. **KIỂM TRA THÔNG TIN TRƯỚC KHI HỎI:** - ĐỌC KỸ chat history - user có thể đã cung cấp thông tin rồi! - Nếu user đã nói "tôi 30 tuổi, nam, muốn giảm cân, có thể tập 45 phút/ngày" → ĐỪNG HỎI LẠI! - Chỉ hỏi thông tin THỰC SỰ còn thiếu - Nếu đã đủ thông tin cơ bản → TẠO LỊCH TẬP NGAY! 2. **THÔNG TIN CẦN THIẾT:** - Cơ bản: Tuổi, giới tính, mục tiêu, thời gian rảnh - Bổ sung: Thể lực, dụng cụ có sẵn, bệnh nền - Nếu thiếu → Hỏi ngắn gọn, không hỏi mãi 3. **TẠO LỊCH TẬP:** - Lịch tập cụ thể theo ngày - Giải thích TẠI SAO tập bài này - Hướng dẫn progression (tuần 1, 2, 3...) - Lưu ý an toàn, tránh chấn thương ⚠️ AN TOÀN: - Người có bệnh tim, huyết áp → khuyên gặp bác sĩ trước - Người có chấn thương → tập nhẹ, tránh vùng bị thương - Người mới bắt đầu → từ từ, không quá sức 💬 PHONG CÁCH: - Động viên, khích lệ 💪🔥 - Thực tế, không lý thuyết suông - Dễ hiểu, dễ làm theo - Hài hước nhẹ nhàng - TỰ NHIÊN, MẠCH LẠC - không lặp lại ý, không copy-paste câu từ context khác - Nếu hỏi thông tin → Hỏi NGẮN GỌN, TRỰC TIẾP - KHÔNG dùng câu như "Bạn thử làm theo xem có đỡ không" (đây là câu của bác sĩ, không phải PT!)""" def set_health_context(self, health_context: HealthContext): """Inject health context and initialize fitness tracker""" self.health_context = health_context self.fitness_tracker = FitnessTracker(health_context) def handle(self, parameters, chat_history=None): """ Handle exercise request Args: parameters (dict): { "user_query": str, "user_data": dict (optional) } chat_history (list): Conversation history Returns: str: Response message """ user_query = parameters.get("user_query", "") user_data = parameters.get("user_data", {}) # Extract and save user info from current message immediately self.extract_and_save_user_info(user_query) # Update memory from chat history if chat_history: self.update_memory_from_history(chat_history) # Check if we should hand off to another agent if self.should_handoff(user_query, chat_history): next_agent = self.suggest_next_agent(user_query) if next_agent: # Save current exercise data for next agent self.save_agent_data('last_exercise_advice', { 'query': user_query, 'user_profile': self.get_user_profile(), 'timestamp': datetime.now().isoformat() }) # Check if nutrition agent shared data with us nutrition_data = self.get_other_agent_data('nutrition_agent', 'nutrition_plan') context = self._generate_exercise_summary(nutrition_data) return self.create_handoff_message(next_agent, context, user_query) # Use health context if available if self.health_context: profile = self.health_context.get_user_profile() user_data = { 'age': profile.age, 'gender': profile.gender, 'weight': profile.weight, 'height': profile.height, 'fitness_level': profile.fitness_level, 'activity_level': profile.activity_level, 'health_conditions': profile.health_conditions } # Extract user data from chat history if not provided elif not user_data and chat_history: user_data = self._extract_user_data_from_history(chat_history) # Save extracted data to shared memory for other agents for key, value in user_data.items(): if value is not None: self.update_user_profile(key, value) # Check if we have enough data - check shared memory first profile = self.get_user_profile() for field in ['age', 'gender', 'weight', 'height']: if not user_data.get(field) and profile.get(field): user_data[field] = profile[field] missing_fields = self._check_missing_data(user_data) if missing_fields: return self._ask_for_missing_data(missing_fields, user_data) # Generate exercise plan try: plan = generate_exercise_plan(user_data) # Adjust difficulty based on fitness tracker if self.fitness_tracker: metrics = self.fitness_tracker.calculate_progress_metrics() if metrics.get('adherence', 0) > 0.8: plan = self.fitness_tracker.adjust_difficulty(plan, 'increase') elif metrics.get('adherence', 0) < 0.5: plan = self.fitness_tracker.adjust_difficulty(plan, 'decrease') response = plan # Persist workout plan to health context if self.health_context: self.health_context.add_health_record('exercise', { 'query': user_query, 'plan': response, 'user_data': user_data, 'timestamp': datetime.now().isoformat() }) return response except Exception as e: return self._handle_error(e, user_query) def _extract_user_data_from_history(self, chat_history): """Extract user data from conversation history""" user_data = { 'age': None, 'gender': None, 'weight': None, 'height': None, 'fitness_level': 'beginner', 'goal': 'health_improvement', 'available_time': 30, 'health_conditions': [] } all_messages = " ".join([msg[0] for msg in chat_history if msg[0]]) # Extract age age_match = re.search(r'(\d+)\s*tuổi|tuổi\s*(\d+)|tôi\s*(\d+)', all_messages.lower()) if age_match: user_data['age'] = int([g for g in age_match.groups() if g][0]) # Extract gender if re.search(r'\bnam\b|male|đàn ông', all_messages.lower()): user_data['gender'] = 'male' elif re.search(r'\bnữ\b|female|đàn bà', all_messages.lower()): user_data['gender'] = 'female' # Extract fitness level if re.search(r'mới bắt đầu|beginner|chưa tập', all_messages.lower()): user_data['fitness_level'] = 'beginner' elif re.search(r'trung bình|intermediate|tập được', all_messages.lower()): user_data['fitness_level'] = 'intermediate' elif re.search(r'nâng cao|advanced|tập lâu', all_messages.lower()): user_data['fitness_level'] = 'advanced' # Extract goal if re.search(r'giảm cân|weight loss|slim', all_messages.lower()): user_data['goal'] = 'weight_loss' elif re.search(r'tăng cân|weight gain|bulk', all_messages.lower()): user_data['goal'] = 'weight_gain' elif re.search(r'tập gym|muscle|cơ bắp|tăng cơ', all_messages.lower()): user_data['goal'] = 'muscle_building' elif re.search(r'khỏe mạnh|health|sức khỏe', all_messages.lower()): user_data['goal'] = 'health_improvement' # Extract available time time_match = re.search(r'(\d+)\s*phút|(\d+)\s*tiếng', all_messages.lower()) if time_match: time_val = int([g for g in time_match.groups() if g][0]) if 'tiếng' in all_messages.lower(): time_val *= 60 user_data['available_time'] = time_val return user_data def _check_missing_data(self, user_data): """Check what data is missing""" required = ['age', 'gender', 'fitness_level', 'goal'] return [field for field in required if not user_data.get(field)] def _ask_for_missing_data(self, missing_fields, current_data): """Ask for missing data""" questions = { 'age': "bạn bao nhiêu tuổi", 'gender': "bạn là nam hay nữ", 'fitness_level': "thể lực hiện tại của bạn thế nào (mới bắt đầu/trung bình/nâng cao)", 'goal': "mục tiêu của bạn là gì (giảm cân/tăng cơ/khỏe mạnh hơn)" } q_list = [questions[f] for f in missing_fields] if len(q_list) == 1: question = q_list[0] elif len(q_list) == 2: question = f"{q_list[0]} và {q_list[1]}" else: question = ", ".join(q_list[:-1]) + f" và {q_list[-1]}" return f"""💪 **Để tạo lịch tập phù hợp, mình cần biết thêm:** Cho mình biết {question} nhé? 💡 **Ví dụ:** "Tôi 30 tuổi, nam, mới bắt đầu tập, muốn giảm cân, có thể tập 45 phút mỗi ngày" Sau khi có đủ thông tin, mình sẽ tạo kế hoạch tập luyện 7 ngày chi tiết cho bạn! 🔥""" def _handle_general_exercise_query(self, user_query, chat_history): """Handle general exercise questions using LLM + RAG""" from config.settings import client, MODEL try: # Smart RAG - only query when needed (inherit from BaseAgent) rag_answer = '' rag_sources = [] if self.should_use_rag(user_query, chat_history): rag_result = self.rag.query_exercise(user_query) rag_answer = rag_result.get('answer', '') rag_sources = rag_result.get('source_docs', []) # Build conversation context with RAG context rag_context = f"Dựa trên kiến thức từ cơ sở dữ liệu:\n{rag_answer}\n\n" if rag_answer else "" messages = [{"role": "system", "content": self.system_prompt}] # Add RAG context if available if rag_context: messages.append({"role": "system", "content": f"Thông tin tham khảo từ cơ sở dữ liệu:\n{rag_context}"}) # Add chat history (last 5 exchanges) if chat_history: recent_history = chat_history[-5:] if len(chat_history) > 5 else chat_history for user_msg, bot_msg in recent_history: if user_msg: messages.append({"role": "user", "content": user_msg}) if bot_msg: messages.append({"role": "assistant", "content": bot_msg}) # Add current query messages.append({"role": "user", "content": user_query}) # Get LLM response response = client.chat.completions.create( model=MODEL, messages=messages, temperature=0.7, max_tokens=500 ) llm_response = response.choices[0].message.content # Add sources using RAG integration formatter (FIXED!) if rag_sources: formatted_response = self.rag.format_response_with_sources({ 'answer': llm_response, 'source_docs': rag_sources }) return formatted_response return llm_response except Exception as e: return f"""Xin lỗi, mình gặp lỗi kỹ thuật. Bạn có thể: 1. Thử lại câu hỏi 2. Hoặc hỏi mình về chủ đề sức khỏe khác nhé! 💙 Chi tiết lỗi: {str(e)}""" def should_handoff(self, user_query: str, chat_history: Optional[List] = None) -> bool: """ Override base method - Determine if should hand off to another agent Specific triggers for exercise agent: - User asks about nutrition/diet - User mentions pain/injury - User asks about mental health """ query_lower = user_query.lower() # Check each agent's triggers for agent, triggers in self.handoff_triggers.items(): if any(trigger in query_lower for trigger in triggers): # Don't handoff if we're in the middle of exercise planning if chat_history and self._is_mid_planning(chat_history): return False return True return False def suggest_next_agent(self, user_query: str) -> Optional[str]: """Override base method - Suggest which agent to hand off to""" query_lower = user_query.lower() # Priority order for handoff if any(trigger in query_lower for trigger in self.handoff_triggers.get('symptom_agent', [])): return 'symptom_agent' if any(trigger in query_lower for trigger in self.handoff_triggers.get('nutrition_agent', [])): return 'nutrition_agent' if any(trigger in query_lower for trigger in self.handoff_triggers.get('mental_health_agent', [])): return 'mental_health_agent' if any(trigger in query_lower for trigger in self.handoff_triggers.get('general_health_agent', [])): return 'general_health_agent' return None def _is_mid_planning(self, chat_history: List) -> bool: """Check if we're in the middle of exercise planning""" if not chat_history or len(chat_history) < 2: return False # Check last bot response last_bot_response = chat_history[-1][1] if len(chat_history[-1]) > 1 else "" # If we just asked for user data, don't handoff if any(phrase in last_bot_response for phrase in [ "tuổi", "giới tính", "mục tiêu", "thời gian", "dụng cụ" ]): return True return False def _generate_exercise_summary(self, nutrition_data=None) -> str: """Generate summary of exercise advice for handoff""" exercise_data = self.get_agent_data('exercise_plan') user_profile = self.get_user_profile() # Natural summary without robotic prefix summary_parts = [] if exercise_data and isinstance(exercise_data, dict): if 'goal' in exercise_data: summary_parts.append(f"Mục tiêu: {exercise_data['goal']}") if 'frequency' in exercise_data: summary_parts.append(f"Tần suất: {exercise_data['frequency']}") # Include nutrition data if available (agent-to-agent communication) if nutrition_data and isinstance(nutrition_data, dict): if 'daily_targets' in nutrition_data: targets = nutrition_data['daily_targets'] summary_parts.append(f"Calo: {targets.get('calories', 'N/A')} kcal/ngày") if user_profile and user_profile.get('fitness_level'): summary_parts.append(f"Thể lực: {user_profile['fitness_level']}") return " | ".join(summary_parts)[:100] if summary_parts else "" def _handle_error(self, error, user_query): """Handle errors gracefully""" return f"""Xin lỗi, mình gặp chút vấn đề khi tạo lịch tập. 😅 Lỗi: {str(error)} Bạn có thể thử: 1. Cung cấp lại thông tin: tuổi, giới tính, thể lực, mục tiêu 2. Hỏi câu hỏi cụ thể hơn về tập luyện 3. Hoặc mình có thể tư vấn về chủ đề sức khỏe khác Bạn muốn thử lại không? 💙"""