test_10_patients.py•9.25 kB
#!/usr/bin/env python3
"""
Test FDA Verification Agent on 10 patients from HAPI FHIR
"""
import asyncio
import json
import sys
import requests
from pathlib import Path
from typing import List, Dict, Any
# Add parent directory to path
sys.path.append(str(Path(__file__).parent))
from agents.fda_verification_agent import FDAVerificationAgent, PADecision
# FHIR server URL
FHIR_BASE = "https://hapi-fhir-davinci-data.apps.cluster-sdzgj.sdzgj.sandbox319.opentlc.com/fhir"
def fetch_device_requests_for_patients(patient_ids: List[str]) -> Dict[str, List[Dict]]:
"""Fetch all DeviceRequests for given patient IDs."""
patient_device_requests = {}
for patient_id in patient_ids:
url = f"{FHIR_BASE}/DeviceRequest?subject=Patient/{patient_id}"
try:
response = requests.get(url)
if response.status_code == 200:
bundle = response.json()
if bundle.get("total", 0) > 0:
device_requests = []
for entry in bundle.get("entry", []):
device_requests.append(entry["resource"])
patient_device_requests[patient_id] = device_requests
else:
patient_device_requests[patient_id] = []
else:
print(f"❌ Error fetching for {patient_id}: {response.status_code}")
patient_device_requests[patient_id] = []
except Exception as e:
print(f"❌ Exception for {patient_id}: {e}")
patient_device_requests[patient_id] = []
return patient_device_requests
async def test_patients():
"""Test FDA verification on 10 patients."""
print("=" * 80)
print("FDA Device Verification - Testing 10 Patients")
print("=" * 80)
# Test first 10 aud-patient IDs
patient_ids = [f"aud-patient-{str(i).zfill(5)}" for i in range(1, 11)]
print(f"Testing patients: {', '.join(patient_ids[:5])}")
print(f" {', '.join(patient_ids[5:])}")
print("=" * 80)
# Fetch all DeviceRequests
print("\n📋 Fetching DeviceRequests from FHIR server...")
patient_device_requests = fetch_device_requests_for_patients(patient_ids)
# Count total DeviceRequests
total_requests = sum(len(drs) for drs in patient_device_requests.values())
print(f"✅ Found {total_requests} total DeviceRequest resources across 10 patients\n")
# Initialize FDA agent
import os
api_key = os.getenv("LLAMA_API_KEY", "af78aa99d7ae5bb58f19380ee007277c")
base_url = os.getenv("LLAMA_BASE_URL",
"https://llama-4-scout-17b-16e-w4a16-maas-apicast-production.apps.prod.rhoai.rh-aiservices-bu.com:443/v1")
mcp_url = os.getenv("FDA_MCP_URL", "http://localhost:8090/mcp/")
agent = FDAVerificationAgent(
mcp_url=mcp_url,
api_key=api_key,
base_url=base_url
)
# Track results
results_summary = {
"approved": [],
"denied": [],
"pending": [],
"manual_review": [],
"no_device_requests": []
}
print("🔍 Analyzing DeviceRequests with FDA Verification Agent...")
print("-" * 80)
# Process each patient
for patient_num, patient_id in enumerate(patient_ids, 1):
device_requests = patient_device_requests[patient_id]
print(f"\n{patient_num}. Patient: {patient_id}")
if not device_requests:
print(f" ⚪ No DeviceRequest resources found")
results_summary["no_device_requests"].append(patient_id)
continue
print(f" 📦 {len(device_requests)} DeviceRequest(s) found")
for dr_num, device_request in enumerate(device_requests, 1):
# Extract device info
device_text = device_request.get("codeCodeableConcept", {}).get("text", "Unknown device")
# Check for FDA info
has_fda_info = False
fda_number = None
# Check in parameters
if "parameter" in device_request:
for param in device_request["parameter"]:
if "fda" in str(param).lower() and "valueCodeableConcept" in param:
fda_number = param["valueCodeableConcept"].get("text", "")
has_fda_info = True
# Check in notes
if "note" in device_request:
for note in device_request["note"]:
text = note.get("text", "")
if "K" in text or "P" in text:
import re
k_match = re.search(r'K\d{6}', text)
p_match = re.search(r'P\d{6}', text)
if k_match:
fda_number = k_match.group()
has_fda_info = True
elif p_match:
fda_number = p_match.group()
has_fda_info = True
print(f" └─ Device {dr_num}: {device_text[:50]}...")
if fda_number:
print(f" FDA Number: {fda_number}")
# Analyze with agent
try:
result = await agent.analyze_device_request(device_request)
decision = result.decision.value
# Display result with emoji
if decision == "approved":
emoji = "✅"
results_summary["approved"].append({
"patient": patient_id,
"device": device_text,
"fda_number": fda_number
})
elif decision == "denied":
emoji = "❌"
results_summary["denied"].append({
"patient": patient_id,
"device": device_text,
"reason": result.reasoning[:100]
})
elif decision == "pending_information":
emoji = "⚠️"
results_summary["pending"].append({
"patient": patient_id,
"device": device_text
})
else:
emoji = "🔍"
results_summary["manual_review"].append({
"patient": patient_id,
"device": device_text
})
print(f" {emoji} Decision: {decision.upper()}")
if result.fda_verification:
fda = result.fda_verification
if fda.fda_number:
print(f" FDA Status: {fda.fda_status} ({fda.fda_number})")
else:
print(f" FDA Status: {fda.fda_status}")
print(f" Confidence: {fda.confidence:.0%}")
except Exception as e:
print(f" ❌ Error: {str(e)[:100]}")
results_summary["denied"].append({
"patient": patient_id,
"device": device_text,
"reason": f"Error: {str(e)[:50]}"
})
# Print summary
print("\n" + "=" * 80)
print("📊 SUMMARY RESULTS")
print("=" * 80)
total_analyzed = sum([
len(results_summary["approved"]),
len(results_summary["denied"]),
len(results_summary["pending"]),
len(results_summary["manual_review"])
])
print(f"\n📈 Statistics:")
print(f" Total Patients: 10")
print(f" Patients with DeviceRequests: {10 - len(results_summary['no_device_requests'])}")
print(f" Total DeviceRequests Analyzed: {total_analyzed}")
print(f"\n🎯 Decision Breakdown:")
print(f" ✅ Approved: {len(results_summary['approved'])}")
print(f" ❌ Denied: {len(results_summary['denied'])}")
print(f" ⚠️ Pending Info: {len(results_summary['pending'])}")
print(f" 🔍 Manual Review: {len(results_summary['manual_review'])}")
if results_summary["approved"]:
print(f"\n✅ APPROVED DEVICES:")
for item in results_summary["approved"]:
print(f" • Patient {item['patient']}: {item['device'][:50]}")
if item['fda_number']:
print(f" FDA Number: {item['fda_number']}")
else:
print(f"\n📝 No devices were approved (as expected with synthetic data)")
if results_summary["no_device_requests"]:
print(f"\n⚪ Patients without DeviceRequests:")
print(f" {', '.join(results_summary['no_device_requests'])}")
print("\n" + "=" * 80)
print("✅ Testing Complete!")
print("=" * 80)
# Final verdict
if not results_summary["approved"]:
print("\n💡 CONCLUSION: As expected, no devices were approved because the")
print(" synthetic data contains made-up FDA numbers that don't exist")
print(" in the real FDA database.")
else:
print(f"\n🎉 SURPRISE: Found {len(results_summary['approved'])} approvable device(s)!")
print(" This means some synthetic data happened to use real FDA numbers!")
if __name__ == "__main__":
asyncio.run(test_patients())