Update hedis_engine.py

hedis_engine.py

@@ -1,18 +1,16 @@
 # hedis_engine.py
+# interactive_hedis_engine.py
 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:
+class InteractiveHedisComplianceEngine:
     """
-    Crew that: (1) states official HEDIS measure criteria,
+    Interactive Crew that: (1) states official HEDIS measure criteria and gets user feedback,
     (2) extracts patient evidence from embeddings, and
     (3) produces a markdown analysis report.
     """
@@ -21,6 +19,7 @@ class HedisComplianceEngine:
         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
 
         @tool("patient_chart_search")
         def patient_chart_search(query: str) -> str:
@@ -70,16 +69,248 @@ class HedisComplianceEngine:
             llm=self.model,
         )
 
-        # Tasks
-        self.hedis_task = Task(
+    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.
+        """
+        # Create task for getting HEDIS criteria
+        hedis_task_description = (
+            f"Provide the official criteria for HEDIS measure {self.measure_code}, including:\n"
+            "- Required tests or procedures\n"
+            "- Diagnosis/procedure codes\n"
+            "- Required medications\n"
+            "- Inclusion and exclusion criteria\n"
+            f"- Timeframes for each requirement (e.g., within X years of {self.measurement_year})\n\n"
+            f"You MUST compute and include absolute date ranges using {self.measurement_year}.\n"
+        )
+        
+        if suggestions:
+            hedis_task_description += f"\n\nIncorporate these suggestions: {suggestions}\n"
+        
+        hedis_task_description += (
+            "Return JSON with fields: 'measure','required_tests','required_medications',"
+            "'codes','timeframes','inclusions','exclusions'."
+        )
+
+        hedis_task = Task(
+            description=hedis_task_description,
+            expected_output="Structured JSON with absolute date ranges.",
+            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,
+        }
+        
+        result = criteria_crew.kickoff(inputs=inputs)
+        
+        try:
+            # Try to parse as JSON
+            criteria_json = json.loads(str(result))
+            return criteria_json
+        except json.JSONDecodeError:
+            # If not valid JSON, return as string
+            return {"raw_output": str(result)}
+
+    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=(
-                "Provide the official criteria for HEDIS measure {measure_code}, including:\n"
+                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_non_interactive(self) -> str:
+        """
+        Run the original non-interactive analysis (for backwards compatibility).
+        """
+        hedis_task = Task(
+            description=(
+                f"Provide the official criteria for HEDIS measure {self.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"
+                f"- Timeframes for each requirement (e.g., within X years of {self.measurement_year})\n\n"
+                f"You MUST compute and include absolute date ranges using {self.measurement_year}.\n"
                 "Return JSON with fields: 'measure','required_tests','required_medications',"
                 "'codes','timeframes','inclusions','exclusions'."
             ),
@@ -88,20 +319,20 @@ class HedisComplianceEngine:
             async_execution=True,
         )
 
-        self.patient_task = Task(
+        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"
+                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"
             ),
             expected_output="Structured list of patient evidence with dates.",
             agent=self.patient_analyzer,
             async_execution=True,
         )
 
-        self.chart_analysis_task = Task(
+        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"
+                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"
                 "### Diagnoses Found\n"
                 "### Procedures Found\n"
@@ -115,20 +346,42 @@ class HedisComplianceEngine:
             ),
             expected_output="A thorough markdown report of HEDIS findings.",
             agent=self.hedis_chart_reporter,
-            context=[self.hedis_task, self.patient_task],
+            context=[hedis_task, patient_task],
         )
 
-        self.crew = Crew(
+        crew = Crew(
             agents=[self.hedis_expert, self.patient_analyzer, self.hedis_chart_reporter],
-            tasks=[self.hedis_task, self.patient_task, self.chart_analysis_task],
+            tasks=[hedis_task, patient_task, chart_analysis_task],
             process=Process.sequential,
             verbose=True,
         )
 
-    def run(self) -> str:
         inputs = {
             "measure_code": self.measure_code,
             "measurement_year": self.measurement_year,
         }
-        result = self.crew.kickoff(inputs=inputs)
-        return str(result)
+        result = crew.kickoff(inputs=inputs)
+        return str(result)
+
+    def run(self) -> str:
+        """
+        Entry point - runs non-interactive analysis by default for app compatibility.
+        Use run_interactive() for interactive mode.
+        """
+        return self.run_non_interactive()
+
+
+# Usage example:
+"""
+# Initialize the engine
+vectordb = your_chroma_db  # Your Chroma vector database
+engine = InteractiveHedisComplianceEngine(
+    vectordb=vectordb,
+    measure_code="CDC-A1C",
+    measurement_year=2024
+)
+
+# Run interactive analysis
+result = engine.run()
+print(result)
+"""