How to Build a transaction monitoring Agent Using LangGraph in Python for insurance

A transaction monitoring agent for insurance watches policy payments, premium collections, claims-related disbursements, refunds, and payout activity for patterns that need review. It matters because insurers deal with fraud, AML exposure, misapplied funds, and regulatory audit trails, and manual review does not scale once volume starts climbing.

Architecture

• •Event ingestion layer
  • •Receives payment events, claim payout events, refund events, and policy updates from Kafka, SQS, or a REST webhook.
• •Normalization node
  • •Converts raw insurer-specific payloads into a standard transaction schema with fields like customer_id, policy_id, amount, currency, channel, and jurisdiction.
• •Risk scoring node
  • •Applies deterministic rules first: amount thresholds, velocity checks, refund frequency, unusual beneficiary changes, and jurisdiction mismatches.
• •LLM analysis node
  • •Summarizes the case for an analyst and classifies the transaction into clear, review, or escalate.
• •Decision router
  • •Routes low-risk cases to auto-clear and high-risk cases to human review or case management.
• •Audit logger
  • •Persists every state transition, score, prompt output, and final decision for compliance and model governance.

Implementation

1) Define the state and build the graph

Use StateGraph from LangGraph to model the workflow. Keep the state explicit so every step is auditable and easy to replay.

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END

class TxState(TypedDict):
    transaction: dict
    normalized: dict
    risk_score: int
    decision: Literal["clear", "review", "escalate"]
    rationale: str

def normalize_tx(state: TxState) -> TxState:
    tx = state["transaction"]
    normalized = {
        "customer_id": tx["customer_id"],
        "policy_id": tx["policy_id"],
        "amount": float(tx["amount"]),
        "currency": tx.get("currency", "USD"),
        "channel": tx.get("channel", "unknown"),
        "jurisdiction": tx.get("jurisdiction", "unknown"),
        "type": tx["type"],
    }
    return {**state, "normalized": normalized}

def score_tx(state: TxState) -> TxState:
    n = state["normalized"]
    score = 0
    if n["amount"] >= 10000:
        score += 40
    if n["type"] in {"refund", "claim_payout"}:
        score += 20
    if n["jurisdiction"] not in {"US", "CA", "GB"}:
        score += 15
    return {**state, "risk_score": score}

def route_tx(state: TxState) -> str:
    if state["risk_score"] >= 50:
        return "escalate"
    if state["risk_score"] >= 25:
        return "review"
    return "clear"

graph = StateGraph(TxState)
graph.add_node("normalize_tx", normalize_tx)
graph.add_node("score_tx", score_tx)

graph.add_edge(START, "normalize_tx")
graph.add_edge("normalize_tx", "score_tx")
graph.add_conditional_edges(
    "score_tx",
    route_tx,
    {
        "clear": END,
        "review": END,
        "escalate": END,
    },
)

app = graph.compile()

2) Add an LLM review node for analyst-ready explanations

For insurance operations, the model should not be making opaque decisions. Use it to produce a concise rationale after deterministic scoring has already done the heavy lifting.

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def llm_review(state: TxState) -> TxState:
    n = state["normalized"]
    prompt = (
        f"Review this insurance transaction for monitoring.\n"
        f"Type: {n['type']}\n"
        f"Amount: {n['amount']} {n['currency']}\n"
        f"Channel: {n['channel']}\n"
        f"Jurisdiction: {n['jurisdiction']}\n"
        f"Risk score: {state['risk_score']}\n"
        f"Return a short rationale and one of clear/review/escalate."
    )
    result = llm.invoke(prompt)
    text = result.content.lower()

    decision = "review"
    if "escalate" in text:
      decision = "escalate"
    elif "clear" in text:
      decision = "clear"

    return {**state, "decision": decision, "rationale": result.content}

If you want the LLM in the graph path, insert it after scoring and before routing. In practice I keep it behind a threshold so only borderline cases hit the model.

3) Wire conditional routing for production behavior

This pattern keeps your fast path deterministic and sends only uncertain cases to language-model analysis.

def needs_llm(state: TxState) -> str:
    if state["risk_score"] >= 25:
        return "llm_review"
    return state.get("decision", "clear")

graph = StateGraph(TxState)
graph.add_node("normalize_tx", normalize_tx)
graph.add_node("score_tx", score_tx)
graph.add_node("llm_review", llm_review)

graph.add_edge(START, "normalize_tx")
graph.add_edge("normalize_tx", "score_tx")
graph.add_conditional_edges(
    "score_tx",
    needs_llm,
    {
        "llm_review": "llm_review",
        END: END,
        # direct clear path can terminate here if you set decision earlier
        # in your scoring logic
    },
)
graph.add_edge("llm_review", END)

app = graph.compile()

4) Run a sample transaction through the agent

sample = {
    "transaction": {
        "customer_id": "C123",
        "policy_id": "P9981",
        "amount": 12500,
        "currency": "USD",
        "channel": "bank_transfer",
        "jurisdiction": "NG",
        "type": "claim_payout",
    },
}

result = app.invoke(sample)
print(result)

That gives you a clean execution trace you can persist in your case system. For insurance teams, this is what makes the workflow defensible during audits.

Production Considerations

• •Auditability
  • •Persist each node output with timestamps, model version, prompt text, and final disposition.
  • •Regulators will ask why a claim payout or refund was flagged; you need replayable evidence.
• •Data residency
  • •Keep customer PII and claim data inside approved regions.
  • •If you call an external model API, redact policyholder identifiers before invocation or use a region-bound deployment.
• •Compliance guardrails
  • •Add deterministic rules for sanctions exposure, suspicious refunds, duplicate payouts, and rapid policy changes.
  • •Do not let the LLM override hard compliance blocks.
• •Monitoring
  • •Track false positives by line of business: motor, health, life, property.
  • •Alert on drift in risk-score distribution because seasonal claims spikes can break static thresholds.

Common Pitfalls

• •Letting the LLM decide everything
  • •Bad pattern. Use rules for first-pass screening and reserve the model for explanation or borderline classification.
• •Not normalizing transaction schemas
  • •If premium payments and claim payouts use different field names across systems, your graph becomes brittle fast.
  • •Normalize upfront into one canonical schema.
• •Ignoring jurisdiction-specific policy
  • •Insurance data handling differs across countries and states.
  • •Bake residency rules and retention policies into the workflow instead of handling them outside the agent.

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.