Part 1: The Division of Labor — LLMs vs. Classical Models
What Classical Models Handle (The Numerical Backbone)
Classical ML models remain the regulatory-compliant, deterministic core of credit scoring. They are preferred here because regulators (OCC, CFPB, FCA) require explainability, reproducibility, and statistical fairness — properties that tabular models provide natively.
| Component | Classical Model | Why Not LLM? |
|---|
| Probability of Default (PD) | Logistic Regression, XGBoost, LightGBM | Requires calibrated probabilities on structured tabular data; LLMs hallucinate numbers |
| FICO-like Scorecard | Weighted logistic regression (WOE/IV binning) | Regulatory requirement for monotonic, auditable scorecards |
| Debt-to-Income (DTI) Analysis | Rule engine + decision trees | Pure arithmetic; LLMs introduce unnecessary latency and error |
| Payment History Modeling | Time-series (ARIMA, survival analysis) | Sequential numerical patterns; transformers overkill |
| Credit Utilization Scoring | Gradient boosting on bureau features | Tabular data with <200 features; classical models dominate |
| Loss Given Default (LGD) | Tobit regression, Beta regression | Continuous bounded outcome; LLMs can't calibrate |
| Application Fraud Scoring | Isolation Forest, XGBoost on velocity features | Structured behavioral signals need statistical rigor |
| Portfolio Risk (Basel III/IV) | Monte Carlo, Vasicek model | Actuarial math; not a language problem |
What LLMs Handle (The Unstructured Intelligence Layer)
LLMs add value where data is unstructured, contextual, or requires reasoning over text:
| Component | LLM Role | Why Classical Fails |
|---|
| Document Parsing | Extracting income from tax returns, bank statements, pay stubs | OCR + regex is brittle; LLMs handle varied formats |
| Applicant Narrative Analysis | Evaluating explanations for credit gaps, disputes, bankruptcies | No structured feature exists for "my medical emergency caused 90-day late" |
| Regulatory Compliance (RAG) | Cross-referencing decisions against FCRA, ECOA, Basel guidelines | Rules change; hard-coding is unmaintainable |
| Adverse Action Notices | Generating compliant, personalized denial letters | Template engines lack contextual nuance |
| Fraud Narrative Detection | Spotting inconsistencies in free-text application fields | Pattern matching misses semantic contradictions |
| Customer Clarification Dialog | Asking follow-up questions about ambiguous income sources | Rule-based chatbots fail on edge cases |
| Policy Interpretation | Determining if a novel scenario fits internal credit policy | Requires reasoning, not classification |
The Hybrid Reality
Classical models produce the score. LLMs produce the context. The final credit decision is a fusion of both, orchestrated by a multi-agent system that ensures compliance, explainability, and accuracy.
Part 2: Enterprise Multi-Agent Architecture
Real-World Use Case: Real-Time SME Loan Underwriting at a Digital Bank
Scenario: A small business applies for a $500K working capital loan. The application includes:
Structured data: financials, credit bureau pulls, DTI, trade lines
Unstructured data: 2 years of tax returns (PDF), bank statements, a written explanation of a prior bankruptcy, and a business plan
The system must:
Score the application numerically (classical)
Parse and validate documents (LLM)
Check compliance against current regulations (RAG)
Detect fraud signals (hybrid)
Render a final decision with full explainability (LLM)
Remember the applicant across sessions (Memory)
Architecture Diagram
![ab]()
Part 3: Full Code Implementation
Prerequisites
pip install langgraph langchain langchain-openai langchain-community \
chromadb xgboost scikit-learn pandas numpy \
langchain-chroma pydantic redis
Step 1: Define the Shared State
LangGraph's TypedDict state is the single source of truth that flows through every agent.
# state.py
from typing import TypedDict, Annotated, Optional, List
from langgraph.graph.message import add_messages
from pydantic import BaseModel
import operator
class DocumentExtraction(BaseModel):
"""Structured output from LLM document parsing."""
annual_revenue: Optional[float] = None
net_income: Optional[float] = None
years_in_business: Optional[int] = None
bankruptcy_explanation: Optional[str] = None
business_purpose: Optional[str] = None
document_confidence: float = 0.0
class ClassicalScores(BaseModel):
"""Output from classical ML models."""
pd_score: float = 0.0 # Probability of Default (0-1)
credit_score: int = 0 # FICO-like score (300-850)
dti_ratio: float = 0.0 # Debt-to-Income
ltv_ratio: float = 0.0 # Loan-to-Value
scorecard_points: int = 0 # Traditional scorecard
model_version: str = "xgb_v3.2"
class ComplianceResult(BaseModel):
"""RAG-based compliance check."""
is_compliant: bool = True
violated_regulations: List[str] = []
applicable_guidelines: List[str] = []
risk_tier: str = "standard" # standard, elevated, prohibited
class FraudSignals(BaseModel):
"""Hybrid fraud detection output."""
fraud_probability: float = 0.0
behavioral_flags: List[str] = []
narrative_inconsistencies: List[str] = []
velocity_alerts: List[str] = []
class CreditDecision(BaseModel):
"""Final orchestrated decision."""
decision: str = "pending" # approve, decline, refer, conditional
approved_amount: float = 0.0
interest_rate: float = 0.0
risk_grade: str = ""
conditions: List[str] = []
confidence: float = 0.0
class UnderwritingState(TypedDict):
"""
The complete state that flows through the LangGraph pipeline.
Every agent reads from and writes to this state.
"""
# --- Identifiers ---
application_id: str
applicant_name: str
session_id: str
# --- Raw Inputs ---
structured_data: dict # Bureau pull, financials
unstructured_documents: List[str] # Raw text from PDFs
applicant_narrative: str # Free-text explanation
# --- Agent Outputs ---
messages: Annotated[list, add_messages] # LangGraph message history
document_extraction: Optional[DocumentExtraction]
classical_scores: Optional[ClassicalScores]
compliance_result: Optional[ComplianceResult]
fraud_signals: Optional[FraudSignals]
final_decision: Optional[CreditDecision]
# --- RAG Context ---
retrieved_regulations: List[str] # Chunks from vector store
internal_policy_docs: List[str]
# --- Memory ---
previous_applications: List[dict] # Historical context
conversation_history: Annotated[list, add_messages]
# --- Control Flow ---
current_step: str
requires_human_review: bool
error_log: List[str]
Step 2: Classical ML Model (The Numerical Engine)
This is a production-grade XGBoost pipeline that would be trained offline and served via API. Here we simulate the trained model.
# classical_models.py
import numpy as np
import xgboost as xgb
from sklearn.linear_model import LogisticRegression
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
import joblib
import os
class CreditScoringEngine:
"""
Classical ML ensemble for credit scoring.
In production: loaded from MLflow/S3 model registry.
Handles: PD, Scorecard, DTI analysis, LGD estimation.
"""
def __init__(self):
self.pd_model = self._build_pd_model()
self.scorecard = self._build_scorecard()
self.dti_thresholds = {
"excellent": 0.28,
"good": 0.36,
"fair": 0.43,
"poor": 0.50
}
def _build_pd_model(self) -> Pipeline:
"""
Probability of Default model.
Architecture: XGBoost with calibrated probabilities.
Features: payment_history, utilization, inquiries, age_of_credit,
derogatory_marks, total_debt, income, employment_length
"""
# In production: xgb.XGBClassifier loaded from model registry
model = Pipeline([
("scaler", StandardScaler()),
("xgb", xgb.XGBClassifier(
n_estimators=200,
max_depth=5,
learning_rate=0.05,
subsample=0.8,
colsample_bytree=0.8,
reg_alpha=0.1,
reg_lambda=1.0,
scale_pos_weight=5, # Imbalanced default rates
random_state=42,
eval_metric="auc"
))
])
return model
def _build_scorecard(self) -> dict:
"""
Traditional scorecard using Weight of Evidence (WOE) binning.
This is what regulators expect: monotonic, interpretable points.
"""
return {
"payment_history": {
"0_lates": 100, "1_late_30": 60, "1_late_60": 30,
"1_late_90": -20, "2_plus_lates": -60, "collection": -100
},
"utilization": {
"0_10": 80, "10_30": 50, "30_50": 20,
"50_75": -10, "75_90": -40, "90_100": -80
},
"credit_age": {
"10_plus": 60, "5_10": 40, "2_5": 20,
"1_2": 0, "less_1": -20
},
"inquiries_6m": {
"0": 30, "1_2": 15, "3_4": -5, "5_plus": -25
},
"derogatory": {
"none": 50, "minor": 0, "major": -50, "severe": -100
}
}
def calculate_pd(self, features: dict) -> float:
"""
Calculate Probability of Default.
This is the core classical model — NOT an LLM task.
"""
feature_vector = np.array([[
features.get("num_late_payments", 0),
features.get("credit_utilization", 0.5),
features.get("num_inquiries_6m", 0),
features.get("credit_age_years", 5),
features.get("num_derogatory", 0),
features.get("total_debt", 50000),
features.get("annual_income", 80000),
features.get("employment_years", 3),
features.get("num_open_accounts", 8),
features.get("debt_to_income", 0.35),
]])
# In production: self.pd_model.predict_proba(feature_vector)[0][1]
# Simulated calibrated output:
pd_score = 0.12 # 12% probability of default
return round(pd_score, 4)
def calculate_scorecard(self, features: dict) -> int:
"""
Traditional scorecard points.
Deterministic, auditable, regulator-friendly.
"""
total_points = 300 # Base score
# Payment history
lates = features.get("num_late_payments", 0)
if lates == 0:
total_points += self.scorecard["payment_history"]["0_lates"]
elif lates <= 2:
total_points += self.scorecard["payment_history"]["1_late_30"]
else:
total_points += self.scorecard["payment_history"]["2_plus_lates"]
# Utilization
util = features.get("credit_utilization", 0.5)
if util < 0.1:
total_points += self.scorecard["utilization"]["0_10"]
elif util < 0.3:
total_points += self.scorecard["utilization"]["10_30"]
elif util < 0.5:
total_points += self.scorecard["utilization"]["30_50"]
else:
total_points += self.scorecard["utilization"]["50_75"]
# Credit age
age = features.get("credit_age_years", 5)
if age >= 10:
total_points += self.scorecard["credit_age"]["10_plus"]
elif age >= 5:
total_points += self.scorecard["credit_age"]["5_10"]
else:
total_points += self.scorecard["credit_age"]["2_5"]
return min(850, max(300, total_points))
def calculate_dti(self, monthly_debt: float, monthly_income: float) -> float:
"""Pure arithmetic. No ML needed."""
if monthly_income == 0:
return 1.0
return round(monthly_debt / monthly_income, 4)
def full_classical_assessment(self, structured_data: dict) -> dict:
"""
Run all classical models.
Returns the numerical backbone of the credit decision.
"""
pd = self.calculate_pd(structured_data)
score = self.calculate_scorecard(structured_data)
dti = self.calculate_dti(
structured_data.get("monthly_debt_obligations", 3000),
structured_data.get("monthly_income", 10000)
)
ltv = structured_data.get("loan_amount", 500000) / \
max(structured_data.get("collateral_value", 600000), 1)
return {
"pd_score": pd,
"credit_score": score,
"dti_ratio": dti,
"ltv_ratio": round(ltv, 4),
"scorecard_points": score,
"model_version": "xgb_v3.2_prod_2026Q2"
}
Step 3: RAG Compliance Engine (Vector Store + Retrieval)
# rag_compliance.py
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document
from typing import List, Tuple
class ComplianceRAGEngine:
"""
RAG system for regulatory compliance.
Retrieves relevant regulations (FCRA, ECOA, Basel III, UDAAP)
and internal credit policies to validate underwriting decisions.
WHY RAG HERE: Regulations change frequently. Hard-coding rules
creates compliance risk. RAG allows real-time policy retrieval
with source attribution for audit trails.
"""
def __init__(self, persist_directory: str = "./chroma_compliance"):
self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
self.vectorstore = Chroma(
persist_directory=persist_directory,
embedding_function=self.embeddings
)
self._seed_knowledge_base()
def _seed_knowledge_base(self):
"""
In production: ingest from Confluence, SharePoint, regulatory feeds.
Here we seed with representative regulatory text.
"""
regulations = [
Document(
page_content="""FCRA Section 615(a): Adverse Action Notice Requirements.
When a creditor takes adverse action based on information in a consumer
report, the creditor must provide oral, written, or electronic notice
that includes: (1) the name, address, and phone number of the CRA that
furnished the report; (2) a statement that the CRA did not make the
decision and cannot explain specific reasons; (3) notice of the
consumer's right to obtain a free disclosure of their file within 60
days; and (4) notice of the right to dispute inaccurate information.
For business credit over $500K, FCRA requirements are relaxed but
ECOA still applies.""",
metadata={"regulation": "FCRA", "section": "615(a)",
"applicability": "consumer_credit", "severity": "high"}
),
Document(
page_content="""ECOA Regulation B: Equal Credit Opportunity Act.
Creditors shall not discriminate against any applicant on the basis
of race, color, religion, national origin, sex, marital status, age,
or receipt of public assistance. For business credit: (1) creditors
must notify applicants of action taken within 30 days for complete
applications, 90 days for incomplete; (2) adverse action notices must
state specific reasons or disclose the right to request reasons;
(3) for applications over $1M with gross revenue >$1M, recordkeeping
requirements are reduced. DTI thresholds must be applied consistently
across all protected classes.""",
metadata={"regulation": "ECOA", "section": "Regulation B",
"applicability": "all_credit", "severity": "critical"}
),
Document(
page_content="""Basel III/IV Credit Risk Framework: For SME lending,
banks must calculate Risk-Weighted Assets (RWA) using either the
Standardized Approach or Internal Ratings-Based (IRB) approach.
Under IRB: PD must be estimated using a minimum 5-year observation
period, LGD must reflect downturn conditions, and EAD must include
committed but undrawn facilities. Risk grades must map to a minimum
7-grade borrower scale. PD floors: 0.05% for corporate, 0.03% for
sovereign. High-volatility commercial real estate gets 150% risk
weight. Working capital facilities under 1 year get 20% CCF.""",
metadata={"regulation": "Basel III", "section": "Credit Risk",
"applicability": "bank_capital", "severity": "high"}
),
Document(
page_content="""Internal Credit Policy CP-2026-04: SME Working Capital.
Maximum exposure per borrower: $2M. Minimum credit score: 620 for
unsecured, 580 for secured. Maximum DTI: 45% for standard tier,
50% for preferred tier with 3+ years banking relationship.
Bankruptcy: automatic decline if discharged within 24 months;
manual review if 24-48 months; standard processing if >48 months.
Revenue requirement: minimum $250K annual revenue for 2 consecutive
years. Time in business: minimum 2 years. Industry exclusions:
cannabis, gambling, adult entertainment, cryptocurrency mining.""",
metadata={"regulation": "Internal Policy", "section": "CP-2026-04",
"applicability": "sme_lending", "severity": "binding"}
),
Document(
page_content="""UDAAP (Unfair, Deceptive, or Abusive Acts or Practices):
Underwriting models must not produce disparate impact on protected
classes. AI/ML models used in credit decisions must pass fair lending
testing: (1) Adverse Impact Ratio (AIR) must exceed 0.80 for all
protected classes; (2) Marginal effects analysis must show no
statistically significant disparate treatment; (3) proxy variables
for protected characteristics must be tested. Model validation must
occur annually per SR 11-7. LLM-generated explanations must be
factually consistent with the actual model output.""",
metadata={"regulation": "UDAAP", "section": "Fair Lending",
"applicability": "all_models", "severity": "critical"}
),
]
splitter = RecursiveCharacterTextSplitter(
chunk_size=500, chunk_overlap=50
)
chunks = splitter.split_documents(regulations)
self.vectorstore.add_documents(chunks)
def retrieve_relevant_regulations(
self, query: str, k: int = 4
) -> List[Tuple[str, dict]]:
"""
Retrieve top-k relevant regulatory chunks.
The query is constructed from the application context.
"""
results = self.vectorstore.similarity_search_with_score(query, k=k)
return [
(doc.page_content, doc.metadata)
for doc, score in results
]
def check_compliance(
self,
application_context: str,
classical_scores: dict,
decision: str
) -> dict:
"""
LLM-powered compliance check using RAG-retrieved regulations.
THIS IS WHERE LLM + RAG SHINES: The LLM reasons over retrieved
regulations to determine if the decision is compliant. A classical
model cannot do this — it requires natural language understanding
of legal text.
"""
regulations = self.retrieve_relevant_regulations(application_context)
reg_text = "\n\n".join([r[0] for r in regulations])
reg_sources = [r[1] for r in regulations]
prompt = f"""You are a senior compliance officer at a regulated bank.
RETRIEVED REGULATIONS:
{reg_text}
APPLICATION CONTEXT:
{application_context}
CLASSICAL MODEL SCORES:
{classical_scores}
PROPOSED DECISION: {decision}
Analyze whether this decision complies with all retrieved regulations.
Check specifically for:
1. FCRA adverse action requirements (if declining)
2. ECOA non-discrimination and notification requirements
3. Basel III capital adequacy for the risk grade
4. Internal policy thresholds (CP-2026-04)
5. UDAAP fair lending considerations
Return JSON with: is_compliant (bool), violated_regulations (list),
applicable_guidelines (list), risk_tier (standard/elevated/prohibited)."""
response = self.llm.invoke(prompt)
return {
"regulations_retrieved": reg_text,
"sources": reg_sources,
"compliance_analysis": response.content
}
Step 4: Define the LangGraph Agents
# agents.py
import json
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from classical_models import CreditScoringEngine
from rag_compliance import ComplianceRAGEngine
from state import (
UnderwritingState, DocumentExtraction, ClassicalScores,
ComplianceResult, FraudSignals, CreditDecision
)
llm = ChatOpenAI(model="gpt-4o", temperature=0)
scoring_engine = CreditScoringEngine()
rag_engine = ComplianceRAGEngine()
# ──────────────────────────────────────────────
# AGENT 1: INTAKE & DOCUMENT PARSING (LLM)
# ──────────────────────────────────────────────
def intake_agent(state: UnderwritingState) -> dict:
"""
LLM Agent: Parses unstructured documents and applicant narratives.
WHY LLM: Tax returns, bank statements, and bankruptcy explanations
are unstructured text with highly variable formats. Classical NLP
(regex, NER) fails on edge cases. LLMs handle this robustly.
"""
documents = state.get("unstructured_documents", [])
narrative = state.get("applicant_narrative", "")
prompt = f"""You are a loan document analyst. Extract structured financial
information from these unstructured documents and the applicant's narrative.
DOCUMENTS:
{chr(10).join(documents)}
APPLICANT NARRATIVE:
{narrative}
Extract and return JSON with:
- annual_revenue (float)
- net_income (float)
- years_in_business (int)
- bankruptcy_explanation (string, if any)
- business_purpose (string)
- document_confidence (float 0-1, how confident you are in extraction)"""
response = llm.invoke([
SystemMessage(content="You are a precise financial document parser. "
"Return only valid JSON."),
HumanMessage(content=prompt)
])
try:
extraction = json.loads(response.content)
except json.JSONDecodeError:
extraction = {
"annual_revenue": 0, "net_income": 0,
"years_in_business": 0, "document_confidence": 0.1
}
return {
"document_extraction": DocumentExtraction(**extraction),
"messages": [AIMessage(
content=f"Intake complete. Extracted revenue: "
f"${extraction.get('annual_revenue', 0):,.0f}"
)],
"current_step": "risk_scoring"
}
# ──────────────────────────────────────────────
# AGENT 2: CLASSICAL RISK SCORING (XGBoost)
# ──────────────────────────────────────────────
def risk_scoring_agent(state: UnderwritingState) -> dict:
"""
Classical ML Agent: Runs XGBoost PD model, scorecard, DTI, LTV.
WHY CLASSICAL: These are calibrated statistical models trained on
millions of loan outcomes. They produce reproducible, auditable
scores that regulators require. LLMs cannot produce calibrated
probabilities and would hallucinate numerical scores.
"""
structured = state.get("structured_data", {})
extraction = state.get("document_extraction")
# Merge LLM-extracted data into structured features
if extraction:
structured["annual_revenue_llm"] = extraction.annual_revenue
structured["net_income_llm"] = extraction.net_income
# Run classical models
scores = scoring_engine.full_classical_assessment(structured)
return {
"classical_scores": ClassicalScores(**scores),
"messages": [AIMessage(
content=f"Risk scoring complete. PD: {scores['pd_score']:.2%}, "
f"Score: {scores['credit_score']}, "
f"DTI: {scores['dti_ratio']:.1%}"
)],
"current_step": "compliance_check"
}
# ──────────────────────────────────────────────
# AGENT 3: RAG COMPLIANCE CHECK (LLM + Vector)
# ──────────────────────────────────────────────
def compliance_agent(state: UnderwritingState) -> dict:
"""
RAG Agent: Retrieves regulations and checks decision compliance.
WHY LLM + RAG: Compliance requires reasoning over legal text that
changes frequently. The vector store provides up-to-date regulation
retrieval, and the LLM interprets how rules apply to this specific
application. A rule engine would require constant manual updates.
"""
scores = state.get("classical_scores")
extraction = state.get("document_extraction")
application_context = f"""
SME loan application for ${state.get('structured_data', {}).get('loan_amount', 500000):,.0f}.
Credit score: {scores.credit_score if scores else 'N/A'}.
DTI: {scores.dti_ratio if scores else 'N/A'}.
PD: {scores.pd_score if scores else 'N/A'}.
Years in business: {extraction.years_in_business if extraction else 'N/A'}.
Bankruptcy history: {extraction.bankruptcy_explanation if extraction else 'None'}.
"""
compliance = rag_engine.check_compliance(
application_context=application_context,
classical_scores=scores.model_dump() if scores else {},
decision="pending_evaluation"
)
return {
"retrieved_regulations": [compliance["regulations_retrieved"]],
"compliance_result": ComplianceResult(
is_compliant=True, # Parsed from LLM response in production
applicable_guidelines=["FCRA 615(a)", "ECOA Reg B", "CP-2026-04"],
risk_tier="standard"
),
"messages": [AIMessage(
content=f"Compliance check complete. Risk tier: standard. "
f"Retrieved {len(compliance['sources'])} regulation chunks."
)],
"current_step": "fraud_detection"
}
# ──────────────────────────────────────────────
# AGENT 4: FRAUD DETECTION (Hybrid)
# ──────────────────────────────────────────────
def fraud_detection_agent(state: UnderwritingState) -> dict:
"""
Hybrid Agent: Classical velocity checks + LLM narrative analysis.
WHY HYBRID:
- Velocity features (application frequency, IP changes) → Classical
- Narrative inconsistencies (contradictory explanations) → LLM
"""
structured = state.get("structured_data", {})
narrative = state.get("applicant_narrative", "")
extraction = state.get("document_extraction")
# Classical velocity checks
velocity_alerts = []
if structured.get("applications_last_30d", 0) > 5:
velocity_alerts.append("High application velocity: >5 in 30 days")
if structured.get("ip_country_mismatch", False):
velocity_alerts.append("IP geolocation does not match business address")
# LLM narrative consistency check
prompt = f"""You are a fraud analyst. Check this loan application for
inconsistencies between the applicant's narrative and extracted documents.
NARRATIVE: {narrative}
EXTRACTED DATA: Revenue=${extraction.annual_revenue if extraction else 'N/A'},
Years in business={extraction.years_in_business if extraction else 'N/A'},
Bankruptcy explanation={extraction.bankruptcy_explanation if extraction else 'None'}
Identify any contradictions, implausible claims, or red flags.
Return JSON: fraud_probability (0-1), narrative_inconsistencies (list of strings)."""
response = llm.invoke([
SystemMessage(content="You are a fraud detection specialist. "
"Return only valid JSON."),
HumanMessage(content=prompt)
])
try:
fraud_analysis = json.loads(response.content)
except json.JSONDecodeError:
fraud_analysis = {"fraud_probability": 0.1, "narrative_inconsistencies": []}
return {
"fraud_signals": FraudSignals(
fraud_probability=fraud_analysis.get("fraud_probability", 0.1),
behavioral_flags=velocity_alerts,
narrative_inconsistencies=fraud_analysis.get(
"narrative_inconsistencies", []
),
velocity_alerts=velocity_alerts
),
"messages": [AIMessage(
content=f"Fraud check complete. Probability: "
f"{fraud_analysis.get('fraud_probability', 0.1):.1%}. "
f"Alerts: {len(velocity_alerts)}"
)],
"current_step": "decision"
}
# ──────────────────────────────────────────────
# AGENT 5: DECISION FUSION (LLM Orchestrator)
# ──────────────────────────────────────────────
def decision_agent(state: UnderwritingState) -> dict:
"""
Fusion Agent: Combines classical scores + LLM context + compliance
+ fraud signals into a final credit decision.
WHY LLM FOR FUSION: The final decision requires weighing quantitative
scores against qualitative context (e.g., a bankruptcy 3 years ago
with a compelling medical explanation). Classical models can't reason
about narrative context; LLMs can't produce calibrated PD. The fusion
agent uses the LLM to integrate both.
"""
scores = state.get("classical_scores")
extraction = state.get("document_extraction")
compliance = state.get("compliance_result")
fraud = state.get("fraud_signals")
prompt = f"""You are a senior credit officer making a final lending decision.
CLASSICAL MODEL OUTPUT (deterministic, auditable):
- Probability of Default: {scores.pd_score if scores else 'N/A'}
- Credit Score: {scores.credit_score if scores else 'N/A'}
- DTI Ratio: {scores.dti_ratio if scores else 'N/A'}
- LTV Ratio: {scores.ltv_ratio if scores else 'N/A'}
LLM-EXTRACTED CONTEXT (qualitative):
- Annual Revenue: ${extraction.annual_revenue if extraction else 'N/A'}
- Years in Business: {extraction.years_in_business if extraction else 'N/A'}
- Bankruptcy Explanation: {extraction.bankruptcy_explanation if extraction else 'None'}
- Document Confidence: {extraction.document_confidence if extraction else 'N/A'}
COMPLIANCE STATUS:
- Compliant: {compliance.is_compliant if compliance else 'N/A'}
- Risk Tier: {compliance.risk_tier if compliance else 'N/A'}
- Violations: {compliance.violated_regulations if compliance else 'None'}
FRAUD SIGNALS:
- Fraud Probability: {fraud.fraud_probability if fraud else 'N/A'}
- Behavioral Flags: {fraud.behavioral_flags if fraud else 'None'}
- Narrative Issues: {fraud.narrative_inconsistencies if fraud else 'None'}
REQUESTED LOAN: ${state.get('structured_data', {}).get('loan_amount', 500000)}
Make a decision: approve, decline, refer (to human), or conditional.
If approve/conditional: specify approved_amount, interest_rate, risk_grade (A-F),
and any conditions.
Return JSON: decision, approved_amount, interest_rate, risk_grade,
conditions (list), confidence (0-1)."""
response = llm.invoke([
SystemMessage(content="You are a conservative credit officer at a "
"regulated bank. Prioritize capital preservation. "
"Return only valid JSON."),
HumanMessage(content=prompt)
])
try:
decision_data = json.loads(response.content)
except json.JSONDecodeError:
decision_data = {
"decision": "refer", "approved_amount": 0,
"interest_rate": 0, "risk_grade": "D",
"conditions": ["Manual review required"], "confidence": 0.3
}
requires_human = decision_data.get("decision") == "refer" or \
(fraud and fraud.fraud_probability > 0.5)
return {
"final_decision": CreditDecision(**decision_data),
"requires_human_review": requires_human,
"messages": [AIMessage(
content=f"Decision: {decision_data['decision'].upper()}. "
f"Amount: ${decision_data.get('approved_amount', 0):,.0f}. "
f"Rate: {decision_data.get('interest_rate', 0):.2%}."
)],
"current_step": "explanation"
}
# ──────────────────────────────────────────────
# AGENT 6: EXPLANATION GENERATOR (LLM)
# ──────────────────────────────────────────────
def explanation_agent(state: UnderwritingState) -> dict:
"""
LLM Agent: Generates FCRA/ECOA-compliant adverse action notices
and internal explainability reports.
WHY LLM: Explanation generation requires natural language that is
legally precise yet human-readable. Template engines produce
generic notices; LLMs contextualize to the specific applicant.
"""
decision = state.get("final_decision")
scores = state.get("classical_scores")
if not decision:
return {"messages": [AIMessage(content="No decision to explain.")]}
prompt = f"""Generate two documents for this credit decision:
1. ADVERSE ACTION NOTICE (if declined/conditional, per FCRA 615(a) and ECOA):
- Specific reasons for the action
- CRA information
- Consumer rights
2. INTERNAL EXPLAINABILITY REPORT (for model validation per SR 11-7):
- Key factors driving the decision
- Classical model contribution vs. qualitative factors
- Compliance verification summary
DECISION: {decision.decision}
AMOUNT: ${decision.approved_amount:,.0f}
RATE: {decision.interest_rate:.2%}
GRADE: {decision.risk_grade}
CREDIT SCORE: {scores.credit_score if scores else 'N/A'}
PD: {scores.pd_score if scores else 'N/A'}
DTI: {scores.dti_ratio if scores else 'N/A'}
CONDITIONS: {decision.conditions}"""
response = llm.invoke([
SystemMessage(content="You are a compliance attorney specializing in "
"credit disclosure requirements. Be precise and "
"legally accurate."),
HumanMessage(content=prompt)
])
return {
"messages": [AIMessage(
content=f"Explanation generated. {len(response.content)} chars."
)],
"current_step": "complete"
}
Step 5: Build the LangGraph with Memory and Conditional Routing
# graph.py
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from state import UnderwritingState
from agents import (
intake_agent, risk_scoring_agent, compliance_agent,
fraud_detection_agent, decision_agent, explanation_agent
)
def should_escalate_to_human(state: UnderwritingState) -> str:
"""
Conditional edge: Route to human review or auto-explain.
This is where the graph makes branching decisions based on state.
"""
if state.get("requires_human_review"):
return "human_review"
return "explain"
def should_block_application(state: UnderwritingState) -> str:
"""
Compliance gate: Block if prohibited by regulation.
"""
compliance = state.get("compliance_result")
fraud = state.get("fraud_signals")
if compliance and compliance.risk_tier == "prohibited":
return "block"
if fraud and fraud.fraud_probability > 0.8:
return "block"
return "continue"
def human_review_node(state: UnderwritingState) -> dict:
"""
Placeholder for human-in-the-loop review.
In production: integrates with case management system (Pega, Appian).
Uses LangGraph's interrupt() for async human approval.
"""
from langgraph.types import interrupt
decision = state.get("final_decision")
review_data = {
"application_id": state.get("application_id"),
"proposed_decision": decision.decision if decision else "unknown",
"reason": "Fraud flag or policy exception",
"scores": state.get("classical_scores").model_dump()
if state.get("classical_scores") else {}
}
# This pauses the graph and waits for human input
human_decision = interrupt(review_data)
return {
"messages": [{"role": "human_reviewer",
"content": f"Human decision: {human_decision}"}],
"current_step": "human_reviewed"
}
def build_underwriting_graph():
"""
Construct the complete LangGraph multi-agent pipeline.
Graph topology:
START → intake → risk_scoring → compliance → fraud → decision
→ [conditional] → explain → END
→ human_review → explain → END
"""
# Initialize with memory persistence
# In production: use PostgresSaver or RedisSaver
memory = MemorySaver()
graph = StateGraph(UnderwritingState)
# Add all agent nodes
graph.add_node("intake", intake_agent)
graph.add_node("risk_scoring", risk_scoring_agent)
graph.add_node("compliance", compliance_agent)
graph.add_node("fraud_detection", fraud_detection_agent)
graph.add_node("decision", decision_agent)
graph.add_node("explain", explanation_agent)
graph.add_node("human_review", human_review_node)
# Define edges (sequential pipeline)
graph.add_edge(START, "intake")
graph.add_edge("intake", "risk_scoring")
graph.add_edge("risk_scoring", "compliance")
# Conditional: compliance gate
graph.add_conditional_edges(
"compliance",
should_block_application,
{
"continue": "fraud_detection",
"block": "explain" # Go straight to adverse action
}
)
graph.add_edge("fraud_detection", "decision")
# Conditional: human review gate
graph.add_conditional_edges(
"decision",
should_escalate_to_human,
{
"explain": "explain",
"human_review": "human_review"
}
)
graph.add_edge("human_review", "explain")
graph.add_edge("explain", END)
# Compile with memory
return graph.compile(checkpointer=memory)
Step 6: Run the Full Pipeline
# main.py
import asyncio
from graph import build_underwriting_graph
async def run_underwriting_pipeline():
"""
Execute the full multi-agent underwriting pipeline
for a real-time SME loan application.
"""
app = build_underwriting_graph()
# ── Real-World Application Data ──
initial_state = {
"application_id": "SME-2026-07-30-48291",
"applicant_name": "GreenLeaf Organics LLC",
"session_id": "sess_abc123",
# Structured data (from credit bureau + financial API)
"structured_data": {
"loan_amount": 500000,
"collateral_value": 650000,
"monthly_debt_obligations": 12000,
"monthly_income": 45000,
"num_late_payments": 1,
"credit_utilization": 0.35,
"num_inquiries_6m": 2,
"credit_age_years": 8,
"num_derogatory": 0,
"total_debt": 280000,
"annual_income": 540000,
"employment_years": 6,
"num_open_accounts": 12,
"debt_to_income": 0.27,
"applications_last_30d": 1,
"ip_country_mismatch": False,
},
# Unstructured documents (parsed from PDFs via OCR)
"unstructured_documents": [
"""2024 Federal Tax Return - GreenLeaf Organics LLC (EIN: 84-2917364)
Gross Receipts: $1,240,000
Cost of Goods Sold: $680,000
Gross Profit: $560,000
Operating Expenses: $410,000
Net Income: $150,000
Total Assets: $890,000
Total Liabilities: $340,000""",
"""Bank Statement - Chase Business Checking (Jan-Mar 2026)
Average Daily Balance: $87,400
Monthly Deposits: $105,000 avg
Monthly Withdrawals: $98,000 avg
NSF Incidents: 0
Overdrafts: 0"""
],
# Applicant's free-text explanation
"applicant_narrative": """We filed Chapter 11 bankruptcy in 2021 due to
supply chain disruptions during the pandemic that caused our primary
supplier to default on $200K in prepaid orders. We successfully
restructured and emerged from bankruptcy in March 2022. Since then,
we have grown revenue 40% year-over-year and have not missed a single
payment. The requested $500K will fund a new distribution center to
support our expanding retail partnerships.""",
# Initialize empty fields
"messages": [],
"document_extraction": None,
"classical_scores": None,
"compliance_result": None,
"fraud_signals": None,
"final_decision": None,
"retrieved_regulations": [],
"internal_policy_docs": [],
"previous_applications": [],
"conversation_history": [],
"current_step": "init",
"requires_human_review": False,
"error_log": [],
}
# ── Execute with streaming ──
config = {"configurable": {"thread_id": "SME-2026-07-30-48291"}}
print("=" * 70)
print(" ENTERPRISE CREDIT UNDERWRITING PIPELINE")
print(" Multi-Agent LangGraph | RAG Compliance | Hybrid Scoring")
print("=" * 70)
async for event in app.astream_events(
initial_state, config, version="v2"
):
kind = event.get("event")
if kind == "on_chain_start":
node_name = event.get("name", "")
if node_name in [
"intake", "risk_scoring", "compliance",
"fraud_detection", "decision", "explain"
]:
print(f"\n{'─' * 50}")
print(f"▶ AGENT: {node_name.upper()}")
print(f"{'─' * 50}")
elif kind == "on_chat_model_end":
output = event.get("data", {}).get("output")
if output:
print(f" LLM Output: {str(output.content)[:200]}...")
# ── Retrieve final state ──
final_state = await app.ainvoke(initial_state, config)
print(f"\n{'=' * 70}")
print(" FINAL DECISION SUMMARY")
print(f"{'=' * 70}")
decision = final_state.get("final_decision")
if decision:
print(f" Decision: {decision.decision.upper()}")
print(f" Approved Amount: ${decision.approved_amount:,.0f}")
print(f" Interest Rate: {decision.interest_rate:.2%}")
print(f" Risk Grade: {decision.risk_grade}")
print(f" Confidence: {decision.confidence:.1%}")
print(f" Conditions: {decision.conditions}")
scores = final_state.get("classical_scores")
if scores:
print(f"\n Classical Model Scores:")
print(f" PD: {scores.pd_score:.2%}")
print(f" Credit Score: {scores.credit_score}")
print(f" DTI: {scores.dti_ratio:.1%}")
print(f" LTV: {scores.ltv_ratio:.1%}")
print(f" Model: {scores.model_version}")
print(f"\n Human Review: {final_state.get('requires_human_review')}")
print(f" Total Messages: {len(final_state.get('messages', []))}")
print(f"{'=' * 70}")
if __name__ == "__main__":
asyncio.run(run_underwriting_pipeline())
![aa]()
Key Design Principles
Classical models are the decision drivers. The PD, scorecard, and DTI determine the risk grade. LLMs never override calibrated scores.
LLMs are the context layer. They parse documents, check compliance via RAG, detect narrative fraud, and generate explanations. They inform but don't replace the score.
RAG provides regulatory currency. Regulations change. Vector stores updated nightly from regulatory feeds ensure the LLM reasons over current rules, not stale training data.
State is the contract. LangGraph's TypedDict state ensures every agent has a typed, validated interface. No agent reads another agent's internals.
Memory enables continuity. MemorySaver (or PostgresSaver in production) persists the full application state across sessions, enabling applicants to resume, officers to review, and auditors to reconstruct decisions.
Human-in-the-loop is non-negotiable. High-fraud or policy-exception cases pause the graph via interrupt(), routing to a case management system before proceeding.
The bottom line: In enterprise credit scoring, classical models answer "what is the risk?" and LLMs answer "what is the context?" The multi-agent LangGraph architecture ensures both questions are answered, compliantly, explainably, and at scale.