LLMs vs. Classical Models in Credit Scoring

Tuhin Paul

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

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())

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.

LLM RAGhybrid AIcredit scoringclassical models

Comments

Join the conversation! Your thoughts help the community grow.