How to Build a claims processing Agent Using LangGraph in Python for healthcare

By Cyprian AaronsUpdated 2026-04-21
claims-processinglanggraphpythonhealthcare

A claims processing agent in healthcare takes a claim, checks it for completeness, validates it against policy and eligibility rules, flags issues, and routes it for auto-adjudication or human review. It matters because every manual touchpoint adds cost, delay, and compliance risk, and in healthcare those delays hit providers, payers, and patients directly.

Architecture

A production claims agent built with LangGraph needs these pieces:

  • Claim intake node

    • Normalizes inbound payloads from EDI 837, FHIR Claim resources, or internal JSON.
    • Validates required fields like member ID, provider NPI, CPT/HCPCS codes, diagnosis codes, and dates of service.
  • Eligibility and benefits node

    • Checks whether the member was active on the service date.
    • Verifies plan coverage, prior authorization requirements, and benefit limits.
  • Policy rules node

    • Applies payer-specific edits.
    • Checks coding conflicts, duplicate claims, bundling rules, and medical necessity constraints.
  • Risk and compliance node

    • Detects PHI exposure issues.
    • Enforces HIPAA-safe handling, audit logging, and jurisdiction/data residency constraints.
  • Decision router

    • Sends clean claims to auto-adjudication.
    • Routes exceptions to human review with reason codes.
  • Audit sink

    • Persists every state transition.
    • Stores decision traces for compliance review and dispute handling.

Implementation

1) Define the claim state

LangGraph works best when you make state explicit. For claims processing, keep the state small but complete: input payload, validation results, rule findings, decision outcome, and audit trail.

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
import operator

class ClaimState(TypedDict):
    claim_id: str
    member_id: str
    provider_npi: str
    procedure_codes: list[str]
    diagnosis_codes: list[str]
    service_date: str
    eligibility_ok: bool
    policy_ok: bool
    risk_flags: list[str]
    decision: str
    audit_log: Annotated[list[str], operator.add]

This pattern uses TypedDict for structure and Annotated[list[str], operator.add] so each node can append audit entries without overwriting prior ones.

2) Build deterministic validation nodes

Do not start with an LLM. Claims should first pass through deterministic checks that are easy to explain in audits. Use plain Python functions as LangGraph nodes.

def validate_claim(state: ClaimState) -> dict:
    missing = []
    for field in ["claim_id", "member_id", "provider_npi", "service_date"]:
        if not state.get(field):
            missing.append(field)

    if missing:
        return {
            "decision": "reject",
            "risk_flags": [f"missing_fields:{','.join(missing)}"],
            "audit_log": [f"Validation failed: {missing}"]
        }

    return {"audit_log": ["Validation passed"]}


def check_eligibility(state: ClaimState) -> dict:
    # Replace with real payer/benefits lookup.
    eligible = state["member_id"].startswith("M")
    return {
        "eligibility_ok": eligible,
        "audit_log": [f"Eligibility check result: {eligible}"]
    }


def apply_policy_rules(state: ClaimState) -> dict:
    # Example rule: reject empty code sets or suspicious patterns.
    if not state["procedure_codes"] or not state["diagnosis_codes"]:
        return {
            "policy_ok": False,
            "risk_flags": ["incomplete_coding"],
            "audit_log": ["Policy check failed: incomplete coding"]
        }

    return {"policy_ok": True, "audit_log": ["Policy check passed"]}

These nodes are boring by design. In healthcare workflows, boring means testable and defensible.

3) Add routing logic for auto-adjudication vs review

Use add_conditional_edges to branch based on the accumulated state. This is where LangGraph becomes useful for claims workflows because you can keep the process deterministic while still supporting exception handling.

def route_claim(state: ClaimState) -> str:
    if state.get("decision") == "reject":
        return "end"
    if not state.get("eligibility_ok"):
        return "manual_review"
    if not state.get("policy_ok"):
        return "manual_review"
    if state.get("risk_flags"):
        return "manual_review"
    return "auto_adjudicate"


