Spaces:
Runtime error
Runtime error
| from typing import Dict, List | |
| from .base_agent import BaseWellnessAgent | |
| import json | |
| from pathlib import Path | |
| class MindfulnessAgent(BaseWellnessAgent): | |
| """Agent specialized in mindfulness and meditation guidance""" | |
| def __init__(self, model_config: Dict, verbose: bool = False): | |
| super().__init__( | |
| name="Mindfulness & Meditation Agent", | |
| role="Mindfulness Guide", | |
| goal="Guide users through mindfulness exercises and meditation sessions", | |
| backstory="""I am an AI agent specialized in mindfulness and meditation guidance. | |
| I provide calming exercises, breathing techniques, and meditation sessions | |
| tailored to users' needs and experience levels.""", | |
| tools=["emotion_detection", "conversation"], | |
| model_config=model_config, | |
| verbose=verbose | |
| ) | |
| self.exercises = self._load_exercises() | |
| self.current_session = None | |
| def _load_exercises(self) -> Dict: | |
| """Load mindfulness exercises and meditation scripts""" | |
| exercises_path = Path(__file__).parent.parent / "knowledge_base" / "resources" / "mindfulness_exercises.json" | |
| try: | |
| with open(exercises_path) as f: | |
| return json.load(f) | |
| except FileNotFoundError: | |
| return { | |
| "breathing": { | |
| "box_breathing": { | |
| "name": "Box Breathing", | |
| "description": "A simple technique to reduce stress and improve focus", | |
| "duration": 300, # 5 minutes | |
| "steps": [ | |
| "Inhale slowly for 4 counts", | |
| "Hold your breath for 4 counts", | |
| "Exhale slowly for 4 counts", | |
| "Hold for 4 counts before the next breath" | |
| ], | |
| "instructions": "Find a comfortable position. We'll practice box breathing for 5 minutes." | |
| }, | |
| "deep_breathing": { | |
| "name": "Deep Breathing", | |
| "description": "Calming deep breathing exercise", | |
| "duration": 300, | |
| "steps": [ | |
| "Take a deep breath in through your nose", | |
| "Feel your belly expand", | |
| "Exhale slowly through your mouth", | |
| "Feel your body relax" | |
| ], | |
| "instructions": "Sit comfortably with your back straight. Let's begin deep breathing." | |
| } | |
| }, | |
| "meditation": { | |
| "body_scan": { | |
| "name": "Body Scan Meditation", | |
| "description": "Progressive relaxation through body awareness", | |
| "duration": 600, # 10 minutes | |
| "steps": [ | |
| "Focus on your toes and feet", | |
| "Move attention to your legs", | |
| "Progress through torso and arms", | |
| "End with neck and head" | |
| ], | |
| "instructions": "Lie down comfortably. We'll guide you through a full body scan." | |
| }, | |
| "loving_kindness": { | |
| "name": "Loving-Kindness Meditation", | |
| "description": "Develop compassion for self and others", | |
| "duration": 600, | |
| "steps": [ | |
| "Direct love to yourself", | |
| "Extend to loved ones", | |
| "Include neutral people", | |
| "Embrace all beings" | |
| ], | |
| "instructions": "Sit in a relaxed position. We'll practice sending loving-kindness." | |
| } | |
| } | |
| } | |
| def start_session(self, session_type: str, exercise_name: str = None) -> Dict: | |
| """Start a new mindfulness session""" | |
| if session_type not in self.exercises: | |
| raise ValueError(f"Unknown session type: {session_type}") | |
| exercises = self.exercises[session_type] | |
| if exercise_name and exercise_name not in exercises: | |
| raise ValueError(f"Unknown exercise: {exercise_name}") | |
| selected_exercise = exercises[exercise_name] if exercise_name else next(iter(exercises.values())) | |
| self.current_session = { | |
| "type": session_type, | |
| "exercise": selected_exercise, | |
| "start_time": self._get_timestamp(), | |
| "current_step": 0, | |
| "completed_steps": [] | |
| } | |
| return self.format_response(self._create_session_intro()) | |
| def process_message(self, message: str) -> Dict: | |
| """Process user message during mindfulness session""" | |
| if not self.current_session: | |
| return self.format_response( | |
| "No active session. Please start a mindfulness session first." | |
| ) | |
| # Check for session control commands | |
| if message.lower() in ["pause", "stop", "end"]: | |
| return self.end_session() | |
| # Progress through exercise steps | |
| return self._progress_session() | |
| def _progress_session(self) -> Dict: | |
| """Progress through the current session's steps""" | |
| exercise = self.current_session["exercise"] | |
| current_step = self.current_session["current_step"] | |
| if current_step >= len(exercise["steps"]): | |
| return self.end_session() | |
| step = exercise["steps"][current_step] | |
| self.current_session["completed_steps"].append({ | |
| "step": current_step, | |
| "timestamp": self._get_timestamp() | |
| }) | |
| self.current_session["current_step"] += 1 | |
| return self.format_response(self._create_step_guidance(step)) | |
| def _create_session_intro(self) -> str: | |
| """Create introduction for the session""" | |
| exercise = self.current_session["exercise"] | |
| return f"""Welcome to {exercise['name']}. | |
| {exercise['description']} | |
| Duration: {exercise['duration'] // 60} minutes | |
| {exercise['instructions']} | |
| When you're ready, respond with 'begin' to start.""" | |
| def _create_step_guidance(self, step: str) -> str: | |
| """Create guidance for the current step""" | |
| return f"""{step} | |
| Take your time with this step. | |
| When you're ready to continue, send any message.""" | |
| def end_session(self) -> Dict: | |
| """End the current mindfulness session""" | |
| if not self.current_session: | |
| return self.format_response( | |
| "No active session to end." | |
| ) | |
| # Calculate session statistics | |
| stats = self._calculate_session_stats() | |
| # Generate session summary | |
| summary = self._generate_session_summary(stats) | |
| # Add to history | |
| self.add_to_history({ | |
| "session_type": self.current_session["type"], | |
| "exercise": self.current_session["exercise"]["name"], | |
| "stats": stats, | |
| "summary": summary | |
| }) | |
| # Reset current session | |
| self.current_session = None | |
| return self.format_response(summary) | |
| def _calculate_session_stats(self) -> Dict: | |
| """Calculate statistics for the completed session""" | |
| start_time = self.current_session["start_time"] | |
| end_time = self._get_timestamp() | |
| completed_steps = len(self.current_session["completed_steps"]) | |
| total_steps = len(self.current_session["exercise"]["steps"]) | |
| return { | |
| "duration": self._calculate_duration(start_time, end_time), | |
| "completion_rate": completed_steps / total_steps, | |
| "steps_completed": completed_steps | |
| } | |
| def _calculate_duration(self, start_time: str, end_time: str) -> int: | |
| """Calculate session duration in seconds""" | |
| from datetime import datetime | |
| start = datetime.fromisoformat(start_time) | |
| end = datetime.fromisoformat(end_time) | |
| return int((end - start).total_seconds()) | |
| def _generate_session_summary(self, stats: Dict) -> str: | |
| """Generate summary of the completed session""" | |
| exercise = self.current_session["exercise"] | |
| return f"""Session Complete: {exercise['name']} | |
| Duration: {stats['duration'] // 60} minutes | |
| Steps Completed: {stats['steps_completed']} of {len(exercise['steps'])} | |
| Completion Rate: {stats['completion_rate'] * 100:.1f}% | |
| Thank you for practicing mindfulness. Remember to: | |
| 1. Take these peaceful feelings with you | |
| 2. Practice regularly for best results | |
| 3. Be gentle with yourself | |
| Would you like to try another exercise?""" | |
| def get_available_exercises(self) -> Dict: | |
| """Get list of available exercises""" | |
| return { | |
| category: list(exercises.keys()) | |
| for category, exercises in self.exercises.items() | |
| } |