Spaces:
Running on Zero
Running on Zero
File size: 11,218 Bytes
90fa9aa | 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 189 | # NeuroSense AI — Deep Clinical Evaluation & Multimodal Stress Assessment via Google Gemini
import os
import json
import re
from pathlib import Path
from src.assistant.guardrails import is_unrelated_to_mental_health, UNRELATED_PROJECT_REPLY
try:
from openai import OpenAI
OPENAI_AVAILABLE = True
except ImportError:
OPENAI_AVAILABLE = False
class GeminiClinicalEvaluator:
"""
Multimodal Clinical Stress Evaluator powered by Google Gemini (Free Tier) / Groq / OpenAI.
Provides deep psychological assessment, biomarker interpretation, and CBT risk evaluation
for both Voice/Speech transcripts and Narrative text inputs.
"""
def __init__(self):
self.provider = "Offline Rule-based Evaluator"
self.model_name = "local-clinical-engine"
self.client = None
# Check environment variables
self.gemini_key = os.getenv("GEMINI_API_KEY", "").strip()
self.groq_key = os.getenv("GROQ_API_KEY", "").strip()
self.openai_key = os.getenv("OPENAI_API_KEY", "").strip()
if OPENAI_AVAILABLE and self.groq_key and len(self.groq_key) > 5:
try:
self.client = OpenAI(api_key=self.groq_key, base_url="https://api.groq.com/openai/v1")
self.provider = "Groq (Llama 3.3 70B)"
self.model_name = "llama-3.3-70b-versatile"
print(f"[Gemini Evaluator] Connected to {self.provider}.")
except Exception as e:
print(f"[Gemini Evaluator] Could not initialize Groq client: {e}")
elif OPENAI_AVAILABLE and self.gemini_key and len(self.gemini_key) > 5:
try:
self.client = OpenAI(api_key=self.gemini_key, base_url="https://generativelanguage.googleapis.com/v1beta/openai/")
self.provider = "Google Gemini (Free Tier)"
self.model_name = "gemini-flash-latest"
print(f"[Gemini Evaluator] Connected to 100% FREE AI: {self.provider}.")
except Exception as e:
print(f"[Gemini Evaluator] Could not initialize Gemini client: {e}")
elif OPENAI_AVAILABLE and self.openai_key and len(self.openai_key) > 5:
try:
self.client = OpenAI(api_key=self.openai_key)
self.provider = "OpenAI ChatGPT"
self.model_name = "gpt-4o-mini"
print(f"[Gemini Evaluator] Connected to {self.provider}.")
except Exception as e:
print(f"[Gemini Evaluator] Could not initialize OpenAI client: {e}")
def evaluate_clinical_stress(self, text: str, modality: str = "Text Narrative", acoustic_features: dict = None) -> dict:
"""
Analyzes a person's stress, depression, anxiety, and coping risk using Google Gemini AI.
Works for both transcribed voice recordings and written text narratives.
Enforces strict guardrails to only evaluate inputs related to mental health and emotional well-being.
"""
if not text or len(text.strip()) < 3:
return self._get_fallback_evaluation("Patient provided minimal or silent input.", modality, 20, "Low / Baseline Calm")
# 1. Check if the query is clearly unrelated to mental health / stress / daily life
if is_unrelated_to_mental_health(text):
return {
"ai_provider": f"{self.provider} (Guardrail Filter)",
"stress_level_index": 0,
"clinical_risk_tier": "Not Related to Project",
"detected_symptoms": ["Unrelated Query / Topic"],
"empathetic_clinical_summary": UNRELATED_PROJECT_REPLY,
"recommended_intervention": "Please submit a text check-in or question directly related to your mental health, emotional state, or daily stress."
}
if self.client and self.model_name:
models_to_try = [self.model_name]
if "gemini" in self.model_name.lower():
models_to_try = ["gemini-flash-latest", "gemini-1.5-flash-latest", "gemma-4-31b-it"]
prompt = (
f"You are an expert dual-modality clinical psychologist and Cognitive Behavioral Therapy (CBT) diagnostician. "
f"Analyze the following patient input from a {modality} check-in:\n\n"
f"PATIENT INPUT: \"{text}\"\n\n"
f"CRITICAL GUARDRAIL: If the patient input is completely unrelated to mental health, stress, anxiety, depression, relationships, daily pressure, or emotional well-being (e.g. coding questions, general trivia, math homework, recipes, etc.), return exactly:\n"
f"{{\n"
f" \"stress_level_index\": 0,\n"
f" \"clinical_risk_tier\": \"Not Related to Project\",\n"
f" \"detected_symptoms\": [\"Unrelated Query / Topic\"],\n"
f" \"empathetic_clinical_summary\": \"{UNRELATED_PROJECT_REPLY}\",\n"
f" \"recommended_intervention\": \"Please submit a text check-in or question directly related to your mental health, emotional state, or daily stress.\"\n"
f"}}\n\n"
f"Otherwise, evaluate the emotional state, stress severity (0-100), presence of anxiety/depression or burnout indicators, and provide actionable psychological recommendations.\n"
f"Respond ONLY with a valid JSON object containing exactly these keys:\n"
f"- \"stress_level_index\": (integer between 0 and 100 representing psychological strain)\n"
f"- \"clinical_risk_tier\": (string, one of: \"Low / Baseline Calm\", \"Mild Stress & Fatigue\", \"Moderate Anxiety & Burnout Risk\", \"Severe Emotional Distress / High Risk\")\n"
f"- \"detected_symptoms\": (array of 3 to 5 specific psychological/symptomatic strings detected or inferred, e.g. [\"Sleep Deprivation\", \"Catastrophizing\", \"Workload Overwhelm\"])\n"
f"- \"empathetic_clinical_summary\": (string under 80 words explaining the clinical diagnosis and underlying stressors with professional empathy)\n"
f"- \"recommended_intervention\": (string under 60 words recommending exact CBT or grounding exercises tailored to this subject)"
)
for model_id in models_to_try:
try:
response = self.client.chat.completions.create(
model=model_id,
messages=[
{"role": "system", "content": "You are a clinical AI stress evaluation assistant. Output only JSON."},
{"role": "user", "content": prompt}
],
temperature=0.4,
max_tokens=400,
timeout=5.5
)
raw_content = response.choices[0].message.content.strip()
# Clean markdown code block if present
if raw_content.startswith("```json"):
raw_content = raw_content[7:]
if raw_content.startswith("```"):
raw_content = raw_content[3:]
if raw_content.endswith("```"):
raw_content = raw_content[:-3]
parsed = json.loads(raw_content.strip())
parsed["ai_provider"] = f"{self.provider} ({model_id})"
return parsed
except Exception as e:
continue # Try next model if busy or JSON parse error
# Offline / Fallback Clinical Evaluation
return self._get_fallback_evaluation(text, modality)
def _get_fallback_evaluation(self, text: str, modality: str, custom_score: int = None, custom_tier: str = None) -> dict:
if is_unrelated_to_mental_health(text):
return {
"ai_provider": f"{self.provider} (Guardrail Filter)",
"stress_level_index": 0,
"clinical_risk_tier": "Not Related to Project",
"detected_symptoms": ["Unrelated Query / Topic"],
"empathetic_clinical_summary": UNRELATED_PROJECT_REPLY,
"recommended_intervention": "Please submit a text check-in or question directly related to your mental health, emotional state, or daily stress."
}
score = custom_score if custom_score is not None else 45
tier = custom_tier if custom_tier is not None else "Mild Stress & Fatigue"
t_lower = text.lower()
if any(w in t_lower for w in ["overwhelm", "burnout", "can't cope", "exhaust", "crisis"]):
score = max(score, 76)
tier = "Severe Emotional Distress / High Risk"
elif any(w in t_lower for w in ["deadline", "exam", "worry", "anxious", "stress", "pressure"]):
score = max(score, 58)
tier = "Moderate Anxiety & Burnout Risk"
elif any(w in t_lower for w in ["calm", "good", "relax", "happy", "fine"]):
score = min(score, 24)
tier = "Low / Baseline Calm"
symptoms = []
if any(w in t_lower for w in ["sleep", "insomnia", "tired", "exhausted"]):
symptoms.append("Sleep Disturbance & Fatigue")
if any(w in t_lower for w in ["deadline", "exam", "job", "work", "class", "study", "project"]):
symptoms.append("Occupational / Academic Pressure")
if any(w in t_lower for w in ["anxious", "worry", "panic", "fear", "scared"]):
symptoms.append("Acute Anxiety Symptoms")
if any(w in t_lower for w in ["depress", "lonely", "hopeless", "sad", "cry"]):
symptoms.append("Depressive Affect & Low Mood")
if not symptoms:
symptoms = ["Cognitive Load", "Daily Routine Adaptation"]
summary = (
f"Analysis of this {modality} check-in indicates {tier.lower()} (Index: {score}/100). "
f"The subject exhibits signs of {' and '.join([s.lower() for s in symptoms[:2]]) if len(symptoms)>=2 else symptoms[0].lower()}. "
"Empathetic grounding and cognitive reappraisal are advised to mitigate stress accumulation."
)
intervention = "Practice structured cognitive reframing, challenge negative assumptions using Socratic questioning, and break pending commitments into manageable 15-minute intervals."
return {
"ai_provider": "Offline Clinical Evaluator (Rule-Based Fallback)",
"stress_level_index": score,
"clinical_risk_tier": tier,
"detected_symptoms": symptoms,
"empathetic_clinical_summary": summary,
"recommended_intervention": intervention
}
if __name__ == "__main__":
evaluator = GeminiClinicalEvaluator()
print("Testing Voice Transcript Evaluation:")
print(json.dumps(evaluator.evaluate_clinical_stress("I feel anxious about my job deadlines and relationship worries.", "Voice Recording"), indent=2))
|