This project implements an automated insurance claim triage pipeline that integrates data ingestion, AI-based document analysis, and rule-based validation.
The pipeline processes claims from Google Sheets and Google Drive, analyzes documents using Gemini AI, and applies business validation rules to flag problematic claims.
The system is organized into a four-tier processing architecture.
Google Sheets + Google Drive
│
▼
Tier 1: Data Integration
- Download claim records
- Download supporting documents
- Remove duplicates
- Prepare clean dataset
│
▼
Tier 2: Claim Segmentation
- Separate claims with missing documents
- Prepare claims ready for AI analysis
│
▼
Tier 3: Gemini AI Extraction
- Document classification
- Item category detection
- Confidence score estimation
- AI recommendation
│
▼
Tier 4: Rule Validation
- Business rule enforcement
- Claim flagging
- Failure reason tracking
│
▼
Final Output
claims_with_flags.csv
flowchart TD
A[Start Pipeline] --> B[Run data_integration.py]
B --> C[Download Claim Sheet from Google Sheets]
C --> D[Remove duplicate ClaimId]
D --> E[Download Documents from Google Drive]
E --> F{Document Download Successful?}
F -->|No| G[Move Record to tier_1/fail]
F -->|Yes| H[Save to tier_1/runtime/drive]
H --> I[Generate Clean Claim Dataset]
I --> J[Save claims_ready_for_gemini.csv]
J --> K[Run gemini_extraction.py]
K --> L[Read claims_ready_for_gemini.csv]
L --> M[For each Claim]
M --> N[Upload document to Gemini]
N --> O[Gemini Analysis]
O --> P[Extract Structured Output]
P --> Q[classification]
O --> R[item_category]
O --> S[confidence_score]
O --> T[gemini_recommendation]
O --> U[analysis_summary]
Q --> V[Combine with claim data]
R --> V
S --> V
T --> V
U --> V
V --> W[Save extracted_claims.csv]
W --> X[Run validation.py]
X --> Y[Load extracted_claims.csv]
Y --> Z[Convert Dates]
Z --> AA[Apply Rule Engine]
AA --> AB{Rules Evaluation}
AB -->|Rule Violated| AC[Add failure_reasons]
AB -->|Pass| AD[Claim Valid]
AC --> AE[claim_flagged = True]
AD --> AF[claim_flagged = False]
AE --> AG[Save claims_with_flags.csv]
AF --> AG
AG --> AH[End Pipeline]
Responsible for data ingestion and preprocessing.
Main tasks:
- Download claim data from Google Sheets
- Remove duplicate Claim IDs
- Normalize document filenames
- Download claim documents from Google Drive
- Segment claims based on document availability
Outputs:
tier_1/YYYYMMDD/claimant_data.csv
tier_1/YYYYMMDD/drive/
tier_2/YYYYMMDD/success/claims_ready_for_gemini.csv
tier_2/YYYYMMDD/fail/claims_missing_documents.csv
Performs AI-powered claim analysis using Gemini 2.5 Flash.
Each document is analyzed for:
- Document classification
- Item category
- Confidence score
- AI recommendation
- Analysis summary
Results are stored in:
tier_3/YYYYMMDD/extracted_claims.csv
tier_3/YYYYMMDD/json/*.json
Applies rule-based claim validation using a configurable rule engine.
The validation layer flags claims that violate business constraints.
Examples of rules:
- Invalid document type
- Low classification confidence
- Inactive policy
- Claimant under 18
- Incident before policy start date
- Invalid item category
Final results are written to:
tier_4/YYYYMMDD/claims_with_flags.csv
Guardrails are implemented across multiple layers to ensure robust and reliable AI outputs.
Before sending data to the AI model:
- Duplicate Claim IDs are removed
- Missing document files are filtered
- Invalid filenames are normalized
- Example in Google Sheet file name: Shopee/ClaimDocument/FromShopee/c249953b-8f5a-4d40-8f63-8f302ad939b7.jpeg but in Google Drive file name: Shopee_ClaimDocument_FromShopee_c249953b-8f5a-4d40-8f63-8f302ad939b7.jpeg
- Claims without documents are excluded from AI processing
This ensures the AI model receives valid and consistent inputs.
The Gemini API response is constrained using structured schema validation via Pydantic.
Expected fields:
classification
item_category
confidence_score
gemini_recommendation
analysis_summary
Each field has defined constraints.
Example:
classification ∈ {Manufacturer Warranty Card, Certificate of Coverage, Unknown}
item_category ∈ {mechanical, electrical, electronic}
confidence_score ∈ [0,1]
If the model produces an invalid response, parsing fails and the claim is skipped.
After AI extraction, business rules enforce deterministic validation.
Examples:
Document must be Warranty Card or COC
Confidence score must exceed threshold
Policy must be active
Claimant must be >= 18 years old
Incident date must occur after policy start
Item category must be valid
If any rule fails:
claim_flagged = True
failure_reasons = list of violated rules
Several edge cases are handled throughout the pipeline.
Claims with missing documents are automatically moved to:
tier_2/YYYYMMDD/fail/
These claims are excluded from AI processing.
Before sending a document to Gemini:
if not os.path.exists(file_path):
return None
This prevents runtime crashes.
Duplicate Claim IDs are removed during ingestion:
df.drop_duplicates(subset=['ClaimId'])
This ensures each claim is processed once.
If Gemini fails for a document:
try:
extraction = process_with_gemini(row)
except:
continue
The pipeline continues processing remaining claims.
The prompt provided to Gemini is designed to ensure clear context and constrained output.
Example prompt:
Analyze this insurance claim:
Item Name: {itemname}
Incident Cause: {incidentcause}
Perform document classification must be
'Manufacturer Warranty Card',
'Certificate of Coverage',
or 'Unknown'.
Determine if the item category is
mechanical, electrical, or electronic.
Prompt design principles:
- Provide minimal but sufficient context
- Use explicit classification categories
- Avoid open-ended instructions
- Focus the model on structured extraction tasks
Several techniques are used to reduce hallucinations.
The model temperature is set to:
temperature = 0.1
This forces deterministic outputs.
The system instruction explicitly prohibits hallucination:
DO NOT hallucinate.
Use 'null' if information is missing.
Classify documents exactly as instructed.
The Gemini response must conform to a strict JSON schema.
Invalid responses are rejected.
The model is asked to perform only classification and structured extraction, not open reasoning.
Deterministic JSON output is enforced using Gemini structured output mode with Pydantic.
Example schema:
class ClaimAnalysis(BaseModel):
classification: str
item_category: str
confidence_score: float
gemini_recommendation: str
analysis_summary: str
The schema is passed to the Gemini API:
response_json_schema = ClaimAnalysis.model_json_schema()
The response format is enforced using:
response_mime_type = 'application/json'
This guarantees:
- consistent field names
- correct data types
- predictable structure
The JSON is then parsed using:
json.loads(response.text)
-
https://ai.google.dev/gemini-api/docs/structured-output?example=recipe
-
https://developers.google.com/workspace/drive/api/guides/manage-downloads#python