File size: 3,803 Bytes
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 | import os
import uuid
import datetime
from supabase import create_client, Client
from langchain_core.tools import tool
from src.utils.logger import setup_logger
logger = setup_logger("FHIRPatientMemory")
def _get_client() -> Client:
url = os.getenv("SUPABASE_URL")
key = os.getenv("SUPABASE_KEY")
return create_client(url, key)
@tool
def save_patient_memory(patient_id: str, glucose_history: str = None, medications: str = None, diet: str = None):
"""
Save patient memory (narrative context) as strict FHIR Observation resources.
"""
logger.info(f"Saving FHIR narrative memory for patient: {patient_id}")
client = _get_client()
results = []
# Map narrative categories to FHIR-like coding
categories = {
"glucose_history": {"code": "narrative-glucose", "display": "Glucose History Narrative"},
"medications": {"code": "narrative-meds", "display": "Medications Narrative"},
"diet": {"code": "narrative-diet", "display": "Dietary Narrative"}
}
for key, value in [("glucose_history", glucose_history), ("medications", medications), ("diet", diet)]:
if value is not None:
obs_id = str(uuid.uuid4())
fhir_obs = {
"resourceType": "Observation",
"id": obs_id,
"status": "final",
"code": {
"coding": [{
"system": "http://dm-chatbot.ai/codes",
"code": categories[key]["code"],
"display": categories[key]["display"]
}]
},
"subject": {"reference": f"Patient/{patient_id}"},
"effectiveDateTime": datetime.datetime.now(datetime.timezone.utc).isoformat(),
"valueString": value # Using valueString for narrative text
}
data = {
"id": obs_id,
"patient_id": patient_id,
"resource": fhir_obs,
"last_updated": datetime.datetime.now(datetime.timezone.utc).isoformat()
}
client.table("observations").insert(data).execute()
results.append(key)
return f"Successfully saved FHIR narrative memory for: {', '.join(results)}"
@tool
def get_patient_memory(patient_id: str):
"""
Retrieve patient narrative memory from FHIR Observation resources.
"""
logger.info(f"Retrieving FHIR narrative memory for patient: {patient_id}")
client = _get_client()
try:
# Fetch observations with narrative codes
response = client.table("observations").select("resource").eq("patient_id", patient_id).execute()
observations = [r["resource"] for r in response.data]
# Filter for narrative codes
memory = {}
codes_map = {
"narrative-glucose": "Glucose History",
"narrative-meds": "Medications",
"narrative-diet": "Diet"
}
for obs in observations:
for coding in obs.get("code", {}).get("coding", []):
code = coding.get("code")
if code in codes_map:
# Keep only the latest one for each category
memory[codes_map[code]] = obs.get("valueString", "N/A")
if not memory:
return f"No FHIR narrative memory found for patient {patient_id}."
output = f"Patient Narrative Memory (FHIR) for {patient_id}:\n"
for cat, val in memory.items():
output += f"- {cat}: {val}\n"
return output
except Exception as e:
logger.error(f"Error retrieving FHIR memory: {e}")
return f"Error retrieving FHIR memory: {str(e)}"
|