📚 Symptom Checkers 24 min read
How to Implement a Clinical Triage System: Complete Implementation Guide
Comprehensive guide to building production-ready clinical triage systems with ESI integration, red flag detection, multi-step assessment workflows, real-time clinician escalation, and HIPAA-compliant audit logging.
✍️
Dr. Sarah Chen
Building Production-Ready Clinical Triage Systems
Clinical triage systems represent the critical first step in patient care, determining urgency levels and appropriate care pathways that can mean the difference between life and death. Modern AI-powered triage systems combine evidence-based assessment protocols, machine learning algorithms, and intelligent workflow automation to deliver consistent, accurate triage decisions while reducing clinician burden and improving patient flow.
This comprehensive implementation guide walks through building a production-ready clinical triage system, from architectural design through deployment, including ESI (Emergency Severity Index) integration, red flag symptom detection, multi-step assessment workflows, real-time clinician escalation, mobile applications, audit logging, and HIPAA compliance.
Understanding Clinical Triage Standards
Before diving into implementation, it’s essential to understand the clinical standards that govern triage systems:
Emergency Severity Index (ESI)
The ESI is the most widely used triage system in US emergency departments, providing a five-level categorization:
ESI Level 1: Resuscitation
- Immediate life-threatening conditions
- Requires immediate intervention
- Examples: Cardiac arrest, severe trauma, respiratory failure
ESI Level 2: Emergent
- High-risk situations or severe pain/distress
- Should be seen within 10 minutes
- Examples: Chest pain, stroke symptoms, severe asthma
ESI Level 3: Urgent
- Stable with multiple resource needs
- Should be seen within 30-60 minutes
- Examples: Abdominal pain, fever with comorbidities
ESI Level 4: Less Urgent
- Stable with one resource need
- Can wait 1-2 hours
- Examples: Minor lacerations, simple urinary symptoms
ESI Level 5: Non-Urgent
- Stable with no anticipated resource needs
- Can wait several hours
- Examples: Medication refills, chronic stable conditions
Modified Early Warning Score (MEWS)
MEWS uses vital signs to identify deteriorating patients:
Vital Sign Scoring:
- Respiratory Rate: <9 (3), 9-14 (0), 15-20 (1), 21-29 (2), ≥30 (3)
- Heart Rate: <40 (3), 41-50 (1), 51-100 (0), 101-110 (1), 111-129 (2), ≥130 (3)
- Systolic BP: <70 (3), 71-80 (2), 81-100 (1), 101-199 (0), ≥200 (3)
- Temperature: <35°C (3), 35-38.4 (0), ≥38.5 (2)
- AVPU Score: Alert (0), Voice (1), Pain (2), Unresponsive (3)
Total Score Interpretation:
- 0-1: Low risk
- 2-3: Medium risk (increased monitoring)
- 4+: High risk (immediate clinical review)System Architecture
A production clinical triage system requires these integrated components:
┌──────────────────────────────────────────────────────────────────┐
│ Patient Interface Layer │
│ (Mobile App, Web Portal, Kiosk, Telehealth Integration) │
└────────────────────────┬─────────────────────────────────────────┘
│
┌────────────────────────┴─────────────────────────────────────────┐
│ Assessment Workflow Engine │
│ (Multi-Step Questionnaires, Adaptive Branching Logic) │
└────────────────────────┬─────────────────────────────────────────┘
│
┌────────────────────────┴─────────────────────────────────────────┐
│ AI-Powered Triage Classification │
│ (ESI Level Prediction, Urgency Scoring, Risk Assessment) │
└────────────────────────┬─────────────────────────────────────────┘
│
┌────────────────────────┴─────────────────────────────────────────┐
│ Red Flag Detection System │
│ (Life-Threatening Symptom Recognition, Immediate Alerts) │
└────────────────────────┬─────────────────────────────────────────┘
│
┌────────────────────────┴─────────────────────────────────────────┐
│ Care Pathway Recommendation Engine │
│ (ED, Urgent Care, Primary Care, Telehealth, Self-Care) │
└────────────────────────┬─────────────────────────────────────────┘
│
┌────────────────────────┴─────────────────────────────────────────┐
│ Real-Time Clinician Escalation │
│ (Nurse Override, Physician Consultation, Video Assessment) │
└────────────────────────┬─────────────────────────────────────────┘
│
┌────────────────────────┴─────────────────────────────────────────┐
│ EHR Integration & Documentation │
│ (FHIR, HL7, Encounter Creation, Triage Note Generation) │
└────────────────────────┬─────────────────────────────────────────┘
│
┌────────────────────────┴─────────────────────────────────────────┐
│ Audit Logging & Compliance │
│ (HIPAA Audit Trails, Quality Assurance, Outcome Tracking) │
└──────────────────────────────────────────────────────────────────┘Component 1: Multi-Step Assessment Workflow Engine
The workflow engine guides patients through adaptive assessment questionnaires:
# Clinical Triage Workflow Engine
# Built with JustCopy.ai's clinical workflow templates
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass
from enum import Enum
import logging
class QuestionType(Enum):
"""Types of assessment questions"""
SINGLE_CHOICE = "single_choice"
MULTIPLE_CHOICE = "multiple_choice"
NUMERIC_INPUT = "numeric_input"
TEXT_INPUT = "text_input"
YES_NO = "yes_no"
SEVERITY_SCALE = "severity_scale"
VITAL_SIGNS = "vital_signs"
BODY_DIAGRAM = "body_diagram"
@dataclass
class AssessmentQuestion:
"""Represents a single triage assessment question"""
id: str
question_text: str
question_type: QuestionType
options: Optional[List[Dict]] = None
validation_rules: Optional[Dict] = None
skip_logic: Optional[Callable] = None
red_flag_indicators: Optional[List[str]] = None
required: bool = True
help_text: Optional[str] = None
@dataclass
class TriageAssessmentPath:
"""Defines a complete triage assessment pathway"""
path_id: str
path_name: str
trigger_conditions: Dict
questions: List[AssessmentQuestion]
priority: int
class TriageWorkflowEngine:
"""
Manages multi-step triage assessment workflows with adaptive
branching based on patient responses.
"""
def __init__(self, workflow_config: Dict):
self.config = workflow_config
self.assessment_paths = self._load_assessment_paths()
self.logger = logging.getLogger(__name__)
def initialize_assessment(
self,
chief_complaint: str,
patient_demographics: Dict,
arrival_method: str = 'walk_in'
) -> Dict:
"""
Initialize triage assessment based on chief complaint.
Args:
chief_complaint: Primary symptom/reason for visit
patient_demographics: Age, sex, pregnancy status
arrival_method: walk_in, ambulance, transfer
Returns:
Initial assessment session with first questions
"""
# Select appropriate assessment path
assessment_path = self._select_assessment_path(
chief_complaint, patient_demographics, arrival_method
)
# Create assessment session
session = {
'session_id': self._generate_session_id(),
'assessment_path': assessment_path,
'current_question_index': 0,
'responses': [],
'red_flags_detected': [],
'requires_immediate_escalation': False,
'created_at': datetime.now().isoformat()
}
# Get first question
first_question = self._get_next_question(session)
return {
'session': session,
'question': first_question,
'progress': self._calculate_progress(session)
}
def process_response(
self,
session: Dict,
question_id: str,
response: Dict
) -> Dict:
"""
Process patient response and determine next step in assessment.
Args:
session: Current assessment session
question_id: ID of answered question
response: Patient's response
Returns:
Next question or assessment completion
"""
# Validate response
validation_result = self._validate_response(question_id, response)
if not validation_result['valid']:
return {
'error': 'invalid_response',
'message': validation_result['error_message'],
'question': self._get_question_by_id(question_id)
}
# Store response
session['responses'].append({
'question_id': question_id,
'response': response,
'timestamp': datetime.now().isoformat()
})
# Check for red flags in response
red_flags = self._check_red_flags(question_id, response)
if red_flags:
session['red_flags_detected'].extend(red_flags)
# Immediate escalation if critical red flag
if self._is_critical_red_flag(red_flags):
session['requires_immediate_escalation'] = True
return {
'escalation_required': True,
'urgency': 'immediate',
'red_flags': red_flags,
'recommendation': 'Immediate clinical evaluation required'
}
# Determine next question based on skip logic
next_question = self._get_next_question(session)
if next_question:
return {
'session': session,
'question': next_question,
'progress': self._calculate_progress(session)
}
else:
# Assessment complete - perform triage classification
return self._complete_assessment(session)
def _select_assessment_path(
self,
chief_complaint: str,
patient_demographics: Dict,
arrival_method: str
) -> TriageAssessmentPath:
"""
Select appropriate assessment pathway based on presentation.
"""
# Immediate life threats skip detailed assessment
if arrival_method == 'ambulance' and self._is_life_threatening(chief_complaint):
return self.assessment_paths['immediate_resuscitation']
# Chest pain has specialized assessment
if 'chest pain' in chief_complaint.lower():
return self.assessment_paths['chest_pain_pathway']
# Pediatric assessments use age-specific pathways
if patient_demographics.get('age_years', 100) < 18:
return self.assessment_paths['pediatric_general']
# Pregnancy-specific pathway
if patient_demographics.get('pregnancy_status') == 'pregnant':
return self.assessment_paths['pregnancy_related']
# Default general adult pathway
return self.assessment_paths['general_adult']
def _load_assessment_paths(self) -> Dict[str, TriageAssessmentPath]:
"""
Load predefined assessment pathways.
"""
return {
'chest_pain_pathway': TriageAssessmentPath(
path_id='chest_pain',
path_name='Chest Pain Assessment',
trigger_conditions={'chief_complaint': 'chest pain'},
questions=self._build_chest_pain_questions(),
priority=1
),
'general_adult': TriageAssessmentPath(
path_id='general_adult',
path_name='General Adult Assessment',
trigger_conditions={'age': '>=18'},
questions=self._build_general_adult_questions(),
priority=5
),
'pediatric_general': TriageAssessmentPath(
path_id='pediatric',
path_name='Pediatric Assessment',
trigger_conditions={'age': '<18'},
questions=self._build_pediatric_questions(),
priority=2
),
# Additional pathways...
}
def _build_chest_pain_questions(self) -> List[AssessmentQuestion]:
"""
Build chest pain-specific assessment questions.
Uses HEART score components and cardiac risk factors.
"""
return [
AssessmentQuestion(
id='chest_pain_quality',
question_text='How would you describe your chest pain?',
question_type=QuestionType.SINGLE_CHOICE,
options=[
{
'value': 'pressure_squeezing',
'label': 'Pressure or squeezing sensation',
'risk_score': 3
},
{
'value': 'sharp_stabbing',
'label': 'Sharp or stabbing',
'risk_score': 1
},
{
'value': 'burning',
'label': 'Burning sensation',
'risk_score': 1
},
{
'value': 'aching',
'label': 'Dull aching',
'risk_score': 2
}
],
red_flag_indicators=['pressure_squeezing'],
help_text='The type of pain can help determine the cause'
),
AssessmentQuestion(
id='chest_pain_radiation',
question_text='Does the pain spread to other areas?',
question_type=QuestionType.MULTIPLE_CHOICE,
options=[
{'value': 'left_arm', 'label': 'Left arm', 'risk_score': 3},
{'value': 'jaw', 'label': 'Jaw or neck', 'risk_score': 3},
{'value': 'back', 'label': 'Back', 'risk_score': 2},
{'value': 'right_arm', 'label': 'Right arm', 'risk_score': 2},
{'value': 'none', 'label': 'No radiation', 'risk_score': 0}
],
red_flag_indicators=['left_arm', 'jaw'],
help_text='Radiation to arm or jaw is concerning for cardiac causes'
),
AssessmentQuestion(
id='chest_pain_associated_symptoms',
question_text='Are you experiencing any of these symptoms?',
question_type=QuestionType.MULTIPLE_CHOICE,
options=[
{'value': 'shortness_of_breath', 'label': 'Shortness of breath', 'risk_score': 3},
{'value': 'sweating', 'label': 'Sweating', 'risk_score': 3},
{'value': 'nausea', 'label': 'Nausea or vomiting', 'risk_score': 2},
{'value': 'dizziness', 'label': 'Dizziness or lightheadedness', 'risk_score': 3},
{'value': 'palpitations', 'label': 'Heart racing or palpitations', 'risk_score': 2},
{'value': 'none', 'label': 'None of these', 'risk_score': 0}
],
red_flag_indicators=['shortness_of_breath', 'sweating', 'dizziness'],
help_text='These symptoms together with chest pain are concerning'
),
AssessmentQuestion(
id='cardiac_risk_factors',
question_text='Do you have any of these medical conditions or risk factors?',
question_type=QuestionType.MULTIPLE_CHOICE,
options=[
{'value': 'prior_heart_attack', 'label': 'Previous heart attack', 'risk_score': 4},
{'value': 'diabetes', 'label': 'Diabetes', 'risk_score': 2},
{'value': 'high_blood_pressure', 'label': 'High blood pressure', 'risk_score': 1},
{'value': 'high_cholesterol', 'label': 'High cholesterol', 'risk_score': 1},
{'value': 'smoking', 'label': 'Current or former smoker', 'risk_score': 2},
{'value': 'family_history', 'label': 'Family history of heart disease', 'risk_score': 1},
{'value': 'none', 'label': 'None of these', 'risk_score': 0}
],
help_text='Risk factors increase likelihood of cardiac causes'
),
AssessmentQuestion(
id='chest_pain_onset',
question_text='How did the chest pain start?',
question_type=QuestionType.SINGLE_CHOICE,
options=[
{'value': 'sudden', 'label': 'Suddenly', 'risk_score': 3},
{'value': 'gradual', 'label': 'Gradually over time', 'risk_score': 1},
{'value': 'exertion', 'label': 'During physical activity', 'risk_score': 3},
{'value': 'rest', 'label': 'While resting', 'risk_score': 2}
],
red_flag_indicators=['sudden', 'exertion']
),
AssessmentQuestion(
id='chest_pain_duration',
question_text='How long have you had this pain?',
question_type=QuestionType.SINGLE_CHOICE,
options=[
{'value': 'less_5_min', 'label': 'Less than 5 minutes', 'risk_score': 1},
{'value': '5_30_min', 'label': '5-30 minutes', 'risk_score': 3},
{'value': '30_min_2_hr', 'label': '30 minutes to 2 hours', 'risk_score': 3},
{'value': 'more_2_hr', 'label': 'More than 2 hours', 'risk_score': 2},
{'value': 'days', 'label': 'Days or longer', 'risk_score': 1}
],
red_flag_indicators=['5_30_min', '30_min_2_hr']
),
AssessmentQuestion(
id='vital_signs_check',
question_text='If you have measured your vital signs, please enter them',
question_type=QuestionType.VITAL_SIGNS,
validation_rules={
'systolic_bp': {'min': 60, 'max': 250},
'diastolic_bp': {'min': 40, 'max': 150},
'heart_rate': {'min': 30, 'max': 220},
'oxygen_saturation': {'min': 70, 'max': 100}
},
required=False,
help_text='This helps assess severity (optional but helpful)'
)
]
def _check_red_flags(
self,
question_id: str,
response: Dict
) -> List[Dict]:
"""
Check if response contains red flag indicators requiring
immediate attention.
"""
question = self._get_question_by_id(question_id)
red_flags = []
if not question.red_flag_indicators:
return red_flags
# Check if response contains any red flag values
response_values = response.get('values', [])
if isinstance(response_values, str):
response_values = [response_values]
for red_flag_value in question.red_flag_indicators:
if red_flag_value in response_values:
red_flags.append({
'question_id': question_id,
'red_flag_value': red_flag_value,
'severity': self._get_red_flag_severity(
question_id, red_flag_value
),
'detected_at': datetime.now().isoformat()
})
return red_flags
def _is_critical_red_flag(self, red_flags: List[Dict]) -> bool:
"""
Determine if any detected red flags require immediate escalation.
"""
critical_red_flags = [
'unresponsive',
'not_breathing',
'severe_bleeding',
'chest_pain_with_sob',
'stroke_symptoms',
'severe_allergic_reaction'
]
return any(
flag['red_flag_value'] in critical_red_flags
for flag in red_flags
)
def _complete_assessment(self, session: Dict) -> Dict:
"""
Complete assessment and perform triage classification.
"""
# Calculate triage scores
triage_result = self._calculate_triage_level(session)
# Generate care pathway recommendation
care_pathway = self._recommend_care_pathway(triage_result, session)
# Create clinical documentation
triage_note = self._generate_triage_note(session, triage_result)
return {
'assessment_complete': True,
'session_id': session['session_id'],
'triage_level': triage_result['esi_level'],
'urgency_score': triage_result['urgency_score'],
'red_flags': session['red_flags_detected'],
'care_pathway': care_pathway,
'triage_note': triage_note,
'estimated_wait_time': self._estimate_wait_time(triage_result['esi_level'])
}
def _calculate_triage_level(self, session: Dict) -> Dict:
"""
Calculate ESI triage level based on assessment responses.
"""
# Extract relevant data from responses
responses = session['responses']
red_flags = session['red_flags_detected']
# Check for ESI Level 1 criteria (life-threatening)
if self._meets_esi_level_1_criteria(responses, red_flags):
return {
'esi_level': 1,
'esi_description': 'Resuscitation',
'urgency_score': 1.0,
'reasoning': 'Life-threatening condition requiring immediate intervention'
}
# Check for ESI Level 2 criteria (emergent, high-risk)
if self._meets_esi_level_2_criteria(responses, red_flags):
return {
'esi_level': 2,
'esi_description': 'Emergent',
'urgency_score': 0.85,
'reasoning': 'High-risk situation requiring rapid evaluation'
}
# Calculate resource needs for ESI 3-5
resource_needs = self._estimate_resource_needs(responses)
# ESI Level 3: Multiple resources
if resource_needs >= 2:
return {
'esi_level': 3,
'esi_description': 'Urgent',
'urgency_score': 0.65,
'reasoning': f'Anticipated {resource_needs} resource needs',
'estimated_resources': resource_needs
}
# ESI Level 4: One resource
if resource_needs == 1:
return {
'esi_level': 4,
'esi_description': 'Less Urgent',
'urgency_score': 0.35,
'reasoning': 'One anticipated resource need',
'estimated_resources': 1
}
# ESI Level 5: No resources
return {
'esi_level': 5,
'esi_description': 'Non-Urgent',
'urgency_score': 0.15,
'reasoning': 'No anticipated resource needs',
'estimated_resources': 0
}
def _meets_esi_level_1_criteria(
self,
responses: List[Dict],
red_flags: List[Dict]
) -> bool:
"""
Check if patient meets ESI Level 1 (resuscitation) criteria.
ESI-1 criteria:
- Immediate life-threatening condition
- Requires immediate physician evaluation and intervention
"""
level_1_indicators = [
'unresponsive',
'not_breathing',
'no_pulse',
'severe_respiratory_distress',
'uncontrolled_bleeding',
'seizure_ongoing'
]
return any(
flag['red_flag_value'] in level_1_indicators
for flag in red_flags
)
def _meets_esi_level_2_criteria(
self,
responses: List[Dict],
red_flags: List[Dict]
) -> bool:
"""
Check if patient meets ESI Level 2 (emergent) criteria.
ESI-2 criteria:
- High-risk situation
- OR severe pain/distress
- OR confused/lethargic/disoriented
"""
# Check high-risk situations
high_risk_indicators = [
'chest_pain_cardiac',
'stroke_symptoms',
'severe_asthma',
'significant_trauma',
'suicidal_ideation',
'severe_pain'
]
has_high_risk = any(
flag['red_flag_value'] in high_risk_indicators
for flag in red_flags
)
# Check vital sign abnormalities
vital_signs = self._extract_vital_signs(responses)
has_abnormal_vitals = self._has_abnormal_vitals_esi2(vital_signs)
return has_high_risk or has_abnormal_vitalsComponent 2: ESI Integration and Risk Scoring
Implementing the Emergency Severity Index requires sophisticated clinical logic:
# ESI Implementation
# Built with JustCopy.ai's emergency medicine templates
from typing import Dict, List, Optional
from enum import Enum
class ResourceType(Enum):
"""Types of hospital resources considered in ESI calculation"""
LABS = "laboratory_tests"
IMAGING = "imaging_studies"
IV_FLUIDS = "iv_fluids_medications"
PROCEDURES = "procedures"
CONSULTATION = "specialist_consultation"
COMPLEX_CARE = "complex_nursing_care"
class ESICalculator:
"""
Implements Emergency Severity Index triage algorithm.
"""
def calculate_esi_level(
self,
patient_presentation: Dict,
vital_signs: Dict,
medical_history: Dict,
symptoms: List[Dict]
) -> Dict:
"""
Calculate ESI level using standard algorithm.
Decision tree:
1. Requires immediate life-saving intervention? → ESI-1
2. High-risk situation or severe pain/distress? → ESI-2
3. Estimate resource needs:
- ≥2 resources → ESI-3
- 1 resource → ESI-4
- 0 resources → ESI-5
"""
# Step 1: Check for ESI-1 (immediate life-threatening)
if self._requires_immediate_intervention(patient_presentation, vital_signs):
return {
'esi_level': 1,
'category': 'Resuscitation',
'max_wait_time_minutes': 0,
'reasoning': 'Immediate life-saving intervention required',
'interventions_needed': self._identify_immediate_interventions(
patient_presentation
)
}
# Step 2: Check for ESI-2 (high-risk or severe distress)
esi_2_assessment = self._assess_esi_level_2_criteria(
patient_presentation, vital_signs, medical_history, symptoms
)
if esi_2_assessment['meets_criteria']:
return {
'esi_level': 2,
'category': 'Emergent',
'max_wait_time_minutes': 10,
'reasoning': esi_2_assessment['reasoning'],
'high_risk_factors': esi_2_assessment['risk_factors']
}
# Step 3: Estimate resource needs for ESI 3-5
resource_estimate = self._estimate_resource_needs(
symptoms, patient_presentation, medical_history
)
if resource_estimate['count'] >= 2:
return {
'esi_level': 3,
'category': 'Urgent',
'max_wait_time_minutes': 30,
'reasoning': f"Estimated {resource_estimate['count']} resources needed",
'anticipated_resources': resource_estimate['resources']
}
elif resource_estimate['count'] == 1:
return {
'esi_level': 4,
'category': 'Less Urgent',
'max_wait_time_minutes': 60,
'reasoning': 'One resource anticipated',
'anticipated_resources': resource_estimate['resources']
}
else:
return {
'esi_level': 5,
'category': 'Non-Urgent',
'max_wait_time_minutes': 120,
'reasoning': 'No resources anticipated',
'disposition': 'May be appropriate for urgent care or primary care'
}
def _requires_immediate_intervention(
self,
presentation: Dict,
vital_signs: Dict
) -> bool:
"""
Determine if patient requires immediate life-saving intervention (ESI-1).
ESI-1 conditions:
- Intubation required
- Immediate need for procedural sedation
- Life-threatening vital signs
- Cardiac arrest
- Severe trauma
"""
# Check vital signs for life-threatening values
if vital_signs:
# Severe hypotension
if vital_signs.get('systolic_bp', 999) < 70:
return True
# Severe tachycardia or bradycardia
heart_rate = vital_signs.get('heart_rate', 0)
if heart_rate > 180 or (heart_rate < 40 and heart_rate > 0):
return True
# Severe hypoxia
if vital_signs.get('oxygen_saturation', 100) < 85:
return True
# Severe respiratory distress
resp_rate = vital_signs.get('respiratory_rate', 0)
if resp_rate > 35 or resp_rate < 6:
return True
# Check presenting symptoms
life_threatening_presentations = [
'cardiac_arrest',
'respiratory_arrest',
'severe_respiratory_distress',
'unresponsive',
'active_seizure',
'massive_hemorrhage',
'severe_trauma'
]
chief_complaint = presentation.get('chief_complaint', '').lower()
return any(
condition in chief_complaint
for condition in life_threatening_presentations
)
def _assess_esi_level_2_criteria(
self,
presentation: Dict,
vital_signs: Dict,
medical_history: Dict,
symptoms: List[Dict]
) -> Dict:
"""
Assess if patient meets ESI Level 2 criteria.
ESI-2 criteria:
A. High-risk situations:
- New onset confusion/lethargy/disorientation
- Severe pain (8-10/10)
- Distress requiring immediate intervention
B. High-risk conditions by presentation
"""
meets_criteria = False
risk_factors = []
reasoning = []
# Check for altered mental status
if self._has_altered_mental_status(symptoms, presentation):
meets_criteria = True
risk_factors.append('altered_mental_status')
reasoning.append('New onset confusion or altered consciousness')
# Check for severe pain
pain_level = self._extract_pain_level(symptoms)
if pain_level and pain_level >= 8:
meets_criteria = True
risk_factors.append('severe_pain')
reasoning.append(f'Severe pain ({pain_level}/10)')
# Check high-risk chief complaints
high_risk_presentations = self._evaluate_high_risk_presentations(
presentation['chief_complaint'],
medical_history,
vital_signs
)
if high_risk_presentations:
meets_criteria = True
risk_factors.extend(high_risk_presentations)
reasoning.append('High-risk presentation requiring urgent evaluation')
# Check vital sign abnormalities indicating ESI-2
vital_sign_concerns = self._check_esi2_vital_signs(vital_signs)
if vital_sign_concerns:
meets_criteria = True
risk_factors.append('abnormal_vitals')
reasoning.extend(vital_sign_concerns)
return {
'meets_criteria': meets_criteria,
'risk_factors': risk_factors,
'reasoning': '; '.join(reasoning) if reasoning else None
}
def _evaluate_high_risk_presentations(
self,
chief_complaint: str,
medical_history: Dict,
vital_signs: Dict
) -> List[str]:
"""
Identify high-risk presentations that warrant ESI-2.
"""
high_risk = []
chief_complaint_lower = chief_complaint.lower()
# Chest pain in patient with cardiac risk factors
if 'chest pain' in chief_complaint_lower:
if self._has_cardiac_risk_factors(medical_history):
high_risk.append('chest_pain_cardiac_risk')
# Severe headache ("worst headache of life")
if any(phrase in chief_complaint_lower for phrase in ['worst headache', 'severe headache', 'sudden headache']):
high_risk.append('severe_headache_ruptured_aneurysm_concern')
# Stroke symptoms (facial droop, arm weakness, speech difficulty)
if any(symptom in chief_complaint_lower for symptom in ['weakness', 'numbness', 'facial droop', 'slurred speech']):
high_risk.append('stroke_symptoms')
# Shortness of breath with hypoxia
if 'short' in chief_complaint_lower and 'breath' in chief_complaint_lower:
if vital_signs and vital_signs.get('oxygen_saturation', 100) < 92:
high_risk.append('dyspnea_with_hypoxia')
# Abdominal pain in certain populations
if 'abdominal pain' in chief_complaint_lower:
age = medical_history.get('age_years', 0)
if age > 65: # Elderly with abdominal pain
high_risk.append('abdominal_pain_elderly')
if medical_history.get('pregnancy_status') == 'pregnant':
high_risk.append('abdominal_pain_pregnancy')
# Active suicide ideation
if any(term in chief_complaint_lower for term in ['suicide', 'kill myself', 'end my life']):
high_risk.append('active_suicidal_ideation')
return high_risk
def _estimate_resource_needs(
self,
symptoms: List[Dict],
presentation: Dict,
medical_history: Dict
) -> Dict:
"""
Estimate anticipated resource needs (ESI 3-5).
Resources counted:
- Laboratory tests
- Imaging studies (X-ray, CT, MRI, ultrasound)
- IV fluids or medications
- Simple procedures (suturing, splinting)
- Specialist consultation
"""
resources = []
chief_complaint = presentation['chief_complaint'].lower()
# Trauma/injury → Likely needs imaging
if any(term in chief_complaint for term in ['injury', 'trauma', 'fracture', 'fall', 'hit', 'accident']):
resources.append(ResourceType.IMAGING)
# Abdominal pain → Likely needs labs and possibly imaging
if 'abdominal pain' in chief_complaint:
resources.append(ResourceType.LABS)
resources.append(ResourceType.IMAGING)
# Fever with concerning features → Labs
if self._has_symptom(symptoms, 'fever'):
if self._has_concerning_fever_features(symptoms, medical_history):
resources.append(ResourceType.LABS)
# Dehydration → IV fluids
if self._shows_dehydration_signs(symptoms):
resources.append(ResourceType.IV_FLUIDS)
# Lacerations → Procedure (suturing)
if 'laceration' in chief_complaint or 'cut' in chief_complaint:
resources.append(ResourceType.PROCEDURES)
# Sprains/strains → Imaging
if 'sprain' in chief_complaint or 'strain' in chief_complaint:
if self._meets_ottawa_rules(presentation):
resources.append(ResourceType.IMAGING)
# Chest pain → Labs, ECG, possibly imaging
if 'chest pain' in chief_complaint:
resources.append(ResourceType.LABS) # Troponin, etc.
# ECG not counted as "resource" per ESI guidelines
# Remove duplicates
unique_resources = list(set(resources))
return {
'count': len(unique_resources),
'resources': [r.value for r in unique_resources],
'rationale': self._generate_resource_rationale(unique_resources)
}Component 3: React Native Mobile Triage Application
Modern triage systems require mobile accessibility:
// React Native Mobile Triage App
// Built with JustCopy.ai's mobile healthcare templates
import React, { useState, useEffect } from "react";
import {
View,
Text,
StyleSheet,
ScrollView,
TouchableOpacity,
Alert,
ActivityIndicator,
} from "react-native";
import { useNavigation } from "@react-navigation/native";
interface TriageQuestion {
id: string;
questionText: string;
questionType: "single_choice" | "multiple_choice" | "numeric" | "yes_no";
options?: Array<{ value: string; label: string; riskScore?: number }>;
redFlagIndicators?: string[];
helpText?: string;
}
interface TriageSession {
sessionId: string;
currentQuestionIndex: number;
responses: Array<{ questionId: string; response: any }>;
redFlagsDetected: string[];
requiresImmediateEscalation: boolean;
}
const TriageAssessmentScreen: React.FC = () => {
const navigation = useNavigation();
const [session, setSession] = useState<TriageSession | null>(null);
const [currentQuestion, setCurrentQuestion] = useState<TriageQuestion | null>(
null
);
const [selectedResponses, setSelectedResponses] = useState<string[]>([]);
const [loading, setLoading] = useState(false);
const [progress, setProgress] = useState(0);
useEffect(() => {
initializeTriageAssessment();
}, []);
const initializeTriageAssessment = async () => {
setLoading(true);
try {
// Call triage API to initialize assessment
const response = await fetch(
"https://api.yourplatform.com/triage/initialize",
{
method: "POST",
headers: {
"Content-Type": "application/json",
Authorization: `Bearer ${await getAuthToken()}`,
},
body: JSON.stringify({
chiefComplaint: route.params.chiefComplaint,
patientDemographics: route.params.demographics,
arrivalMethod: "walk_in",
}),
}
);
const data = await response.json();
setSession(data.session);
setCurrentQuestion(data.question);
setProgress(data.progress);
} catch (error) {
console.error("Error initializing triage:", error);
Alert.alert(
"Error",
"Failed to start triage assessment. Please try again."
);
} finally {
setLoading(false);
}
};
const handleResponseSubmit = async () => {
if (selectedResponses.length === 0) {
Alert.alert("Required", "Please select at least one option");
return;
}
setLoading(true);
try {
const response = await fetch(
"https://api.yourplatform.com/triage/respond",
{
method: "POST",
headers: {
"Content-Type": "application/json",
Authorization: `Bearer ${await getAuthToken()}`,
},
body: JSON.stringify({
sessionId: session?.sessionId,
questionId: currentQuestion?.id,
response: {
values: selectedResponses,
timestamp: new Date().toISOString(),
},
}),
}
);
const data = await response.json();
// Check if immediate escalation required
if (data.escalationRequired) {
navigation.navigate("ImmediateEscalation", {
urgency: data.urgency,
redFlags: data.redFlags,
recommendation: data.recommendation,
});
return;
}
// Check if assessment complete
if (data.assessmentComplete) {
navigation.navigate("TriageResults", {
triageLevel: data.triageLevel,
urgencyScore: data.urgencyScore,
carePathway: data.carePathway,
triageNote: data.triageNote,
estimatedWaitTime: data.estimatedWaitTime,
});
return;
}
// Continue to next question
setSession(data.session);
setCurrentQuestion(data.question);
setProgress(data.progress);
setSelectedResponses([]);
} catch (error) {
console.error("Error submitting response:", error);
Alert.alert("Error", "Failed to process response. Please try again.");
} finally {
setLoading(false);
}
};
const handleOptionSelect = (optionValue: string) => {
if (currentQuestion?.questionType === "single_choice") {
setSelectedResponses([optionValue]);
} else if (currentQuestion?.questionType === "multiple_choice") {
if (selectedResponses.includes(optionValue)) {
setSelectedResponses(
selectedResponses.filter((v) => v !== optionValue)
);
} else {
setSelectedResponses([...selectedResponses, optionValue]);
}
}
};
if (loading) {
return (
<View style={styles.loadingContainer}>
<ActivityIndicator size="large" color="#007AFF" />
<Text style={styles.loadingText}>Processing...</Text>
</View>
);
}
return (
<View style={styles.container}>
{/* Progress Bar */}
<View style={styles.progressContainer}>
<View style={[styles.progressBar, { width: `${progress * 100}%` }]} />
</View>
<ScrollView style={styles.content}>
{/* Question Text */}
<Text style={styles.questionText}>{currentQuestion?.questionText}</Text>
{/* Help Text */}
{currentQuestion?.helpText && (
<View style={styles.helpTextContainer}>
<Text style={styles.helpText}>{currentQuestion.helpText}</Text>
</View>
)}
{/* Response Options */}
<View style={styles.optionsContainer}>
{currentQuestion?.options?.map((option) => {
const isSelected = selectedResponses.includes(option.value);
return (
<TouchableOpacity
key={option.value}
style={[
styles.optionButton,
isSelected && styles.optionButtonSelected,
]}
onPress={() => handleOptionSelect(option.value)}
>
<View style={styles.optionContent}>
<View
style={[
styles.checkbox,
isSelected && styles.checkboxSelected,
]}
>
{isSelected && <Text style={styles.checkmark}>✓</Text>}
</View>
<Text
style={[
styles.optionLabel,
isSelected && styles.optionLabelSelected,
]}
>
{option.label}
</Text>
</View>
</TouchableOpacity>
);
})}
</View>
</ScrollView>
{/* Submit Button */}
<View style={styles.footer}>
<TouchableOpacity
style={[
styles.submitButton,
selectedResponses.length === 0 && styles.submitButtonDisabled,
]}
onPress={handleResponseSubmit}
disabled={selectedResponses.length === 0}
>
<Text style={styles.submitButtonText}>Continue</Text>
</TouchableOpacity>
</View>
</View>
);
};
const styles = StyleSheet.create({
container: {
flex: 1,
backgroundColor: "#F5F5F5",
},
progressContainer: {
height: 4,
backgroundColor: "#E0E0E0",
},
progressBar: {
height: "100%",
backgroundColor: "#007AFF",
},
loadingContainer: {
flex: 1,
justifyContent: "center",
alignItems: "center",
backgroundColor: "#F5F5F5",
},
loadingText: {
marginTop: 16,
fontSize: 16,
color: "#666",
},
content: {
flex: 1,
padding: 20,
},
questionText: {
fontSize: 22,
fontWeight: "600",
color: "#000",
marginBottom: 12,
},
helpTextContainer: {
backgroundColor: "#E3F2FD",
padding: 12,
borderRadius: 8,
marginBottom: 20,
},
helpText: {
fontSize: 14,
color: "#1976D2",
lineHeight: 20,
},
optionsContainer: {
marginTop: 8,
},
optionButton: {
backgroundColor: "#FFF",
borderRadius: 12,
padding: 16,
marginBottom: 12,
borderWidth: 2,
borderColor: "#E0E0E0",
},
optionButtonSelected: {
borderColor: "#007AFF",
backgroundColor: "#F0F8FF",
},
optionContent: {
flexDirection: "row",
alignItems: "center",
},
checkbox: {
width: 24,
height: 24,
borderRadius: 12,
borderWidth: 2,
borderColor: "#CCC",
marginRight: 12,
justifyContent: "center",
alignItems: "center",
},
checkboxSelected: {
borderColor: "#007AFF",
backgroundColor: "#007AFF",
},
checkmark: {
color: "#FFF",
fontSize: 16,
fontWeight: "bold",
},
optionLabel: {
fontSize: 16,
color: "#333",
flex: 1,
},
optionLabelSelected: {
color: "#007AFF",
fontWeight: "500",
},
footer: {
padding: 20,
backgroundColor: "#FFF",
borderTopWidth: 1,
borderTopColor: "#E0E0E0",
},
submitButton: {
backgroundColor: "#007AFF",
borderRadius: 12,
padding: 16,
alignItems: "center",
},
submitButtonDisabled: {
backgroundColor: "#CCC",
},
submitButtonText: {
color: "#FFF",
fontSize: 18,
fontWeight: "600",
},
});
export default TriageAssessmentScreen;Component 4: HIPAA-Compliant Audit Logging
Comprehensive audit logging is essential for compliance and quality assurance:
# HIPAA-Compliant Audit Logging System
# Built with JustCopy.ai's healthcare compliance templates
from typing import Dict, Optional
from datetime import datetime
import hashlib
import json
import logging
from enum import Enum
class AuditEventType(Enum):
"""Types of auditable events"""
TRIAGE_STARTED = "triage_assessment_started"
TRIAGE_COMPLETED = "triage_assessment_completed"
QUESTION_ANSWERED = "assessment_question_answered"
RED_FLAG_DETECTED = "red_flag_symptom_detected"
CLINICAL_OVERRIDE = "clinician_override_applied"
ESCALATION_TRIGGERED = "escalation_to_clinician"
ASSESSMENT_VIEWED = "triage_results_viewed"
DATA_ACCESSED = "patient_data_accessed"
DATA_MODIFIED = "patient_data_modified"
SYSTEM_ALERT = "system_alert_generated"
class HIPAAAuditLogger:
"""
HIPAA-compliant audit logging for clinical triage system.
Implements requirements from:
- 45 CFR § 164.308(a)(1)(ii)(D) - Information system activity review
- 45 CFR § 164.312(b) - Audit controls
"""
def __init__(self, config: Dict):
self.config = config
self.logger = self._setup_audit_logger()
self.encryption_key = config.get('encryption_key')
def log_triage_event(
self,
event_type: AuditEventType,
session_id: str,
patient_id: str,
user_id: Optional[str],
event_data: Dict,
ip_address: str,
user_agent: str
) -> str:
"""
Log a triage-related audit event.
Args:
event_type: Type of event being logged
session_id: Triage session identifier
patient_id: Patient MRN (will be hashed in log)
user_id: User performing action (if applicable)
event_data: Event-specific data
ip_address: Source IP address
user_agent: User agent string
Returns:
Audit log entry ID
"""
# Generate audit log entry
audit_entry = {
'audit_id': self._generate_audit_id(),
'timestamp': datetime.now().isoformat(),
'event_type': event_type.value,
'session_id': session_id,
'patient_id_hash': self._hash_identifier(patient_id),
'user_id': user_id,
'event_data': self._sanitize_event_data(event_data),
'ip_address': ip_address,
'user_agent': user_agent,
'system_version': self.config.get('system_version'),
'facility_id': self.config.get('facility_id')
}
# Encrypt sensitive fields
encrypted_entry = self._encrypt_sensitive_fields(audit_entry)
# Write to audit log
self._write_audit_log(encrypted_entry)
# Check if event requires alerting
if self._requires_security_alert(event_type, event_data):
self._send_security_alert(audit_entry)
return audit_entry['audit_id']
def log_red_flag_detection(
self,
session_id: str,
patient_id: str,
red_flag_type: str,
red_flag_data: Dict,
system_response: str
) -> None:
"""
Log detection of red flag symptom requiring immediate attention.
"""
event_data = {
'red_flag_type': red_flag_type,
'red_flag_details': red_flag_data,
'system_response': system_response,
'escalation_triggered': red_flag_data.get('critical', False)
}
self.log_triage_event(
event_type=AuditEventType.RED_FLAG_DETECTED,
session_id=session_id,
patient_id=patient_id,
user_id=None,
event_data=event_data,
ip_address=self._get_current_ip(),
user_agent='system'
)
def log_clinician_override(
self,
session_id: str,
patient_id: str,
clinician_id: str,
original_triage_level: int,
override_triage_level: int,
override_reason: str
) -> None:
"""
Log when clinician overrides automated triage recommendation.
"""
event_data = {
'original_esi_level': original_triage_level,
'override_esi_level': override_triage_level,
'override_reason': override_reason,
'clinician_credentials': self._get_clinician_credentials(clinician_id)
}
self.log_triage_event(
event_type=AuditEventType.CLINICAL_OVERRIDE,
session_id=session_id,
patient_id=patient_id,
user_id=clinician_id,
event_data=event_data,
ip_address=self._get_current_ip(),
user_agent=self._get_current_user_agent()
)
def generate_audit_report(
self,
start_date: datetime,
end_date: datetime,
event_types: Optional[List[AuditEventType]] = None
) -> Dict:
"""
Generate audit report for compliance review.
Used for:
- HIPAA compliance audits
- Quality assurance reviews
- Outcome analysis
- System performance monitoring
"""
# Query audit logs
audit_entries = self._query_audit_logs(
start_date, end_date, event_types
)
# Aggregate statistics
stats = {
'total_assessments': self._count_assessments(audit_entries),
'red_flags_detected': self._count_red_flags(audit_entries),
'clinical_overrides': self._count_overrides(audit_entries),
'escalations': self._count_escalations(audit_entries),
'esi_level_distribution': self._calculate_esi_distribution(audit_entries),
'average_assessment_time': self._calculate_avg_time(audit_entries),
'override_rate': self._calculate_override_rate(audit_entries)
}
# Identify concerning patterns
alerts = self._identify_concerning_patterns(audit_entries)
return {
'report_period': {
'start': start_date.isoformat(),
'end': end_date.isoformat()
},
'statistics': stats,
'alerts': alerts,
'detailed_entries': self._format_entries_for_report(audit_entries)
}
def _hash_identifier(self, identifier: str) -> str:
"""
Hash patient identifiers for privacy in audit logs.
Uses HMAC-SHA256 for consistent, secure hashing.
"""
return hashlib.sha256(
f"{identifier}{self.config['hash_salt']}".encode()
).hexdigest()
def _sanitize_event_data(self, event_data: Dict) -> Dict:
"""
Remove or hash PHI from event data before logging.
"""
sanitized = event_data.copy()
# Remove sensitive fields
sensitive_fields = ['ssn', 'credit_card', 'password']
for field in sensitive_fields:
if field in sanitized:
sanitized[field] = '[REDACTED]'
# Hash identifiable information
if 'patient_name' in sanitized:
sanitized['patient_name_hash'] = self._hash_identifier(
sanitized['patient_name']
)
del sanitized['patient_name']
return sanitized
def _requires_security_alert(
self,
event_type: AuditEventType,
event_data: Dict
) -> bool:
"""
Determine if event requires immediate security alerting.
"""
# Alert on unauthorized access attempts
if event_type == AuditEventType.DATA_ACCESSED:
if event_data.get('authorization_failed'):
return True
# Alert on critical red flags
if event_type == AuditEventType.RED_FLAG_DETECTED:
if event_data.get('red_flag_type') in ['life_threatening', 'critical']:
return True
# Alert on suspicious patterns
if self._detect_suspicious_pattern(event_type, event_data):
return True
return FalseThe JustCopy.ai Implementation Advantage
Building a production clinical triage system from scratch requires:
Timeline: 18-24 months
- Clinical workflow design: 3-4 months
- ESI algorithm implementation: 4-6 months
- Mobile app development: 5-7 months
- EHR integration: 3-4 months
- Clinical validation: 4-6 months
- Compliance review: 2-3 months
Cost: $1.5M - $5.2M
- Clinical informaticists: $500K - $1.2M
- Software development: $600K - $2.4M
- Mobile app development: $200K - $800K
- Integration: $150K - $600K
- Validation studies: $50K - $200K
JustCopy.ai provides production-ready triage systems deployable in 4-6 weeks:
JustCopy.ai Deployment: 4-6 weeks
- Platform configuration: 5-7 days
- Workflow customization: 7-10 days
- EHR integration: 7-10 days
- Clinical validation review: 7-10 days
- Production deployment: 3-5 days
Cost: $45,000 - $125,000
- 97% cost reduction vs. custom build
- 95% faster time-to-market
- Pre-validated ESI algorithms
- Mobile apps included
- 10 AI agents handle deployment
Conclusion
Clinical triage systems are mission-critical applications where accuracy and reliability directly impact patient safety. This guide has demonstrated the technical complexity involved in building production-ready triage systems with ESI integration, red flag detection, mobile accessibility, and HIPAA compliance.
For most healthcare organizations, JustCopy.ai’s platform approach provides the optimal path to deployment: pre-built, clinically validated triage systems deployable in weeks instead of the 18-24 months required for custom development, at a fraction of the cost, with continuous improvements and expert support included.
Ready to deploy a clinical triage system without the 24-month development cycle? Start with JustCopy.ai’s pre-built templates and have your system operational in under 6 weeks.
Related Articles
🚀
Build This with JustCopy.ai
Skip months of development with 10 specialized AI agents. JustCopy.ai can copy, customize, and deploy this application instantly. Our AI agents write code, run tests, handle deployment, and monitor your application—all following healthcare industry best practices and HIPAA compliance standards.