Spaces:
Runtime error
Runtime error
File size: 7,219 Bytes
76166e3 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 |
from typing import Dict, List
from .base_agent import BaseWellnessAgent
import json
from pathlib import Path
class AssessmentAgent(BaseWellnessAgent):
"""Agent specialized in mental health assessments"""
def __init__(self, model_config: Dict, verbose: bool = False):
super().__init__(
name="Mental Health Assessment Agent",
role="Mental Health Evaluator",
goal="Conduct thorough mental health assessments and provide actionable insights",
backstory="""I am an AI agent specialized in mental health assessment.
I use validated assessment tools and techniques to evaluate mental well-being
and provide personalized recommendations.""",
tools=["emotion_detection", "conversation"],
model_config=model_config,
verbose=verbose
)
self.assessment_tools = self._load_assessment_tools()
self.current_assessment = None
def _load_assessment_tools(self) -> Dict:
"""Load assessment questionnaires and scoring guides"""
tools_path = Path(__file__).parent.parent / "knowledge_base" / "assessment_tools.json"
try:
with open(tools_path) as f:
return json.load(f)
except FileNotFoundError:
return {
"phq9": {
"name": "PHQ-9 Depression Scale",
"questions": [
"Little interest or pleasure in doing things?",
"Feeling down, depressed, or hopeless?",
# Add more questions...
],
"scoring": {
"ranges": {
"0-4": "Minimal depression",
"5-9": "Mild depression",
"10-14": "Moderate depression",
"15-19": "Moderately severe depression",
"20-27": "Severe depression"
}
}
},
"gad7": {
"name": "GAD-7 Anxiety Scale",
"questions": [
"Feeling nervous, anxious, or on edge?",
"Not being able to stop or control worrying?",
# Add more questions...
],
"scoring": {
"ranges": {
"0-4": "Minimal anxiety",
"5-9": "Mild anxiety",
"10-14": "Moderate anxiety",
"15-21": "Severe anxiety"
}
}
}
}
def start_assessment(self, assessment_type: str) -> Dict:
"""Start a new assessment"""
if assessment_type not in self.assessment_tools:
raise ValueError(f"Unknown assessment type: {assessment_type}")
self.current_assessment = {
"type": assessment_type,
"tool": self.assessment_tools[assessment_type],
"responses": [],
"current_question": 0
}
return self.get_next_question()
def process_message(self, message: str) -> Dict:
"""Process user response and continue assessment"""
if not self.current_assessment:
return self.format_response(
"No active assessment. Please start an assessment first."
)
# Record response
self.current_assessment["responses"].append({
"question": self.current_assessment["current_question"],
"response": message,
"emotion": self.analyze_emotion(message)
})
# Move to next question or finish assessment
return (
self.finish_assessment()
if self._is_assessment_complete()
else self.get_next_question()
)
def get_next_question(self) -> Dict:
"""Get the next question in the assessment"""
if not self.current_assessment:
return self.format_response(
"No active assessment. Please start an assessment first."
)
tool = self.current_assessment["tool"]
question_idx = self.current_assessment["current_question"]
if question_idx >= len(tool["questions"]):
return self.finish_assessment()
question = tool["questions"][question_idx]
self.current_assessment["current_question"] += 1
return self.format_response(question)
def _is_assessment_complete(self) -> bool:
"""Check if all questions have been answered"""
if not self.current_assessment:
return False
return len(self.current_assessment["responses"]) >= len(
self.current_assessment["tool"]["questions"]
)
def finish_assessment(self) -> Dict:
"""Complete the assessment and generate report"""
if not self.current_assessment:
return self.format_response(
"No active assessment to finish."
)
# Calculate scores
scores = self._calculate_scores()
# Generate report
report = self._generate_report(scores)
# Add to history
self.add_to_history({
"assessment_type": self.current_assessment["type"],
"scores": scores,
"report": report
})
# Reset current assessment
self.current_assessment = None
return self.format_response(report)
def _calculate_scores(self) -> Dict:
"""Calculate assessment scores"""
# Implement scoring logic based on assessment type
return {
"total_score": 0, # Calculate actual score
"subscores": {},
"severity": "Unknown" # Determine from scoring ranges
}
def _generate_report(self, scores: Dict) -> str:
"""Generate detailed assessment report"""
report = f"""Assessment Report
Type: {self.current_assessment["type"].upper()}
Total Score: {scores["total_score"]}
Severity: {scores["severity"]}
Recommendations:
1. Continue monitoring your mental health
2. Consider discussing results with a mental health professional
3. Practice self-care and stress management techniques
Note: This assessment is for screening purposes only and does not constitute
a clinical diagnosis. Please consult with a qualified mental health professional
for proper evaluation and treatment."""
return report
def get_assessment_history(self) -> List[Dict]:
"""Get history of completed assessments"""
return self.get_history()
def get_available_assessments(self) -> List[str]:
"""Get list of available assessment tools"""
return list(self.assessment_tools.keys()) |