Spaces:
Build error
Build error
| # hedis_engine.py | |
| # hedis_engine.py - Updated with Interactive Features | |
| import os | |
| import json | |
| from crewai import Agent, Task, Crew, Process | |
| from crewai_tools import tool | |
| from langchain_community.vectorstores import Chroma | |
| from langchain_openai import ChatOpenAI | |
| class HedisComplianceEngine: | |
| """ | |
| Crew that: (1) states official HEDIS measure criteria, | |
| (2) extracts patient evidence from embeddings, and | |
| (3) produces a markdown analysis report. | |
| Now includes interactive feedback capabilities. | |
| """ | |
| def __init__(self, vectordb: Chroma, measure_code: str, measurement_year: int): | |
| self.vectordb = vectordb | |
| self.measure_code = measure_code | |
| self.measurement_year = measurement_year | |
| self.model = ChatOpenAI(model="gpt-4o", temperature=0, max_tokens=5096) | |
| self.hedis_criteria = None # Store approved criteria | |
| def patient_chart_search(query: str) -> str: | |
| """Return top patient chart passages relevant to the query.""" | |
| results = self.vectordb.similarity_search(query, k=15) | |
| return "\n".join([res.page_content for res in results]) | |
| # 1) HEDIS measure expert (external knowledge via model) | |
| self.hedis_expert = Agent( | |
| role="HEDIS Measure Expert", | |
| goal=( | |
| f"Provide accurate, official details for HEDIS measure {measure_code} " | |
| "including required tests, codes, timeframes, and exclusions. " | |
| "ALWAYS respond with valid JSON format only." | |
| ), | |
| verbose=True, | |
| memory=False, | |
| tools=[], # design choice: no web/search tools here; rely on model knowledge | |
| backstory=( | |
| "Healthcare quality expert with deep knowledge of NCQA HEDIS specifications. " | |
| "Must compute concrete absolute date ranges given the measurement year. " | |
| "CRITICAL: Always respond with properly formatted JSON without any markdown or extra text." | |
| ), | |
| llm=self.model, | |
| ) | |
| # 2) Patient chart analyzer (RAG over embeddings) | |
| self.patient_analyzer = Agent( | |
| role="Patient Chart Analyzer", | |
| goal=f"Extract tests, diagnoses, procedures, meds & dates relevant to HEDIS {measure_code} only. Don't include anything which is not related to this measure.", | |
| verbose=True, | |
| memory=False, | |
| tools=[patient_chart_search], | |
| backstory="Specialist in retrieving structured clinical data from patient chart embeddings.", | |
| llm=self.model, | |
| ) | |
| # 3) Reporter who composes a clean markdown report | |
| self.hedis_chart_reporter = Agent( | |
| role="HEDIS Chart Findings Reporter", | |
| goal=f"Create a comprehensive markdown report of findings for HEDIS {measure_code}.", | |
| verbose=True, | |
| memory=False, | |
| tools=[patient_chart_search], | |
| backstory=( | |
| "Clinical documentation specialist who organizes evidence and highlights " | |
| "relevant data for quality measure analysis." | |
| ), | |
| llm=self.model, | |
| ) | |
| def get_hedis_criteria_with_feedback(self, suggestions: str = "") -> dict: | |
| """ | |
| Get HEDIS criteria from expert and present to user for feedback. | |
| Returns approved criteria or None if user exits. | |
| """ | |
| # Build the task description carefully to avoid string formatting issues | |
| base_description = f"""Provide the official criteria for HEDIS measure {self.measure_code}, including: | |
| - Required tests or procedures | |
| - Diagnosis/procedure codes | |
| - Required medications | |
| - Inclusion and exclusion criteria | |
| - Timeframes for each requirement (compute absolute date ranges using {self.measurement_year}) | |
| You MUST compute and include absolute date ranges using {self.measurement_year}.""" | |
| if suggestions: | |
| base_description += f"\n\nIncorporate these suggestions: {suggestions}" | |
| formatting_requirements = """ | |
| CRITICAL FORMATTING REQUIREMENTS: | |
| - Return ONLY valid JSON without any markdown formatting, code blocks, or extra text | |
| - Do not wrap the JSON in code blocks | |
| - Do not add any explanatory text before or after the JSON | |
| - The response must start with {{ and end with }} | |
| - Ensure all JSON keys and string values are properly quoted | |
| - Required JSON fields: measure, required_tests, required_medications, codes, timeframes, inclusions, exclusions""" | |
| full_description = base_description + formatting_requirements | |
| hedis_task = Task( | |
| description=full_description, | |
| expected_output="Valid JSON object with required fields (no markdown, no code blocks)", | |
| agent=self.hedis_expert, | |
| ) | |
| # Create temporary crew just for getting criteria | |
| criteria_crew = Crew( | |
| agents=[self.hedis_expert], | |
| tasks=[hedis_task], | |
| process=Process.sequential, | |
| verbose=True, | |
| ) | |
| # Execute to get criteria | |
| inputs = { | |
| "measure_code": self.measure_code, | |
| "measurement_year": self.measurement_year, | |
| } | |
| try: | |
| result = criteria_crew.kickoff(inputs=inputs) | |
| print(f"Raw crew result type: {type(result)}") | |
| print(f"Raw crew result: {repr(str(result)[:200])}...") # Show first 200 chars | |
| except Exception as e: | |
| print(f"Crew execution error: {e}") | |
| return {"raw_output": f"Crew execution failed: {str(e)}", "parse_error": str(e)} | |
| try: | |
| # Clean the result string - remove markdown code blocks and extra whitespace | |
| result_str = str(result).strip() | |
| # Remove markdown code blocks if present | |
| if result_str.startswith("```json"): | |
| result_str = result_str[7:] # Remove ```json | |
| elif result_str.startswith("```"): | |
| result_str = result_str[3:] # Remove ``` | |
| if result_str.endswith("```"): | |
| result_str = result_str[:-3] # Remove closing ``` | |
| # Remove any leading/trailing whitespace and newlines | |
| result_str = result_str.strip() | |
| # Handle quoted JSON strings | |
| if len(result_str) >= 2 and result_str.startswith('"') and result_str.endswith('"'): | |
| try: | |
| # Try to unquote and parse | |
| unquoted = result_str[1:-1] | |
| # Handle escaped quotes in the JSON | |
| unquoted = unquoted.replace('\\"', '"').replace('\\n', '\n') | |
| test_parse = json.loads(unquoted) | |
| result_str = unquoted | |
| except: | |
| # If unquoting fails, keep original | |
| pass | |
| # Try to parse as JSON | |
| criteria_json = json.loads(result_str) | |
| return criteria_json | |
| except json.JSONDecodeError as e: | |
| print(f"Primary method failed, trying simplified approach...") | |
| try: | |
| return self.get_hedis_criteria_simple(suggestions) | |
| except Exception as fallback_error: | |
| print(f"Fallback method also failed: {fallback_error}") | |
| return {"raw_output": str(result), "parse_error": f"Primary: {str(e)}, Fallback: {str(fallback_error)}"} | |
| except Exception as e: | |
| print(f"Unexpected error: {e}") | |
| return {"raw_output": str(result) if 'result' in locals() else "No result", "parse_error": str(e)} | |
| def get_hedis_criteria_simple(self, suggestions: str = "") -> dict: | |
| """ | |
| Simplified method to get HEDIS criteria with minimal string formatting. | |
| """ | |
| try: | |
| # Create a simple, direct prompt | |
| prompt_parts = [ | |
| f"Provide HEDIS measure {self.measure_code} criteria for year {self.measurement_year}.", | |
| "Include: required tests, medications, codes, timeframes, inclusions, exclusions.", | |
| ] | |
| if suggestions: | |
| prompt_parts.append(f"Suggestions: {suggestions}") | |
| prompt_parts.append("Respond with valid JSON only. No markdown, no code blocks, just JSON.") | |
| simple_description = " ".join(prompt_parts) | |
| hedis_task = Task( | |
| description=simple_description, | |
| expected_output="JSON object with measure criteria", | |
| agent=self.hedis_expert, | |
| ) | |
| # Create temporary crew | |
| criteria_crew = Crew( | |
| agents=[self.hedis_expert], | |
| tasks=[hedis_task], | |
| process=Process.sequential, | |
| verbose=True, | |
| ) | |
| # Execute | |
| inputs = {"measure_code": self.measure_code, "measurement_year": self.measurement_year} | |
| result = criteria_crew.kickoff(inputs=inputs) | |
| # Try to parse the result | |
| result_str = str(result).strip() | |
| # Basic cleaning | |
| if result_str.startswith("```"): | |
| lines = result_str.split('\n') | |
| # Remove first and last lines if they contain ``` | |
| if lines[0].strip().startswith('```'): | |
| lines = lines[1:] | |
| if lines and lines[-1].strip() == '```': | |
| lines = lines[:-1] | |
| result_str = '\n'.join(lines).strip() | |
| return json.loads(result_str) | |
| except Exception as e: | |
| print(f"Simple method error: {e}") | |
| return {"raw_output": str(result) if 'result' in locals() else "No result", "parse_error": str(e)} | |
| def display_criteria(self, criteria: dict) -> None: | |
| """Display the HEDIS criteria in a user-friendly format.""" | |
| print(f"\n{'='*60}") | |
| print(f"HEDIS MEASURE {self.measure_code} CRITERIA") | |
| print(f"Measurement Year: {self.measurement_year}") | |
| print(f"{'='*60}") | |
| if "raw_output" in criteria: | |
| print(criteria["raw_output"]) | |
| else: | |
| if "measure" in criteria: | |
| print(f"\n📋 Measure: {criteria['measure']}") | |
| if "required_tests" in criteria: | |
| print(f"\n🔬 Required Tests/Procedures:") | |
| if isinstance(criteria['required_tests'], list): | |
| for test in criteria['required_tests']: | |
| print(f" • {test}") | |
| else: | |
| print(f" {criteria['required_tests']}") | |
| if "required_medications" in criteria: | |
| print(f"\n💊 Required Medications:") | |
| if isinstance(criteria['required_medications'], list): | |
| for med in criteria['required_medications']: | |
| print(f" • {med}") | |
| else: | |
| print(f" {criteria['required_medications']}") | |
| if "codes" in criteria: | |
| print(f"\n🏷️ Relevant Codes:") | |
| if isinstance(criteria['codes'], dict): | |
| for code_type, codes in criteria['codes'].items(): | |
| print(f" {code_type}: {codes}") | |
| else: | |
| print(f" {criteria['codes']}") | |
| if "timeframes" in criteria: | |
| print(f"\n⏰ Timeframes:") | |
| if isinstance(criteria['timeframes'], dict): | |
| for requirement, timeframe in criteria['timeframes'].items(): | |
| print(f" {requirement}: {timeframe}") | |
| else: | |
| print(f" {criteria['timeframes']}") | |
| if "inclusions" in criteria: | |
| print(f"\n✅ Inclusion Criteria:") | |
| if isinstance(criteria['inclusions'], list): | |
| for inclusion in criteria['inclusions']: | |
| print(f" • {inclusion}") | |
| else: | |
| print(f" {criteria['inclusions']}") | |
| if "exclusions" in criteria: | |
| print(f"\n❌ Exclusion Criteria:") | |
| if isinstance(criteria['exclusions'], list): | |
| for exclusion in criteria['exclusions']: | |
| print(f" • {exclusion}") | |
| else: | |
| print(f" {criteria['exclusions']}") | |
| def get_user_feedback(self) -> tuple[str, str]: | |
| """ | |
| Get user feedback on the displayed criteria. | |
| Returns (action, suggestions) where action is 'continue', 'modify', or 'exit' | |
| """ | |
| print(f"\n{'='*60}") | |
| print("REVIEW THE CRITERIA ABOVE") | |
| print(f"{'='*60}") | |
| print("Options:") | |
| print("• Press ENTER to continue with these criteria") | |
| print("• Type suggestions to modify the criteria") | |
| print("• Type 'exit' to stop the analysis") | |
| print(f"{'='*60}") | |
| user_input = input("\nYour choice: ").strip() | |
| if user_input.lower() == 'exit': | |
| return 'exit', '' | |
| elif user_input == '': | |
| return 'continue', '' | |
| else: | |
| return 'modify', user_input | |
| def run_interactive(self) -> str: | |
| """ | |
| Run the interactive HEDIS compliance analysis with user feedback loop. | |
| """ | |
| print(f"Starting Interactive HEDIS Analysis for measure {self.measure_code}") | |
| suggestions = "" | |
| while True: | |
| # Get criteria from expert | |
| print(f"\n🔍 Getting HEDIS criteria for {self.measure_code}...") | |
| criteria = self.get_hedis_criteria_with_feedback(suggestions) | |
| # Display criteria to user | |
| self.display_criteria(criteria) | |
| # Get user feedback | |
| action, user_suggestions = self.get_user_feedback() | |
| if action == 'exit': | |
| print("\n❌ Analysis cancelled by user.") | |
| return "Analysis cancelled by user." | |
| elif action == 'continue': | |
| print("\n✅ Criteria approved! Proceeding with chart analysis...") | |
| self.hedis_criteria = criteria | |
| break | |
| elif action == 'modify': | |
| print(f"\n🔄 Modifying criteria based on your suggestions...") | |
| suggestions = user_suggestions | |
| continue | |
| # Now proceed with the rest of the analysis using approved criteria | |
| return self.run_full_analysis() | |
| def run_full_analysis(self) -> str: | |
| """ | |
| Run the full analysis with approved criteria. | |
| """ | |
| # Create tasks for patient analysis and reporting | |
| patient_task = Task( | |
| description=( | |
| f"Using the patient_chart_search tool, extract only the tests, diagnoses, procedures, " | |
| f"medications, and dates relevant to {self.measure_code}. Only include measure related info from the patient charts. " | |
| f"Use these approved criteria as reference: {json.dumps(self.hedis_criteria, indent=2)}" | |
| ), | |
| expected_output="Structured list of patient evidence with dates.", | |
| agent=self.patient_analyzer, | |
| async_execution=True, | |
| ) | |
| chart_analysis_task = Task( | |
| description=( | |
| f"Create a comprehensive markdown report analyzing the patient chart for HEDIS {self.measure_code} " | |
| f"for measurement year {self.measurement_year}.\n\n" | |
| "Sections:\n" | |
| "### HEDIS Measure Criteria (Approved)\n" | |
| "### Diagnoses Found\n" | |
| "### Procedures Found\n" | |
| "### Laboratory Tests\n" | |
| "### Medications\n" | |
| "### Vital Signs\n" | |
| "### Exclusions Found\n" | |
| "### Missing Information\n" | |
| "### Compliance Summary\n" | |
| f"Use these approved criteria as reference: {json.dumps(self.hedis_criteria, indent=2)}\n" | |
| "Make sure only measure related information is included." | |
| ), | |
| expected_output="A thorough markdown report of HEDIS findings with compliance assessment.", | |
| agent=self.hedis_chart_reporter, | |
| context=[patient_task], | |
| ) | |
| # Create crew for full analysis | |
| analysis_crew = Crew( | |
| agents=[self.patient_analyzer, self.hedis_chart_reporter], | |
| tasks=[patient_task, chart_analysis_task], | |
| process=Process.sequential, | |
| verbose=True, | |
| ) | |
| # Execute full analysis | |
| inputs = { | |
| "measure_code": self.measure_code, | |
| "measurement_year": self.measurement_year, | |
| } | |
| print(f"\n🔄 Running patient chart analysis...") | |
| result = analysis_crew.kickoff(inputs=inputs) | |
| print(f"\n✅ Analysis complete!") | |
| return str(result) | |
| def run(self) -> str: | |
| """ | |
| Entry point - runs original non-interactive analysis for app compatibility. | |
| Use run_interactive() for interactive mode. | |
| """ | |
| # Original tasks | |
| hedis_task = Task( | |
| description=( | |
| "Provide the official criteria for HEDIS measure {measure_code}, including:\n" | |
| "- Required tests or procedures\n" | |
| "- Diagnosis/procedure codes\n" | |
| "- Required medications\n" | |
| "- Inclusion and exclusion criteria\n" | |
| "- Timeframes for each requirement (e.g., within 10 years of {measurement_year})\n\n" | |
| "You MUST compute and include absolute date ranges using {measurement_year}.\n" | |
| "Return JSON with fields: 'measure','required_tests','required_medications'," | |
| "'codes','timeframes','inclusions','exclusions'." | |
| ), | |
| expected_output="Structured JSON with absolute date ranges.", | |
| agent=self.hedis_expert, | |
| async_execution=True, | |
| ) | |
| patient_task = Task( | |
| description=( | |
| "Using the patient_chart_search tool, extract only the tests, diagnoses, procedures, " | |
| "medications, and dates relevant to {measure_code}. Only include measure related info from the patient charts" | |
| ), | |
| expected_output="Structured list of patient evidence with dates.", | |
| agent=self.patient_analyzer, | |
| async_execution=True, | |
| ) | |
| chart_analysis_task = Task( | |
| description=( | |
| "Create a comprehensive markdown report analyzing the patient chart for HEDIS {measure_code} " | |
| "for measurement year {measurement_year}.\n\n" | |
| "Sections:\n" | |
| "### Diagnoses Found\n" | |
| "### Procedures Found\n" | |
| "### Laboratory Tests\n" | |
| "### Medications\n" | |
| "### Vital Signs\n" | |
| "### Exclusions Found\n" | |
| "### Missing Information\n" | |
| "### Summary\n" | |
| "Make sure only measure related information is there." | |
| ), | |
| expected_output="A thorough markdown report of HEDIS findings.", | |
| agent=self.hedis_chart_reporter, | |
| context=[hedis_task, patient_task], | |
| ) | |
| crew = Crew( | |
| agents=[self.hedis_expert, self.patient_analyzer, self.hedis_chart_reporter], | |
| tasks=[hedis_task, patient_task, chart_analysis_task], | |
| process=Process.sequential, | |
| verbose=True, | |
| ) | |
| inputs = { | |
| "measure_code": self.measure_code, | |
| "measurement_year": self.measurement_year, | |
| } | |
| result = crew.kickoff(inputs=inputs) | |
| return str(result) |