How to Build a KYC verification Agent Using LangGraph in Python for payments

A KYC verification agent for payments takes customer identity data, checks it against internal policy and external signals, and decides whether the account can move forward, needs manual review, or must be rejected. That matters because payment flows are where fraud, AML exposure, chargeback risk, and onboarding friction collide; if your KYC logic is weak, you either lose customers or let bad actors through.

Architecture

A production KYC agent for payments usually needs these components:

• •

  Input normalization

  • •Parse raw onboarding payloads into a consistent schema.
  • •Validate required fields like name, DOB, address, country, and ID document metadata.
• •

  Policy engine

  • •Encode jurisdiction-specific rules.
  • •Apply thresholds for sanctions hits, PEP flags, document mismatch, and residency constraints.
• •

  Verification tools

  • •Call external services for ID validation, sanctions screening, address verification, and liveness checks.
  • •Keep tool outputs structured so the graph can make deterministic decisions.
• •

  Decision state

  • •Track evidence, risk score, decision status, and audit trail across nodes.
  • •Persist enough context for compliance review without storing unnecessary PII.
• •

  Human review branch

  • •Route ambiguous cases to an analyst queue.
  • •Capture reviewer decisions and reasons for auditability.
• •

  Audit logging

  • •Store every state transition and tool call result.
  • •This is non-negotiable in payments because regulators will ask why a customer was approved or blocked.

Implementation

1) Define the graph state and decision model

Use a typed state so every node in the graph works against the same contract. For payments KYC, keep the state small and explicit: input data, evidence collected, risk score, final decision, and an audit trail.

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

Decision = Literal["approve", "manual_review", "reject"]

class KYCState(TypedDict):
    applicant_id: str
    full_name: str
    country: str
    dob: str
    id_number: str
    sanctions_hit: bool
    pep_hit: bool
    doc_verified: bool
    risk_score: int
    decision: Decision
    audit_log: list[str]

2) Add deterministic nodes for screening and scoring

Do not put business policy inside a prompt. Use plain Python functions for screening logic so the outcome is explainable and testable.

def initialize_state(state: KYCState) -> KYCState:
    state["audit_log"] = ["received_application"]
    state["risk_score"] = 0
    return state

def screen_sanctions_and_pep(state: KYCState) -> KYCState:
    # Replace with real provider calls in production.
    state["sanctions_hit"] = state["full_name"].lower() in {"john doe"}
    state["pep_hit"] = state["country"] in {"IR", "KP"}
    
    if state["sanctions_hit"]:
        state["risk_score"] += 80
        state["audit_log"].append("sanctions_hit")
    if state["pep_hit"]:
        state["risk_score"] += 40
        state["audit_log"].append("pep_or_high_risk_country")
        
    return state

def verify_documents(state: KYCState) -> KYCState:
    # Replace with OCR / IDV vendor integration.
    state["doc_verified"] = len(state["id_number"]) >= 8
    
    if not state["doc_verified"]:
        state["risk_score"] += 50
        state["audit_log"].append("document_verification_failed")
    else:
        state["audit_log"].append("document_verified")
        
    return state

def decide(state: KYCState) -> KYCState:
    if state["sanctions_hit"]:
        state["decision"] = "reject"
    elif state["risk_score"] >= 60:
        state["decision"] = "manual_review"
    else:
        state["decision"] = "approve"
    
    state["audit_log"].append(f"decision:{state['decision']}")
    return state

3) Wire the workflow with LangGraph

This is the actual LangGraph pattern: create a StateGraph, add nodes with add_node, connect them with add_edge, set conditional routing with add_conditional_edges, then compile it. The router keeps approval logic explicit and easy to test.

def route_after_screening(state: KYCState) -> str:
    if state["sanctions_hit"]:
        return "reject"
    if not state["doc_verified"]:
        return "manual_review"
    if state["risk_score"] >= 60:
        return "manual_review"
    return "approve"

graph = StateGraph(KYCState)

graph.add_node("initialize_state", initialize_state)
graph.add_node("screen_sanctions_and_pep", screen_sanctions_and_pep)
graph.add_node("verify_documents", verify_documents)
graph.add_node("decide", decide)

graph.add_edge(START, "initialize_state")
graph.add_edge("initialize_state", "screen_sanctions_and_pep")
graph.add_edge("screen_sanctions_and_pep", "verify_documents")

graph.add_conditional_edges(
    "verify_documents",
    route_after_screening,
    {
        "approve": "decide",
        "manual_review": "decide",
        "reject": "decide",
    },
)

graph.add_edge("decide", END)

app = graph.compile()

result = app.invoke({
    "applicant_id": "app_123",
    "full_name": "Jane Smith",
    "country": "GB",
    "dob": "1990-04-12",
    "id_number": "A12345678",
})
print(result)

4) Add human review as a separate branch when needed

For payments ops teams, manual review is part of the control plane. In LangGraph you can route suspicious cases to a reviewer node instead of forcing a binary answer from automation.

def manual_review(state: KYCState) -> KYCState:
    # In production this would write to a queue or case management system.
    state["audit_log"].append("sent_to_manual_review")
    
def route_to_review(state: KYCState) -> str:
   return "review" if state["decision"] == "manual_review" else END

# If you want a separate branch in your graph:
# graph.add_node("manual_review", manual_review)
# graph.add_conditional_edges("decide", route_to_review, {"review": "manual_review", END: END})

Production Considerations

• •

  Persist audit trails

  • •Store node inputs/outputs and final decisions in immutable logs.
  • •Payments compliance teams need traceability for every approval and rejection.
• •

  Control data residency

  • •Keep PII processing inside approved regions.
  • •If you call third-party verification APIs cross-border, confirm legal basis and contractual controls first.
• •

  Add observability at node level

  • •Measure latency per node, external vendor error rates, false positives on sanctions screening, and manual review volume.
  • •A spike in review rate often means your thresholds are too aggressive or a vendor feed changed format.
• •

  Put guardrails around tool access

  • •Only allow tools that are necessary for verification.
  • •Do not let an LLM freely decide which external systems to query; wrap each provider behind deterministic functions with strict schemas.

Common Pitfalls

1. •

   Using free-form LLM output for compliance decisions

   • •Mistake: asking an LLM to “decide” whether someone passes KYC.
   • •Fix: keep policy in Python logic and use LLMs only for extraction or summarization where needed.
2. •

   Ignoring jurisdiction-specific rules

   • •Mistake: applying one global threshold to every customer.
   • •Fix: parameterize policy by country or region so sanctions handling, document requirements, and retention rules match local regulation.
3. •

   Not preserving enough evidence for audits

   • •Mistake: storing only the final decision.
   • •Fix: persist the reason codes, vendor responses, timestamps, and graph path taken so compliance can reconstruct the case later.

If you build this as a deterministic LangGraph workflow first, you get something that is testable under regulation pressure. Then you can add AI where it helps most: extraction from messy documents, summarizing reviewer notes, or classifying edge cases that still end up in human hands.

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.