Spaces:
Sleeping
Sleeping
| import io | |
| import json | |
| from PIL import Image | |
| from google import genai | |
| from google.genai import types | |
| from app.core.config import settings | |
| from app.models.schemas import ExtractedClaimData, ItemizedClaimData | |
| from typing import List, Dict | |
| # Initialize the official modern Google GenAI Client | |
| # It automatically picks up GEMINI_API_KEY from the environment/settings | |
| client = genai.Client(api_key=settings.GEMINI_API_KEY) | |
| async def analyze_claim_documents(images_bytes_list: List[Dict[str, any]], rag_policy_context: str) -> ExtractedClaimData: | |
| """ | |
| Sends the medical document image and dynamic RAG policy context to Gemini 2.5 Flash. | |
| Enforces a strict structured JSON output matching the ExtractedClaimData schema. | |
| """ | |
| try: | |
| # Build the single-pass prompt combining extraction rules, RAG guidelines, and formatting schemas | |
| system_prompt = ( | |
| "You are an expert medical insurance claims adjudication engine. " | |
| "Your job is to extract data from the provided document image and evaluate it " | |
| "against the accompanying Policy Context constraints.\n\n" | |
| "STRICT RULES:\n" | |
| "1. Extract patient info, treatment dates, provider details, and itemized lists accurately.\n" | |
| "2. Cross-reference all extracted data against the provided Policy Context to determine " | |
| "if any items match listed exclusions.\n" | |
| "3. Assess medical necessity: verify if the clinical diagnosis logically justifies the " | |
| "prescribed drugs, procedures, or diagnostic tests.\n" | |
| "4. Return structural results matching the requested JSON schema exactly." | |
| "5.Cross-reference the Patient's Diagnosis and Treatment against the POLICY CONTEXT below. You must medically evaluate if the treatment is a synonym for an excluded category (e.g., Bariatric/Obesity = Weight Loss). If it matches an exclusion, you MUST set `is_treatment_excluded` to true and provide the exact `exclusion_reason`." | |
| "6.Check the `pre_authorization_required` list in the POLICY CONTEXT. If the billed items include a procedure on that list (like an MRI Scan), you MUST set `requires_pre_auth` to true. If the receipt/prescription does NOT explicitly mention an approved authorization code, you MUST also set `missing_pre_auth` to true." | |
| "CRITICAL - ITEMIZED EVALUATION (NOT reject all if one item fails):\n" | |
| "7. For EVERY line item (medicine, procedure, test, consultation), determine:\n" | |
| " - Is it in the policy exclusions list?\n" | |
| " - Is it medically necessary given the diagnosis?\n" | |
| " - Does it match standard clinical protocols?\n" | |
| "8. Mark is_covered and is_medically_necessary for each item INDEPENDENTLY.\n" | |
| "9. IMPORTANT: Do NOT reject entire claim—evaluate items separately.\n" | |
| " Some items may be covered, others excluded. That's OK—return individual assessments.\n" | |
| "10. Break down the bill into individual line items. For each item, populate the itemized_claims list.\n" | |
| "11. Return ONLY valid JSON matching the schema. No markdown, no code blocks.\n\n" | |
| "RESPONSE FORMAT - Return ONLY this JSON structure (no other text):\n" | |
| "{\n" | |
| ' "patient_name": "string",\n' | |
| ' "treatment_date": "YYYY-MM-DD",\n' | |
| ' "doctor_name": "string",\n' | |
| ' "doctor_reg_no": "string",\n' | |
| ' "diagnosis": "string",\n' | |
| ' "itemized_claims": [\n' | |
| ' {\n' | |
| ' "description": "item name",\n' | |
| ' "amount": 1000.50,\n' | |
| ' "category": "medicine|procedure|diagnostic_test|consultation",\n' | |
| ' "is_covered": true/false,\n' | |
| ' "exclusion_reason": "reason or None",\n' | |
| ' "is_medically_necessary": true/false,\n' | |
| ' "medical_necessity_reason": "reason"\n' | |
| " }\n" | |
| " ],\n" | |
| ' "total_billed_amount": 5000.00,\n' | |
| ' "overall_medically_necessary": true,\n' | |
| ' "medical_necessity_reasoning": "overall assessment"\n' | |
| "}" | |
| ) | |
| user_content = f""" | |
| POLICY CONTEXT (Retrieved rules for this claim context): | |
| ----------------------------------------------------- | |
| {rag_policy_context} | |
| ----------------------------------------------------- | |
| Please evaluate the attached medical document image according to the rules above. | |
| Extract each line item (medicine, procedure, test, consultation) separately. | |
| For EACH item, determine: is it covered by policy? Is it medically necessary? | |
| Return ONLY JSON, no explanations. | |
| """ | |
| contents = [user_content] | |
| for img_bytes in images_bytes_list: | |
| contents.append( | |
| types.Part.from_bytes( | |
| data=img_bytes["bytes"], | |
| mime_type=img_bytes["mime_type"] | |
| ) | |
| ) | |
| print(">>> 1. Preparing to call Gemini API...") | |
| # Execute the single-pass multimodal API call without strict schema (use text JSON instead) | |
| response = await client.aio.models.generate_content( | |
| model='gemini-2.5-flash', | |
| contents=contents, | |
| config=types.GenerateContentConfig( | |
| system_instruction=system_prompt, | |
| response_mime_type="application/json", | |
| temperature=0.1, # Low temperature ensures highly predictable, deterministic evaluation | |
| ) | |
| ) | |
| print(">>> 2. Gemini API Call Complete!") | |
| # Parse the response text as JSON and validate with Pydantic | |
| response_text = response.text.strip() | |
| # Remove markdown code blocks if present | |
| if response_text.startswith("```json"): | |
| response_text = response_text[7:] | |
| if response_text.startswith("```"): | |
| response_text = response_text[3:] | |
| if response_text.endswith("```"): | |
| response_text = response_text[:-3] | |
| response_text = response_text.strip() | |
| print(f"DEBUG: Raw Gemini Response:\n{response_text[:500]}") | |
| response_json = json.loads(response_text) | |
| return ExtractedClaimData.model_validate(response_json) | |
| except json.JSONDecodeError as je: | |
| print(f"❌ JSON Parse Error: {str(je)}") | |
| print(f"Response text: {response_text[:1000]}") | |
| raise RuntimeError(f"Failed to parse Gemini JSON response: {str(je)}") | |
| except Exception as e: | |
| print(f"❌ Gemini Adjudication Call Failed: {str(e)}") | |
| raise RuntimeError(f"Failed to process document via Gemini AI: {str(e)}") |