How to Build a compliance checking Agent Using LangGraph in Python for investment banking

A compliance checking agent in investment banking reviews proposed client communications, trade-related messages, research drafts, and internal requests against policy rules before anything leaves the firm. It matters because one missed restriction can trigger regulatory breaches, audit findings, or client harm, and the cost of manual review does not scale with deal flow.

Architecture

• •

  Input normalizer

  • •Takes raw text from email, chat, memo, or ticketing systems.
  • •Extracts metadata like desk, jurisdiction, client type, product class, and timestamp.
• •

  Policy retrieval layer

  • •Pulls the relevant controls from a versioned compliance knowledge base.
  • •Uses jurisdiction-aware rules for SEC/FINRA, FCA, MiFID II, MAR, and internal policy.
• •

  Risk analysis node

  • •Classifies the content into buckets like market abuse risk, unsuitable advice risk, disclosure risk, or recordkeeping risk.
  • •Produces structured findings with severity and rationale.
• •

  Decision engine

  • •Converts findings into an action: approve, escalate to human compliance, or block.
  • •Applies deterministic thresholds so the agent is auditable.
• •

  Audit logger

  • •Persists inputs, outputs, policy version, model version, and decision trace.
  • •Supports post-trade review and regulator-facing evidence.
• •

  Human escalation interface

  • •Routes ambiguous cases to compliance officers.
  • •Captures reviewer overrides for continuous tuning.

Implementation

1. •Define a typed state for the graph

Use TypedDict to keep the workflow explicit. In regulated environments, I prefer structured state over passing free-form blobs between nodes.

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

class ComplianceState(TypedDict):
    message: str
    jurisdiction: str
    desk: str
    policy_text: str
    findings: list[str]
    risk_level: Literal["low", "medium", "high"]
    decision: Literal["approve", "escalate", "block"]

2. •Build node functions for retrieval, analysis, and decisioning

Keep retrieval deterministic where possible. If you use an LLM for classification, force structured output and keep the final decision rule-based.

def retrieve_policy(state: ComplianceState) -> dict:
    jurisdiction = state["jurisdiction"].lower()
    if jurisdiction == "uk":
        policy = "UK policy: no misleading statements; recordkeeping required; escalate potential market abuse."
    else:
        policy = "US policy: no misleading statements; no MNPI handling; escalate suspicious trading commentary."
    return {"policy_text": policy}

def analyze_message(state: ComplianceState) -> dict:
    text = state["message"].lower()
    findings = []

    if any(term in text for term in ["guaranteed return", "risk-free", "insider"]):
        findings.append("Potential misleading or prohibited statement")
    if any(term in text for term in ["front-run", "tip off", "mnpi"]):
        findings.append("Possible market abuse / MNPI concern")

    risk_level = "low"
    if len(findings) == 1:
        risk_level = "medium"
    elif len(findings) > 1:
        risk_level = "high"

    return {"findings": findings, "risk_level": risk_level}

def decide(state: ComplianceState) -> dict:
    if state["risk_level"] == "high":
        return {"decision": "block"}
    if state["risk_level"] == "medium":
        return {"decision": "escalate"}
    return {"decision": "approve"}

3. •Wire the workflow with StateGraph

This is the core LangGraph pattern: define nodes with add_node, connect them with add_edge, then compile. For investment banking use cases, this gives you a stable execution trace that compliance teams can inspect later.

from langgraph.graph import StateGraph, START, END

graph = StateGraph(ComplianceState)

graph.add_node("retrieve_policy", retrieve_policy)
graph.add_node("analyze_message", analyze_message)
graph.add_node("decide", decide)

graph.add_edge(START, "retrieve_policy")
graph.add_edge("retrieve_policy", "analyze_message")
graph.add_edge("analyze_message", "decide")
graph.add_edge("decide", END)

compliance_app = graph.compile()

result = compliance_app.invoke({
    "message": "Can we tell the client this trade has a guaranteed return?",
    "jurisdiction": "US",
    "desk": "Equities",
    "policy_text": "",
    "findings": [],
    "risk_level": "low",
    "decision": "approve",
})

print(result)

4. •Add escalation logic with conditional edges

Real compliance workflows are not linear. Use add_conditional_edges when high-risk cases need human review instead of an automatic block.

def route_after_analysis(state: ComplianceState) -> str:
    if state["risk_level"] == "high":
        return "block"
    if state["risk_level"] == "medium":
        return "escalate"
    return "approve"

def escalate_to_human(state: ComplianceState) -> dict:
    return {"decision": "escalate"}

workflow = StateGraph(ComplianceState)
workflow.add_node("retrieve_policy", retrieve_policy)
workflow.add_node("analyze_message", analyze_message)
workflow.add_node("escalate_to_human", escalate_to_human)
workflow.add_node("decide", decide)

workflow.add_edge(START, "retrieve_policy")
workflow.add_edge("retrieve_policy", "analyze_message")
workflow.add_conditional_edges(
    "analyze_message",
    route_after_analysis,
    {
        "approve": END,
        "escalate": "escalate_to_human",
        "block": END,
    },
)
workflow.add_edge("escalate_to_human", END)

app = workflow.compile()

Production Considerations

• •

  Deploy in-region

  • •Keep message content and audit logs inside approved data residency boundaries.
  • •For cross-border desks, separate EU/UK/US workloads so policies and storage stay jurisdiction-specific.
• •

  Log every decision artifact

  • •Persist input text hash, extracted metadata, policy version ID, graph run ID (thread_id if you use checkpointing), and final outcome.
  • •Regulators care about why a decision was made as much as the decision itself.
• •

  Use hard guardrails

  • •Do not let the model override deterministic blocks for prohibited phrases or restricted topics.
  • •Treat market abuse indicators and MNPI references as automatic escalation triggers.
• •

  Monitor false negatives aggressively

  • •Track precision/recall by desk and jurisdiction.
  • •A low false-positive rate is not enough if you are missing sensitive communications in M&A or sales-and-trading flows.

Common Pitfalls

1. •

   Using a free-form LLM response as the final decision

   • •That is not acceptable for regulated workflows.
   • •Use the model for classification or extraction only; keep approval/block/escalation deterministic.
2. •

   Ignoring jurisdiction-specific policy differences

   • •A rule set that works for US equities may fail for UK advisory or EU research distribution.
   • •Parameterize policies by region and business line from day one.
3. •

   Skipping auditability

   • •If you cannot reproduce why a message was blocked six months later, the system is incomplete.
   • •Store graph inputs/outputs plus policy snapshots and model versions for every run.

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.