File size: 4,675 Bytes
56edde7 480f467 56edde7 480f467 56edde7 1915c66 480f467 1915c66 56edde7 480f467 56edde7 480f467 56edde7 1915c66 480f467 1915c66 480f467 56edde7 480f467 56edde7 480f467 56edde7 480f467 56edde7 480f467 1915c66 56edde7 480f467 1915c66 480f467 56edde7 1915c66 480f467 56edde7 1915c66 480f467 1915c66 480f467 56edde7 1915c66 56edde7 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 |
import json
import logging
from groq import Groq
from config.settings import settings
from utils.prompts import get_coding_prompt
logger = logging.getLogger(__name__)
class GroqService:
def __init__(self):
"""Initialize Groq service (client created on first use)"""
self._client = None
self.model = settings.groq_model
@property
def client(self):
"""Lazy initialization of Groq client"""
if self._client is None:
if not settings.groq_api_key:
logger.error("GROQ_API_KEY is not set!")
raise ValueError("GROQ_API_KEY environment variable is required")
try:
logger.info("Initializing Groq client...")
self._client = Groq(api_key=settings.groq_api_key)
logger.info(f"Groq client initialized successfully")
except Exception as e:
logger.error(f"Failed to initialize Groq client: {str(e)}")
raise
return self._client
def analyze_provider_notes(self, provider_notes: str) -> dict:
"""
Analyze provider notes and extract ICD-10 and CPT codes
Args:
provider_notes: Clinical provider notes text
Returns:
dict: Parsed coding response with ICD-10, CPT codes and summary
"""
try:
logger.info(f"Analyzing provider notes (length: {len(provider_notes)})")
# Get the prompt
prompt = get_coding_prompt(provider_notes)
# Call Groq API (client initialized here if needed)
logger.info(f"Calling Groq API with model: {self.model}")
response = self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "system",
"content": "You are a medical coding expert specializing in ICD-10 and CPT coding. Always respond with valid JSON."
},
{
"role": "user",
"content": prompt
}
],
temperature=0.3,
max_tokens=2000,
response_format={"type": "json_object"}
)
# Extract response
raw_response = response.choices[0].message.content
logger.info(f"Received response from Groq (length: {len(raw_response)})")
# Parse JSON
parsed_response = json.loads(raw_response)
# Validate and structure response
result = self._structure_response(parsed_response)
logger.info(f"Successfully analyzed: {len(result.get('icd_codes', []))} ICD codes, {len(result.get('cpt_codes', []))} CPT codes")
return result
except json.JSONDecodeError as e:
logger.error(f"JSON parsing error: {str(e)}")
raise ValueError(f"Failed to parse Groq response as JSON: {str(e)}")
except Exception as e:
logger.error(f"Error analyzing provider notes: {str(e)}")
raise
def _structure_response(self, parsed_response: dict) -> dict:
"""Structure and validate the response from Groq"""
icd_codes = []
cpt_codes = []
overall_summary = parsed_response.get("overall_summary", "")
# Extract ICD codes
raw_icd = parsed_response.get("icd_codes", [])
if isinstance(raw_icd, list):
for icd in raw_icd:
if isinstance(icd, dict):
icd_codes.append({
"code": icd.get("code", ""),
"description": icd.get("description", ""),
"explanation": icd.get("explanation", "")
})
# Extract CPT codes
raw_cpt = parsed_response.get("cpt_codes", [])
if isinstance(raw_cpt, list):
for cpt in raw_cpt:
if isinstance(cpt, dict):
cpt_codes.append({
"code": cpt.get("code", ""),
"description": cpt.get("description", ""),
"explanation": cpt.get("explanation", "")
})
return {
"icd_codes": icd_codes,
"cpt_codes": cpt_codes,
"overall_summary": overall_summary
}
# Global instance (client initialized on first use)
groq_service = GroqService() |