from tools.base import Tool import json import re from typing import Any, Dict, Optional from tools.utils import load_prompt, ToolExecutionError, logger def load_jurisdiction_requirements(jurisdiction: str): """ Load required fields and format string for a given jurisdiction from JSON config. Supports mapping facility names to jurisdictions. """ import os import json config_path = os.path.join(os.path.dirname(__file__), "ipc_reporting_requirements.json") with open(config_path, "r", encoding="utf-8") as f: data = json.load(f) # Facility to jurisdiction mapping facility_mapping = { "parkland": "Texas", "parkland health": "Texas", "parkland hospital": "Texas", "dallas county": "Texas", "ut southwestern": "Texas", "texas health": "Texas", "methodist": "Texas", "baylor": "Texas", "cleveland clinic": "Ohio", "mayo clinic": "Minnesota", "johns hopkins": "Maryland", "mass general": "Massachusetts", "ucla": "California", "ucsf": "California", "stanford": "California", "cedars-sinai": "California", "mount sinai": "New York", "nyu": "New York", "presbyterian": "New York", "memorial sloan": "New York" } # First try direct jurisdiction match key = jurisdiction.strip().upper() for k in data: if k.upper() == key: return data[k] # Check if it's a US state in the nested structure if "US_States" in data: for state_name, state_data in data["US_States"].items(): if state_name.upper() == key: return state_data # Try facility name mapping jurisdiction_lower = jurisdiction.strip().lower() for facility_name, mapped_jurisdiction in facility_mapping.items(): if facility_name in jurisdiction_lower: # Recursively look up the mapped jurisdiction return load_jurisdiction_requirements(mapped_jurisdiction) return data.get("default", {}) async def search_facility_location(facility_name: str) -> Dict[str, Any]: """ Search the internet to find the location (city, state) of a healthcare facility. Returns a dictionary with location information or an error. """ try: from tools.internet_search import InternetSearchTool search_tool = InternetSearchTool() # Search for the facility with location context search_query = f"{facility_name} hospital medical center location city state address" search_result = await search_tool.run(search_query) if isinstance(search_result, dict) and "error" in search_result: return {"error": "Unable to search for facility location"} # Try to extract state information from the search results search_text = str(search_result).lower() # Common state patterns in search results us_states = [ "alabama", "alaska", "arizona", "arkansas", "california", "colorado", "connecticut", "delaware", "florida", "georgia", "hawaii", "idaho", "illinois", "indiana", "iowa", "kansas", "kentucky", "louisiana", "maine", "maryland", "massachusetts", "michigan", "minnesota", "mississippi", "missouri", "montana", "nebraska", "nevada", "new hampshire", "new jersey", "new mexico", "new york", "north carolina", "north dakota", "ohio", "oklahoma", "oregon", "pennsylvania", "rhode island", "south carolina", "south dakota", "tennessee", "texas", "utah", "vermont", "virginia", "washington", "west virginia", "wisconsin", "wyoming" ] detected_state = None for state in us_states: if state in search_text: detected_state = state.title() break if detected_state: return {"state": detected_state, "search_result": search_result} else: return {"error": "Could not determine state from search results", "search_result": search_result} except Exception as e: return {"error": f"Search failed: {str(e)}"} class IPCReportingTool(Tool): def openai_spec(self, legacy=False): return { "name": self.name, "description": self.description, "parameters": self.args_schema } """ Tool for generating infection prevention and control (IPC) reports with current, up-to-date requirements. This tool uses real-time internet search to find the most current reporting requirements for specific organisms and jurisdictions, ensuring compliance with the latest public health guidelines. Features: - Searches online for current reporting requirements by organism and location - Extracts organism information from case summaries automatically - Determines location from facility names or direct city/state input - Generates comprehensive reports using current requirements - Includes proper metadata and source attribution Use this tool when healthcare workers need to: - Report infectious diseases to public health authorities - Understand current reporting requirements for specific organisms - Generate compliant infection control documentation - Find up-to-date reporting timelines and contact information """ def __init__(self) -> None: """ Initialize the IPCReportingTool with its name, description, and argument schema. """ super().__init__() self.name = "IPC_reporting" self.description = ( "Generate infection control reports using current, up-to-date reporting requirements found through online search. " "Searches for the latest public health reporting requirements for specific organisms and locations. " "Use this tool when users ask about reporting requirements, need current guidelines, " "or want help with infection control reporting for any facility or jurisdiction. " "IMPORTANT: When calling this tool, include ALL relevant information from the conversation in the case_summary, " "especially the specific organism/pathogen mentioned (e.g., 'typhus fever', 'VRE', 'MRSA'). " "Accepts facility names (e.g., 'Parkland Health', 'Mayo Clinic') or " "jurisdictions (e.g., 'Texas', 'CDC', 'WHO'). " "Automatically determines location and searches for current requirements online. " "Provides real-time, accurate reporting information instead of outdated static data. " "First discovers required fields for the jurisdiction, then generates formatted report." ) self.args_schema = { "type": "object", "properties": { "case_summary": { "type": "string", "description": "Complete summary including ALL relevant context from the conversation, especially the specific organism/pathogen mentioned (e.g., 'User asked about typhus fever reporting in Dallas TX', 'VRE infection at facility', 'MRSA case'). Include the organism name prominently." }, "jurisdiction": { "type": "string", "description": "Facility name (e.g., 'Parkland Health') or jurisdiction (e.g., 'Texas', 'CDC', 'WHO'). If unknown, will search for facility location." }, "facility": { "type": "string", "description": "Facility name (alternative to jurisdiction) - will be mapped to jurisdiction if provided." }, "city_state": { "type": "string", "description": "City and state (e.g., 'Dallas, Texas') - use when facility name is unclear or search fails" }, "fields": { "type": "object", "description": "Dictionary of required field values for report generation" } }, "required": ["case_summary"] } self._required_fields_cache: Optional[list[str]] = None self._format_instructions_cache: Optional[str] = None async def run( self, case_summary: str, jurisdiction: Optional[str] = None, city_state: Optional[str] = None, fields: Optional[Dict[str, Any]] = None, facility: Optional[str] = None # Legacy compatibility for OpenAI function calling ) -> Dict[str, Any]: """ Generate an IPC report by searching online for current reporting requirements. Uses real-time search to find up-to-date requirements for the specific organism and location. """ try: # Handle legacy 'facility' parameter for OpenAI function calling compatibility if facility and not jurisdiction: jurisdiction = facility elif facility and jurisdiction: # If both are provided, prefer jurisdiction but use facility as fallback pass if not jurisdiction or not str(jurisdiction).strip(): return { "missing_fields": ["jurisdiction"], "message": "Please specify the facility name or reporting jurisdiction (e.g., 'Parkland Health', 'Texas', 'CDC', 'WHO')." } logger.info(f"IPC Reporting: Processing jurisdiction '{jurisdiction}'") # Extract organism from case summary for targeted search organism = await self._extract_organism_from_case(case_summary) # Determine location for reporting requirements location = None if city_state: location = city_state else: # Try to detect location from facility name search_result = await search_facility_location(jurisdiction) if "state" in search_result and "city" in search_result: location = f"{search_result['city']}, {search_result['state']}" elif "state" in search_result: location = search_result["state"] else: return { "missing_fields": ["city_state"], "message": f"I couldn't find location information for '{jurisdiction}'. Please provide the city and state (e.g., 'Dallas, Texas') so I can search for current reporting requirements." } logger.info(f"IPC Reporting: Searching for current requirements - Organism: {organism}, Location: {location}") # Search for current reporting requirements online requirements_result = await self._search_current_reporting_requirements(organism, location) if "error" in requirements_result: return { "error": requirements_result["error"], "message": f"Unable to find current reporting requirements for {organism} in {location}. Please check the organism name and location." } required_fields = requirements_result.get("required_fields", []) reporting_info = requirements_result.get("reporting_info", "") if fields is None: # Phase-1: return discovered requirements return { "required_fields": required_fields, "organism": organism, "location": location, "reporting_info": reporting_info, "message": f"Current reporting requirements for {organism} in {location}:\n{reporting_info}\n\nPlease provide: {', '.join(required_fields)}" } # Phase-2: build the final report missing = [f for f in required_fields if f not in fields or not fields[f]] if missing: return { "missing_fields": missing, "message": f"Still missing required fields: {', '.join(missing)}" } # Generate the report using current requirements report = await self._generate_current_report(organism, location, case_summary, fields, requirements_result) file_name = f"ipc_report_{organism.lower().replace(' ','_')}_{location.lower().replace(' ','_').replace(',','')}.md" logger.info(f"IPC Reporting: Successfully generated current report") return { "report": report, "file_name": file_name, "organism": organism, "location": location } except Exception as e: logger.error(f"IPCReportingTool failed: {e}", exc_info=True) raise ToolExecutionError(f"IPCReportingTool failed: {e}") async def _extract_organism_from_case(self, case_summary: str, conversation_context: Optional[list] = None) -> str: """Extract the primary organism/pathogen from the case summary and conversation context.""" # Use LLM to extract organism name from case summary try: from core.utils.llm_connector import call_llm # If case_summary is empty or very short, try to use conversation context if not case_summary or len(case_summary.strip()) < 5: logger.info(f"IPC Reporting: Case summary is empty or too short: '{case_summary}'") # Try to extract from recent conversation context if conversation_context: logger.info("IPC Reporting: Attempting to extract organism from conversation context") recent_messages = [] for msg in conversation_context[-5:]: # Last 5 messages content = msg.get("content", "") if content and len(content.strip()) > 3: recent_messages.append(content) if recent_messages: case_summary = "\n".join(recent_messages) logger.info(f"IPC Reporting: Using conversation context as case summary: {case_summary[:100]}...") else: logger.info("IPC Reporting: No useful conversation context found") return "Unknown organism" else: logger.info("IPC Reporting: No conversation context provided") return "Unknown organism" prompt = f""" Extract the primary organism or pathogen from this case summary or conversation. Special instructions: - If "typhus" or "typhus fever" is mentioned, return "typhus fever" - If "VRE" is mentioned, return "VRE" - If "vancomycin-resistant enterococcus" is mentioned, return "VRE" - If "Enterococcus" with resistance is mentioned, return "VRE" - Otherwise return the specific organism name (e.g., "Candida albicans", "MRSA", "C. difficile") If multiple organisms are mentioned, return the most significant one for reporting purposes. If no specific organism is mentioned, return "Unknown organism". Text to analyze: {case_summary} Organism:""" organism = await call_llm(prompt) return organism.strip() except Exception as e: logger.error(f"Error extracting organism: {e}") # Fallback: try to extract manually search_text = case_summary.lower() if case_summary else "" common_organisms = [ "typhus fever", "typhus", "C. difficile", "Clostridium difficile", "MRSA", "VRE", "CRE", "Candida albicans", "Aspergillus", "Tuberculosis", "TB", "E. coli", "Klebsiella", "Pseudomonas", "Acinetobacter" ] for organism in common_organisms: if organism.lower() in search_text: return organism return "Unknown organism" async def _search_current_reporting_requirements(self, organism: str, location: str) -> Dict[str, Any]: """Search online for current reporting requirements for the specific organism and location.""" try: from tools.internet_search import InternetSearchTool search_tool = InternetSearchTool() # Extract state from location for broader search state = location.split(",")[-1].strip() if "," in location else location # Create multiple search strategies with different approaches search_strategies = [ # Strategy 1: Specific organism + state [ f"{organism} reportable disease {state} 2024 2025", f"{organism} notifiable disease {state} health department", f"VRE vancomycin resistant enterococcus reportable {state}" if "VRE" in organism else f"{organism} reportable {state}" ], # Strategy 2: General state reporting lists [ f"{state} reportable diseases list 2024 2025 health department", f"{state} notifiable diseases surveillance CDC", f"{state} public health reportable conditions list" ], # Strategy 3: CDC/national guidelines [ f"CDC reportable diseases {organism} surveillance", f"NHSN {organism} reporting requirements healthcare", f"vancomycin resistant enterococcus CDC reporting" if "VRE" in organism else f"{organism} CDC reporting" ] ] requirements_info = "" found_relevant_info = False # Try each strategy until we find good information for strategy_num, queries in enumerate(search_strategies, 1): logger.info(f"IPC Reporting: Trying search strategy {strategy_num}") for query in queries: logger.info(f"IPC Reporting: Searching with query: {query}") search_result = await search_tool.run(query) # InternetSearchTool returns a formatted string, not a dict with "results" if isinstance(search_result, str) and search_result.strip(): # Check if this result contains reporting requirements relevant_keywords = [ "reportable", "notifiable", "reporting requirements", "public health", "health department", "cdc", "nhsn", "surveillance", "reporting", "notify", "submit" ] search_text = search_result.lower() if any(keyword in search_text for keyword in relevant_keywords): logger.info(f"IPC Reporting: Found relevant search results for query: {query}") requirements_info += f"\n--- Search Results for: {query} ---\n{search_result}\n" found_relevant_info = True else: logger.info(f"IPC Reporting: Search results not relevant for query: {query}") elif isinstance(search_result, dict) and "results" in search_result: # Handle legacy dict format if it exists for result in search_result["results"][:5]: # Check top 5 results content = result.get("content", "") title = result.get("title", "") url = result.get("url", "") # Check if this result contains reporting requirements relevant_keywords = [ "reportable", "notifiable", "reporting requirements", "public health", "health department", "cdc", "nhsn", "surveillance", "reporting", "notify", "submit" ] if any(keyword in content.lower() for keyword in relevant_keywords): requirements_info += f"\n--- {title} ---\n{content[:800]}...\nSource: {url}\n" found_relevant_info = True else: logger.warning(f"IPC Reporting: Unexpected search result format: {type(search_result)}") # If we found good info in this strategy, stop searching if found_relevant_info and len(requirements_info) > 500: break if not requirements_info: # Try fallback for common antimicrobial-resistant organisms fallback_result = self._get_fallback_aro_requirements(organism, location) if fallback_result.get("required_fields"): logger.info(f"IPC Reporting: Using fallback requirements for {organism}") return fallback_result return {"error": f"No current reporting requirements found for {organism} in {location}"} # Use LLM to extract structured requirements from search results try: from core.utils.llm_connector import call_llm analysis_prompt = f""" Based on the following search results about {organism} (Vancomycin-Resistant Enterococcus) reporting requirements in {location}, extract the key information needed for infection control reporting. Search Results: {requirements_info} Please analyze and provide: 1. Is {organism} reportable in {location}? (Yes/No/Depends) 2. What are the required fields for reporting? List them clearly 3. What is the reporting timeline? (e.g., within 24 hours, immediately) 4. Who should receive the report? (e.g., local health department, CDC, hospital IPC team) 5. Any special requirements or notes? 6. If not specifically mentioned, what would be the likely reporting requirements based on similar organisms? Be specific about {state} requirements if found. If exact requirements for {organism} aren't found, extrapolate from general antimicrobial-resistant organism reporting requirements. Format your response as clear, actionable information that can guide healthcare workers. """ analysis = await call_llm(analysis_prompt) except Exception as e: logger.error(f"Error analyzing requirements: {e}") analysis = f"Current reporting requirements for {organism} in {location} (analysis unavailable due to error: {e})" # Extract required fields from the analysis required_fields = self._parse_required_fields_from_analysis(analysis) # Special handling for VRE if no specific info found if "VRE" in organism and ("no" in analysis.lower() or "not found" in analysis.lower()): # Provide general VRE reporting guidance analysis += f""" **General VRE Reporting Guidance:** VRE (Vancomycin-Resistant Enterococcus) is typically reportable to infection control teams and may be reportable to public health authorities depending on the jurisdiction. In {state}: - Hospital infection control team should be notified immediately - Contact isolation precautions should be implemented - Many states require reporting of antimicrobial-resistant organisms - CDC considers VRE a serious threat requiring surveillance - Check with your local/state health department for specific requirements **Recommended immediate actions:** 1. Implement contact precautions 2. Notify infection control team 3. Check facility-specific reporting protocols 4. Consider environmental cleaning protocols """ # Update required fields for VRE required_fields = [ "patient_demographics", "infection_type", "specimen_collection_date", "specimen_type", "antimicrobial_susceptibility", "isolation_status", "clinical_outcome", "infection_control_measures" ] return { "required_fields": required_fields, "reporting_info": analysis, "source_info": requirements_info[:1000] + "..." if len(requirements_info) > 1000 else requirements_info } except Exception as e: logger.error(f"Error searching for reporting requirements: {e}") return {"error": f"Search failed: {str(e)}"} def _parse_required_fields_from_analysis(self, analysis: str) -> list[str]: """Parse required fields from LLM analysis of reporting requirements.""" # Standard fields that are commonly required standard_fields = [ "patient_demographics", "onset_date", "specimen_collection_date", "specimen_type", "laboratory_results", "clinical_syndrome", "treatment_given", "outcome", "healthcare_facility", "reporting_facility" ] # Look for specific field mentions in the analysis analysis_lower = analysis.lower() found_fields = [] field_keywords = { "patient_demographics": ["patient", "demographics", "age", "gender", "name", "identifier"], "onset_date": ["onset", "symptom", "illness date", "onset date"], "specimen_collection_date": ["specimen", "collection", "sample date"], "specimen_type": ["specimen type", "sample type", "culture", "blood", "urine"], "laboratory_results": ["lab", "laboratory", "culture", "sensitivity", "antimicrobial"], "clinical_syndrome": ["syndrome", "diagnosis", "clinical", "infection type"], "treatment_given": ["treatment", "therapy", "antibiotic", "antifungal"], "outcome": ["outcome", "status", "recovery", "death", "discharge"], "healthcare_facility": ["facility", "hospital", "clinic", "location"], "reporting_facility": ["reporting", "laboratory", "reporter"] } for field, keywords in field_keywords.items(): if any(keyword in analysis_lower for keyword in keywords): found_fields.append(field) # Ensure we have at least basic required fields essential_fields = ["patient_demographics", "onset_date", "specimen_type", "laboratory_results"] for field in essential_fields: if field not in found_fields: found_fields.append(field) return found_fields[:8] # Limit to reasonable number of fields async def _generate_current_report( self, organism: str, location: str, case_summary: str, fields: Dict[str, Any], requirements_result: Dict[str, Any] ) -> str: """Generate a comprehensive report using current requirements.""" try: from core.utils.llm_connector import call_llm reporting_info = requirements_result.get("reporting_info", "") source_info = requirements_result.get("source_info", "") prompt = f""" Generate a comprehensive infection control report for {organism} in {location}. Case Summary: {case_summary} Provided Information: {json.dumps(fields, indent=2)} Current Reporting Requirements (from online search): {reporting_info} Please generate a professional infection control report that includes: 1. Case identification and basic demographics 2. Clinical presentation and onset information 3. Laboratory findings and organism details 4. Treatment and management 5. Infection control measures 6. Reporting compliance information 7. Contact tracing and follow-up as needed Format this as a formal report suitable for submission to public health authorities. Include the current date and ensure all provided information is incorporated appropriately. """ report = await call_llm(prompt) except Exception as e: logger.error(f"Error generating report: {e}") # Fallback: generate basic report template report = f""" INFECTION CONTROL REPORT Organism: {organism} Location: {location} Case Summary: {case_summary} Provided Information: {json.dumps(fields, indent=2)} Current Requirements: {requirements_result.get('reporting_info', 'Unable to retrieve current requirements')} Note: Report generation encountered an error: {e} """ # Add metadata footer from datetime import datetime current_date = datetime.now().strftime("%Y-%m-%d %H:%M") footer = f""" --- **Report Generated:** {current_date} **Organism:** {organism} **Location:** {location} **Generated by:** Infectious Diseases AI Assistant **Requirements Source:** Current online search results *Note: This report was generated using current reporting requirements found through online search. Please verify with local health department for any recent updates to reporting procedures.* """ return report + footer def _get_fallback_aro_requirements(self, organism: str, location: str) -> Dict[str, Any]: """Provide fallback requirements for common antimicrobial-resistant organisms (AROs).""" # Extract state for state-specific guidance state = location.split(",")[-1].strip() if "," in location else location # Common ARO patterns and typical requirements aro_patterns = { "VRE": { "full_name": "Vancomycin-Resistant Enterococcus", "typical_reportable": True, "urgency": "Contact isolation required immediately - Report within 24 hours", "common_requirements": [ "patient_demographics", "infection_type_and_site", "specimen_collection_date", "specimen_type", "vancomycin_mic_result", "antimicrobial_susceptibility_panel", "risk_factors_and_exposures", "isolation_precautions_implemented", "clinical_outcome", "infection_control_measures", "contact_tracing_performed", "environmental_culture_results" ] }, "MRSA": { "full_name": "Methicillin-Resistant Staphylococcus aureus", "typical_reportable": True, "urgency": "Contact isolation required", "common_requirements": [ "patient_demographics", "infection_type", "specimen_collection_date", "specimen_type", "antimicrobial_susceptibility", "healthcare_exposure", "isolation_status" ] }, "CRE": { "full_name": "Carbapenem-Resistant Enterobacteriaceae", "typical_reportable": True, "urgency": "Immediate reporting required - CDC priority pathogen", "common_requirements": [ "patient_demographics", "infection_type", "specimen_collection_date", "specimen_type", "carbapenemase_testing", "antimicrobial_susceptibility", "contact_tracing", "isolation_status" ] }, "Typhus": { "full_name": "Typhus Fever (Rickettsia species)", "typical_reportable": True, "urgency": "Immediate reporting required - CDC notifiable disease", "common_requirements": [ "patient_demographics", "onset_date", "clinical_presentation", "laboratory_confirmation", "epidemiological_factors", "travel_history", "vector_exposure_history", "specimen_collection_date", "treatment_given", "outcome", "contact_investigation" ] }, "Rickettsia": { "full_name": "Rickettsial Disease", "typical_reportable": True, "urgency": "Report within 24 hours - vector-borne disease surveillance", "common_requirements": [ "patient_demographics", "onset_date", "clinical_presentation", "laboratory_confirmation", "epidemiological_factors", "travel_history", "vector_exposure_history", "specimen_type_and_results", "treatment_response", "outcome" ] } } # Find matching organism - check for VRE patterns organism_upper = organism.upper() aro_info = None # Special check for VRE patterns if any(term in organism_upper for term in ["VRE", "VANCOMYCIN", "RESISTANT", "ENTEROCOCCUS"]): if any(vr_term in organism_upper for vr_term in ["VANCOMYCIN", "VRE"]) or any(ent_term in organism_upper for ent_term in ["ENTEROCOCCUS", "FAECIUM", "FAECALIS"]): aro_info = aro_patterns["VRE"] # Check other patterns if not VRE if not aro_info: for key, info in aro_patterns.items(): if key in organism_upper: aro_info = info break if not aro_info: return { "required_fields": ["patient_demographics", "specimen_type", "laboratory_results"], "reporting_info": f"Standard reporting requirements for {organism} - contact your facility's infection control team and local health department for specific requirements.", "source_info": "Fallback guidance" } reporting_info = f""" **{aro_info['full_name']} ({organism}) Reporting in {state}** **Immediate Requirements:** - {aro_info['urgency']} - Implement contact precautions immediately - Notify facility infection control team within 24 hours - Report to {state} Department of Health and Human Services (DHHS) **North Carolina Specific Guidance** (if {state} == "North Carolina"): - NC requires reporting of multidrug-resistant organisms to DHHS - Contact NC DHHS Communicable Disease Branch: (919) 733-3419 - Use the NC Electronic Disease Surveillance System (NC EDSS) if available - Follow NC infection control guidelines for healthcare facilities **Standard Reporting Elements:** {chr(10).join([f"• {field.replace('_', ' ').title()}" for field in aro_info['common_requirements']])} **Clinical Actions Required:** 1. Contact isolation precautions (gown, gloves, dedicated equipment) 2. Single room placement or cohorting with other VRE patients 3. Hand hygiene with soap and water or alcohol-based sanitizer 4. Environmental cleaning with appropriate disinfectants 5. Notify receiving facilities during transfers 6. Consider screening of contacts if outbreak suspected **Antimicrobial Stewardship Considerations:** - Review patient's antibiotic history - Optimize current antimicrobial therapy - Consider linezolid, daptomycin, or tigecycline for treatment - Avoid unnecessary vancomycin use to prevent further resistance *Note: This is general guidance based on standard practices. Always verify current requirements with your local health department and facility policies.* """ return { "required_fields": aro_info["common_requirements"], "reporting_info": reporting_info, "source_info": f"Fallback guidance for {aro_info['full_name']}" }