def manual_review(state: ClaimState) -> dict:
    return {
        "decision": "manual_review",
        "audit_log": ["Routed to human reviewer"]
    }


def auto_adjudicate(state: ClaimState) -> dict:
    return {
        "decision": "approved",
        "audit_log": ["Auto-adjudicated as approved"]
    }

Now assemble the graph:

workflow = StateGraph(ClaimState)

workflow.add_node("validate_claim", validate_claim)
workflow.add_node("check_eligibility", check_eligibility)
workflow.add_node("apply_policy_rules", apply_policy_rules)
workflow.add_node("manual_review", manual_review)
workflow.add_node("auto_adjudicate", auto_adjudicate)

workflow.add_edge(START, "validate_claim")
workflow.add_edge("validate_claim", "check_eligibility")
workflow.add_edge("check_eligibility", "apply_policy_rules")

workflow.add_conditional_edges(
    "apply_policy_rules",
    route_claim,
    {
        "manual_review": "manual_review",
        "auto_adjudicate": "auto_adjudicate",
        "end": END,
    }
)

workflow.add_edge("manual_review", END)
workflow.add_edge("auto_adjudicate", END)

app = workflow.compile()

4) Run the graph with a real claim payload

Invoke the compiled graph with a claim object. In production this would come from an API gateway or message queue consumer.

initial_state = {
    "claim_id": "C12345",
    "member_id": "M99881",
    "provider_npi": "1234567890",
    "procedure_codes": ["99213"],
    "diagnosis_codes": ["J06.9"],
    "service_date": "2026-04-01",
    "eligibility_ok": False,
    "policy_ok": False,
    "risk_flags": [],
}

result = app.invoke(initial_state)
print(result["decision"])
print(result["audit_log"])

If you want to persist intermediate steps for traceability during debugging or operations review, use LangGraph’s stream method:

for event in app.stream(initial_state):
    print(event)

That gives you step-level visibility without wiring your own orchestration layer.

Production Considerations

  • Compliance and audit

    • Log every node transition with claim ID, timestamp, rule version, and decision reason.
    • Keep immutable audit records for appeals and payer disputes.
    • Avoid storing raw PHI in logs; redact member identifiers where possible.
  • Data residency

    • Keep claim data inside approved regional infrastructure.
    • If you use model endpoints or external services, verify where prompts and outputs are processed and stored.
    • For multi-region deployments, pin processing to the correct jurisdiction before any enrichment step.
  • Guardrails

    • Separate deterministic rules from any LLM-based summarization or explanation generation.
    • Never let an LLM decide coverage or payment on its own.
    • Require human approval for low-confidence cases or policy exceptions.
  • Monitoring

    • Track auto-adjudication rate, manual review rate, false rejects, and average time-to-decision.
    • Alert on spikes in missing fields or specific denial reasons; those usually indicate upstream integration issues.
    • Version your rules so you can explain why a claim was approved under one policy snapshot but denied under another.

Common Pitfalls

  • Putting LLMs in the critical path too early

    • Claims adjudication is mostly structured logic. If you let an LLM drive core decisions before deterministic checks exist, you create non-repeatable outcomes.
    • Fix it by using LLMs only for summarization of denial reasons or reviewer assistance after rules have run.
  • Weak state design

    • If your LangGraph state is just a blob of JSON with no typed contract, debugging becomes painful fast.
    • Fix it by using TypedDict or Pydantic models for explicit fields like eligibility status, policy status, risk flags, and decision codes.
  • No audit trail per transition

    • A final decision alone is not enough in healthcare. You need to show how the system got there.
    • Fix it by appending audit entries in every node and persisting them alongside rule versions and input hashes.

Keep learning

By Cyprian Aarons, AI Consultant at Topiax.

Want the complete 8-step roadmap?

Grab the free AI Agent Starter Kit — architecture templates, compliance checklists, and a 7-email deep-dive course.

Get the Starter Kit

Related Guides