Insurance Claims Triage at Scale

pip install requests pandas python-dotenv
export SWARMS_API_KEY="your-api-key-here"

import json
import os
from typing import Any

import pandas as pd
import requests
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("SWARMS_API_KEY")
BASE_URL = "https://api.swarms.world"

if not API_KEY:
    raise ValueError("SWARMS_API_KEY environment variable is required")

headers = {"x-api-key": API_KEY, "Content-Type": "application/json"}

TRIAGE_SYSTEM_PROMPT = (
    "You are a senior P&C insurance claims triage specialist. "
    "Given a claim record and the relevant policy summary, you output a STRICT JSON "
    "object with exactly these keys and nothing else:\n\n"
    "{\n"
    '  "coverage_match": "COVERED" | "PARTIAL" | "NOT_COVERED",\n'
    '  "coverage_rationale": "<one sentence citing the policy clause>",\n'
    '  "fraud_signal": "LOW" | "MEDIUM" | "HIGH",\n'
    '  "fraud_indicators": ["<short bullet>", "..."],\n'
    '  "payout_estimate_usd": <integer>,\n'
    '  "payout_rationale": "<one sentence>",\n'
    '  "recommended_action": "AUTO_APPROVE" | "ADJUSTER_REVIEW" | "SIU_REFERRAL" | "DENY"\n'
    "}\n\n"
    "Be conservative on fraud — only HIGH if multiple indicators align. "
    "Auto-approve only when coverage is COVERED, fraud is LOW, and payout is under $5,000. "
    "Output JSON only. No prose."
)


def build_agent_payload(claim_row: dict) -> dict:
    """Map one claim row into the agent batch payload shape."""
    task = (
        "POLICY SUMMARY:\n"
        f"{claim_row['policy_summary']}\n\n"
        "CLAIM RECORD:\n"
        f"Claim ID: {claim_row['claim_id']}\n"
        f"Policyholder: {claim_row['policyholder']}\n"
        f"Loss Type: {claim_row['loss_type']}\n"
        f"Date of Loss: {claim_row['date_of_loss']}\n"
        f"Reported Amount: ${claim_row['reported_amount']}\n"
        f"Description: {claim_row['description']}\n"
        f"Prior Claims (12mo): {claim_row['prior_claims_12mo']}\n"
    )
    return {
        "agent_config": {
            "agent_name": "Claims Triage Specialist",
            "system_prompt": TRIAGE_SYSTEM_PROMPT,
            "model_name": "gpt-4.1",
            "max_tokens": 1024,
            "temperature": 0.1,
        },
        "task": task,
    }

# Example schema: claim_id, policyholder, policy_summary, loss_type,
# date_of_loss, reported_amount, description, prior_claims_12mo
claims_df = pd.read_csv("claims_inbox.csv")

print(f"Loaded {len(claims_df)} claims for triage")

def run_batch(payloads: list[dict]) -> list[dict]:
    response = requests.post(
        f"{BASE_URL}/v1/agent/batch/completions",
        headers=headers,
        json=payloads,
        timeout=900,
    )
    if response.status_code != 200:
        raise RuntimeError(f"Batch failed: {response.status_code} — {response.text[:300]}")
    return response.json()


def triage_dataframe(df: pd.DataFrame, chunk_size: int = 500) -> pd.DataFrame:
    all_results: list[dict[str, Any]] = []
    for start in range(0, len(df), chunk_size):
        chunk = df.iloc[start : start + chunk_size]
        payloads = [build_agent_payload(row.to_dict()) for _, row in chunk.iterrows()]
        print(f"Submitting rows {start}..{start + len(chunk) - 1}")
        results = run_batch(payloads)
        all_results.extend(results)
    return pd.DataFrame(all_results)


raw_results = triage_dataframe(claims_df)

def extract_decision(result_row: dict) -> dict:
    """Pull the JSON decision out of the agent response."""
    # The batch response carries the agent's text output under 'outputs' or 'output'
    # depending on schema version. Handle both.
    text = result_row.get("outputs") or result_row.get("output") or ""
    if isinstance(text, list):
        # Sometimes a list of message dicts — take the last assistant content
        for item in reversed(text):
            if isinstance(item, dict) and item.get("role") in ("assistant", "Claims Triage Specialist"):
                text = item.get("content", "")
                break
    if isinstance(text, list):
        text = " ".join(str(c) for c in text)
    try:
        return json.loads(str(text).strip())
    except json.JSONDecodeError:
        return {
            "coverage_match": "PARSE_ERROR",
            "fraud_signal": "PARSE_ERROR",
            "payout_estimate_usd": 0,
            "recommended_action": "ADJUSTER_REVIEW",
        }


decisions = pd.DataFrame(
    [extract_decision(r) for r in raw_results.to_dict(orient="records")]
)
decisions["claim_id"] = claims_df["claim_id"].values

triaged = claims_df.merge(decisions, on="claim_id", how="left")
triaged.to_csv("claims_triaged.csv", index=False)

print(triaged["recommended_action"].value_counts())