File size: 8,805 Bytes
76962bf b1198f0 84ae02f 76962bf b1198f0 76962bf b1198f0 76962bf b1198f0 76962bf b1198f0 76962bf b1198f0 76962bf 84ae02f b1198f0 76962bf b1198f0 76962bf b1198f0 76962bf b1198f0 76962bf b1198f0 76962bf 84ae02f b1198f0 76962bf b1198f0 76962bf b1198f0 76962bf b1198f0 76962bf | 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 190 191 192 193 194 195 196 197 198 199 200 201 202 | import json
import time
from datetime import datetime, timedelta
from src.agents.agents import BaseAgent, throttle_agent
from src.core.state import AgentState
from src.tools.fhir_memory import get_observations_by_patient, get_medications_by_patient
from langchain_core.messages import SystemMessage, HumanMessage
from src.utils.logger import setup_logger
logger = setup_logger("CDMAgents")
class HealthCoach(BaseAgent):
def __init__(self):
fallback_prompt = """You are a proactive Chronic Disease Management (CDM) Health Coach.
Your goal is to help patients manage their conditions (like Diabetes or Hypertension) through motivation, education, and lifestyle tracking.
Always review their recent FHIR observations, current medications, and trend analysis to provide personalized advice.
Consider the patient's medication regimen when making lifestyle recommendations.
Be encouraging but firm about safety guidelines."""
super().__init__(fallback_prompt, "HealthCoach.txt")
class TrendAnalyzer(BaseAgent):
def __init__(self):
fallback_prompt = """You are a medical data trend analyzer.
You analyze FHIR observations and identify clinically significant trends.
If you see rising glucose levels or blood pressure, flag them immediately.
Provide a concise summary of the last 7-14 days of data with trend projections."""
super().__init__(fallback_prompt, "TrendAnalyzer.txt")
def _compute_trend_statistics(self, values: list) -> dict:
"""
Compute statistical trend indicators using polynomial regression.
Bug 7.2: Enhanced trend analysis with proper windowing.
Returns:
dict: Contains trend direction, slope, R-squared, and projection
"""
if len(values) < 2:
return {"trend": "insufficient", "slope": 0, "r_squared": 0, "projection": None}
try:
import numpy as np
# Prepare data points
x = np.array(range(len(values)))
y = np.array(values)
# First-order polynomial fit (linear regression)
if len(values) >= 3:
coeffs = np.polyfit(x, y, 1)
slope = coeffs[0]
intercept = coeffs[1]
# Calculate R-squared
y_pred = np.polyval(coeffs, x)
ss_res = np.sum((y - y_pred) ** 2)
ss_tot = np.sum((y - np.mean(y)) ** 2)
r_squared = 1 - (ss_res / ss_tot) if ss_tot != 0 else 0
# Determine trend with threshold
if abs(slope) < 0.5:
trend = "stable"
elif slope > 0.5:
trend = "rising"
else:
trend = "falling"
# Project next value
next_x = len(values)
projection = np.polyval(coeffs, next_x)
return {
"trend": trend,
"slope": float(slope),
"r_squared": float(r_squared),
"projection": float(projection),
"change_rate": float(slope)
}
else:
# For < 3 points, use simple delta
slope = (values[-1] - values[0]) / (len(values) - 1)
trend = "rising" if slope > 0.5 else "falling" if slope < -0.5 else "stable"
return {
"trend": trend,
"slope": float(slope),
"r_squared": 0,
"projection": values[-1] + slope,
"change_rate": float(slope)
}
except ImportError:
logger.warning("NumPy not available, using basic trend analysis")
# Fallback: basic two-point comparison
if len(values) >= 2:
delta = values[-1] - values[-2]
trend = "rising" if delta > 0.5 else "falling" if delta < -0.5 else "stable"
return {
"trend": trend,
"slope": float(delta),
"r_squared": 0,
"projection": values[-1] + delta,
"change_rate": float(delta)
}
return {"trend": "insufficient", "slope": 0, "r_squared": 0, "projection": None}
async def analyze_trends(self, patient_id: str):
"""
Logic to analyze trends for a specific patient with windowing.
Bug 7.2: Implements 7-14 day moving window analysis with regression.
"""
logger.info(f"Analyzing health trends for patient: {patient_id}")
observations = get_observations_by_patient.invoke({"patient_id": patient_id})
if isinstance(observations, str):
return observations, {}
if not observations:
return "No observations available for trend analysis.", {}
# Group by LOINC code and sort by date
data_points = {}
for obs in observations:
try:
code = obs["code"]["coding"][0]["display"]
val = float(obs["valueQuantity"]["value"])
date = obs["effectiveDateTime"]
if code not in data_points:
data_points[code] = []
data_points[code].append({"value": val, "date": date})
except (KeyError, IndexError, ValueError) as e:
logger.warning(f"Error parsing observation: {e}")
continue
# Sort each metric's points by date (descending for most recent first)
for code in data_points:
data_points[code].sort(key=lambda x: x["date"], reverse=True)
# Comprehensive trend analysis
analysis = "**Trend Analysis Report (7-14 Day Window)**\n\n"
structured_data = {}
for code, points in data_points.items():
if len(points) >= 2:
# Extract values for statistical analysis
values = [p["value"] for p in points]
stats = self._compute_trend_statistics(values)
latest = values[0]
prev = values[1] if len(values) > 1 else values[0]
delta = latest - prev
analysis += f"**{code}**\n"
analysis += f"- Latest: {latest:.1f}\n"
analysis += f"- Trend: {stats['trend'].upper()} (slope: {stats['slope']:.2f}/day)\n"
if stats['r_squared'] > 0:
analysis += f"- Trend Quality (R²): {stats['r_squared']:.2f}\n"
if stats['projection'] is not None:
analysis += f"- Projected Next: {stats['projection']:.1f}\n"
# Clinical alert for high-priority metrics
if code.lower() in ["blood glucose", "glucose"] and latest > 250:
analysis += f" ⚠️ **ALERT**: Glucose critically high\n"
elif code.lower() in ["blood glucose", "glucose"] and latest < 70:
analysis += f" ⚠️ **ALERT**: Risk of hypoglycemia\n"
# Add to structured data
structured_data[code] = {
"points": [{"date": p["date"], "value": p["value"]} for p in points],
"stats": stats
}
analysis += "\n"
elif len(points) == 1:
analysis += f"**{code}**: Single data point (latest: {points[0]['value']:.1f}) — insufficient for trend\n\n"
return analysis, structured_data
@throttle_agent
async def run(self, state: AgentState, config=None, **kwargs):
# Extract patient ID from state
patient_id = state.get("patient_id", "unknown")
if patient_id == "unknown":
logger.warning("TrendAnalyzer: No patient ID found in state.")
return {"logs": ["TrendAnalyzer: No patient ID found in state."]}
start_time = time.time()
analysis, structured_data = await self.analyze_trends(patient_id)
end_time = time.time()
metrics = {
"agent": "TrendAnalyzer",
"tokens": 0, # Logic-based, no LLM call
"time": round(end_time - start_time, 3)
}
return {
"trend_analysis": analysis,
"trend_data": structured_data,
"logs": [f"TrendAnalyzer: Completed analysis for {patient_id}"],
"metrics": [metrics]
}
|