How to Build a underwriting Agent Using LangGraph in Python for payments
An underwriting agent for payments takes a transaction, merchant profile, and risk signals, then decides whether to approve, review, or reject the payment. In practice, it matters because bad underwriting means fraud losses, compliance breaches, and false declines that hit conversion.
Architecture
- •
Input normalizer
- •Converts raw payment payloads into a stable schema.
- •Pulls out merchant country, MCC, transaction amount, card-present flags, device signals, and prior chargeback history.
- •
Policy engine
- •Encodes hard rules for sanctions, KYC/KYB status, velocity limits, and prohibited industries.
- •This should be deterministic and auditable.
- •
Risk scoring node
- •Uses a model or heuristic layer to estimate fraud and credit exposure.
- •Returns a score plus reason codes, not just a number.
- •
Decision node
- •Maps risk + policy outcomes into
approve,manual_review, ordecline. - •Keeps thresholds explicit and versioned.
- •Maps risk + policy outcomes into
- •
Audit logger
- •Persists every decision input/output with timestamps and model/policy versions.
- •Required for disputes, compliance reviews, and internal controls.
- •
Human review handoff
- •Routes borderline cases to an operations queue.
- •Needed when the agent lacks confidence or hits regulated edge cases.
Implementation
1) Define the state and decision schema
Use typed state so your graph stays predictable. For payments, keep the fields narrow and explicit; do not pass raw blobs through every node.
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
Decision = Literal["approve", "manual_review", "decline"]
class UnderwritingState(TypedDict):
merchant_id: str
merchant_country: str
amount: float
mcc: str
kyc_passed: bool
sanctions_hit: bool
chargeback_rate: float
risk_score: float
decision: Decision
reason: str
2) Build deterministic policy and scoring nodes
Keep policy checks separate from scoring. That gives you clean audit trails and makes it easy to explain why a payment was blocked.
def policy_check(state: UnderwritingState) -> dict:
if state["sanctions_hit"]:
return {"decision": "decline", "reason": "Sanctions hit"}
if not state["kyc_passed"]:
return {"decision": "manual_review", "reason": "KYC incomplete"}
if state["mcc"] in {"7995", "4829"}:
return {"decision": "manual_review", "reason": "High-risk MCC"}
return {}
def score_risk(state: UnderwritingState) -> dict:
score = 0.0
score += min(state["amount"] / 1000.0, 3.0)
score += state["chargeback_rate"] * 10.0
if state["merchant_country"] not in {"US", "GB", "CA", "DE"}:
score += 1.5
return {"risk_score": score}
3) Add a routing function with LangGraph conditional edges
This is the core pattern. The graph first applies policy checks, then scores risk if needed, then routes based on thresholds.
def route_decision(state: UnderwritingState) -> Decision:
if state.get("decision") == "decline":
return "decline"
if state.get("decision") == "manual_review":
return "manual_review"
score = state.get("risk_score", 0.0)
if score >= 4.0:
return "decline"
if score >= 2.0:
return "manual_review"
return "approve"
def finalize_approval(state: UnderwritingState) -> dict:
return {"decision": "approve", "reason": f"Risk score {state['risk_score']:.2f}"}
def finalize_manual_review(state: UnderwritingState) -> dict:
return {"decision": "manual_review", "reason": f"Escalated with score {state['risk_score']:.2f}"}
def finalize_decline(state: UnderwritingState) -> dict:
return {"decision": "decline", "reason": f"Declined with score {state.get('risk_score', 0.0):.2f}"}
Now wire it together with StateGraph, add_node, add_edge, add_conditional_edges, and compile.
workflow = StateGraph(UnderwritingState)
workflow.add_node("policy_check", policy_check)
workflow.add_node("score_risk", score_risk)
workflow.add_node("approve", finalize_approval)
workflow.add_node("manual_review", finalize_manual_review)
workflow.add_node("decline", finalize_decline)
workflow.add_edge(START, "policy_check")
def after_policy(state: UnderwritingState):
# If policy_check already set a terminal decision, route directly.
if state.get("decision") in {"decline", "manual_review"}:
return state["decision"]
return "score_risk"
workflow.add_conditional_edges(
"policy_check",
after_policy,
{
"score_risk": "score_risk",
"approve": "approve",
"manual_review": "manual_review",
"decline": "decline",
},
)
workflow.add_conditional_edges(
"score_risk",
route_decision,
{
"approve": "approve",
"manual_review": "manual_review",
"decline": "decline",
},
)
workflow.add_edge("approve", END)
workflow.add_edge("manual_review", END)
workflow.add_edge("decline", END)
app = workflow.compile()
result = app.invoke(
{
"merchant_id": "m_123",
"merchant_country": "US",
"amount": 250.00,
"mcc": "5411",
"kyc_passed": True,
"sanctions_hit": False,
"chargeback_rate": 0.01,
# these are filled by nodes if needed
# but keep initial state complete in production pipelines
# for clarity in audits.
# default placeholders:
# risk_score=0.0,
# decision="approve",
# reason=""
}
)
print(result)
4) Add audit logging around the graph execution
For payments, every decision needs traceability. Log the input snapshot, output decision, threshold version, and any external signals used during underwriting.
- •Store logs in an immutable audit table or append-only event stream.
- •Include
merchant_id, request idempotency key, graph version hash, and policy version. - •Keep PII minimized; tokenize or redact before persistence.
- •If you process regulated data across regions, pin execution to the correct region to satisfy data residency requirements.
Production Considerations
- •
Deploy stateless workers
- •Keep the graph code stateless and push state into your request payload or external store.
- •Use idempotency keys so retries do not duplicate decisions or side effects.
- •
Monitor decision drift
- •Track approval rate, manual review rate, false decline rate, chargeback rate by cohort.
- •Alert when a merchant segment shifts sharply after a policy or model update.
- •
Enforce guardrails
- •Hard-block sanctions hits and prohibited MCCs before any LLM-driven step.
- •Never let the model override deterministic compliance rules.
- •
Control data residency
- •Route EU merchants through EU-hosted infrastructure and keep logs in-region.
- •Avoid sending raw PANs or sensitive identity data into prompts or third-party services.
Common Pitfalls
- •
Mixing policy logic with model reasoning
- •If one node both scores risk and decides compliance outcomes, audits become messy.
- •Split hard rules from probabilistic scoring so you can explain every decline.
- •
Using free-form text instead of structured state
- •Payments workflows need exact fields like country codes, MCCs, thresholds, and reason codes.
- •Use typed dictionaries or Pydantic models so downstream systems can validate outputs.
- •
Skipping human review paths
- •Borderline cases will happen: thin-file merchants, cross-border volume spikes, unusual MCCs.
- •Route uncertain cases to manual review instead of forcing an automatic decline.
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.
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