How to Build a fraud detection Agent Using LangGraph in Python for healthcare

A healthcare fraud detection agent reviews claims, provider activity, and supporting evidence to flag suspicious patterns before they become financial loss or compliance incidents. In practice, that means catching duplicate billing, upcoding, phantom services, and abnormal referral behavior while preserving an audit trail that compliance teams can defend.

Architecture

• •

  Claim intake node

  • •Normalizes incoming claim payloads from EHR, billing, or payer systems.
  • •Validates required fields like CPT/HCPCS codes, provider IDs, member IDs, dates of service, and facility location.
• •

  Policy/rule screening node

  • •Applies deterministic checks first.
  • •Flags obvious violations such as duplicate submissions, impossible date sequences, missing authorization, or out-of-network anomalies.
• •

  Risk scoring node

  • •Uses a model or heuristic scorer to assign a fraud likelihood.
  • •Combines claim metadata with provider history, utilization patterns, and prior denials.
• •

  Evidence retrieval node

  • •Pulls supporting context from internal systems.
  • •Examples: prior claims for the same member, provider credential status, pre-auth records, and historical coding patterns.
• •

  Decision node

  • •Produces one of three outcomes: approve, hold for review, or escalate to SIU/compliance.
  • •Returns structured reasons for auditability.
• •

  Audit logging node

  • •Persists every decision input/output with timestamps and trace IDs.
  • •This matters for HIPAA-aligned controls, internal investigations, and regulator review.

Implementation

1) Define the state and the graph nodes

Use StateGraph with a typed state object. Keep the state small and explicit so you can serialize it cleanly into logs or a case management system.

from typing import TypedDict, List, Dict, Any
from langgraph.graph import StateGraph, START, END

class FraudState(TypedDict):
    claim: Dict[str, Any]
    findings: List[str]
    risk_score: float
    decision: str
    audit: List[Dict[str, Any]]

def normalize_claim(state: FraudState) -> FraudState:
    claim = state["claim"]
    claim["provider_id"] = str(claim.get("provider_id", "")).strip()
    claim["member_id"] = str(claim.get("member_id", "")).strip()
    return {
        **state,
        "claim": claim,
        "findings": state.get("findings", []),
        "audit": state.get("audit", []) + [{"node": "normalize_claim"}],
    }

def rule_screen(state: FraudState) -> FraudState:
    findings = list(state.get("findings", []))
    claim = state["claim"]

    if claim.get("amount", 0) > 5000:
        findings.append("high_amount_claim")
    if claim.get("duplicate_flag"):
        findings.append("possible_duplicate_billing")

    return {
        **state,
        "findings": findings,
        "audit": state.get("audit", []) + [{"node": "rule_screen", "findings": findings}],
    }

def score_risk(state: FraudState) -> FraudState:
    score = 0.2
    if "high_amount_claim" in state["findings"]:
        score += 0.4
    if "possible_duplicate_billing" in state["findings"]:
        score += 0.5

    return {
        **state,
        "risk_score": min(score, 1.0),
        "audit": state.get("audit", []) + [{"node": "score_risk", "risk_score": min(score, 1.0)}],
    }

2) Add routing logic for escalation

LangGraph’s add_conditional_edges is the right fit when you need deterministic branching based on risk. In healthcare fraud workflows, this keeps low-risk claims moving while high-risk claims go to SIU review.

def decide_route(state: FraudState) -> str:
    if state["risk_score"] >= 0.8:
        return "escalate"
    if state["risk_score"] >= 0.4:
        return "review"
    return "approve"

def approve(state: FraudState) -> FraudState:
    return {**state, "decision": "approve"}

def review(state: FraudState) -> FraudState:
    return {**state, "decision": "hold_for_review"}

def escalate(state: FraudState) -> FraudState:
    return {**state, "decision": "escalate_to_siu"}

3) Compile the graph and run it

This is the actual LangGraph pattern you want in production code. Notice that each node returns a full updated state; that makes tracing and replay much easier.

builder = StateGraph(FraudState)

builder.add_node("normalize_claim", normalize_claim)
builder.add_node("rule_screen", rule_screen)
builder.add_node("score_risk", score_risk)
builder.add_node("approve", approve)
builder.add_node("review", review)
builder.add_node("escalate", escalate)

builder.add_edge(START, "normalize_claim")
builder.add_edge("normalize_claim", "rule_screen")
builder.add_edge("rule_screen", "score_risk")

builder.add_conditional_edges(
    "score_risk",
    decide_route,
    {
        "approve": "approve",
        "review": "review",
        "escalate": "escalate",
    },
)

builder.add_edge("approve", END)
builder.add_edge("review", END)
builder.add_edge("escalate", END)

graph = builder.compile()

result = graph.invoke({
    "claim": {
        "claim_id": "CLM-1001",
        "provider_id": 88321,
        "member_id": 44102,
        "amount": 7200,
        "duplicate_flag": True,
    },
    "findings": [],
    "risk_score": 0.0,
    "decision": "",
    "audit": [],
})

print(result["decision"])
print(result["audit"])

4) Add persistence and traceability

For healthcare use cases, keep an immutable audit record outside the graph. The graph should compute decisions; your application layer should persist them with request IDs, user identity, source system ID, and region.

def persist_audit(result: FraudState) -> None:
    record = {
        "claim_id": result["claim"]["claim_id"],
        "decision": result["decision"],
        "risk_score": result["risk_score"],
        "findings": result["findings"],
        # write this to your database / SIEM / case system
    }

Production Considerations

• •

  HIPAA and PHI handling

  • •Minimize PHI in graph state.
  • •Redact member names unless a downstream node absolutely needs them.
  • •Encrypt at rest and in transit.
• •

  Data residency

  • •Keep processing in-region if your payer or hospital contract requires it.
  • •If you use external model APIs for scoring or summarization, verify where data is stored and processed.
• •

  Auditability

  • •Log every node transition with trace IDs.
  • •Store inputs that influenced the decision: rule hits, scores, retrieved evidence references.
  • •Make sure investigators can reconstruct why a claim was escalated.
• •

  Human-in-the-loop controls

  • •Do not auto-deny high-value claims without review unless policy explicitly allows it.
  • •Route borderline cases to compliance analysts or SIU staff.

Common Pitfalls

• •

  Putting raw PHI into every node

  • •Avoid passing full patient records through the graph when a claim ID plus feature summary is enough.
  • •Use field-level minimization and fetch sensitive details only inside controlled nodes.
• •

  Relying on LLM judgment for final fraud decisions

  • •Use LLMs for summarization or evidence extraction.
  • •Keep final routing deterministic with rules and calibrated thresholds.
• •

  Skipping audit metadata

  • •A fraud decision without provenance is hard to defend.
  • •Record which rules fired, which evidence was used, who reviewed it later, and which model version produced the score.
• •

  Ignoring regional compliance constraints

  • •If your organization operates across states or countries, don’t assume one deployment fits all.
  • •Separate tenants or environments when residency or contractual obligations differ.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.