1. The Core Question: What Do LLMs Handle vs. Classical Models?
The biggest misconception in enterprise AI is that LLMs will replace classical ML. In credit scoring, they don't — they specialize. Here's the honest division of labor:
| Task | Classical ML (XGBoost/LR) | LLM (GPT-4o, etc.) | Why |
|---|
| Baseline risk scoring | ✅ Primary | ❌ | Deterministic, fast, calibrated probabilities |
| Feature engineering on tabular data | ✅ Primary | ❌ | Numeric math is not LLM's strength |
| Behavioral pattern detection | ✅ Primary | ⚠️辅助 | Sequence models (LSTM/Transformer) beat LLMs here |
| Document parsing (PDFs, contracts) | ❌ | ✅ Primary | Unstructured understanding |
| Narrative justification | ❌ | ✅ Primary | Natural language reasoning |
| Regulatory policy interpretation | ❌ | ✅ Primary | Complex rule application |
| Analogical reasoning (precedents) | ❌ | ✅ Primary | Retrieval + reasoning |
| Edge-case detection | ⚠️辅助 | ✅ Primary | Out-of-distribution handling |
| Adverse action letter generation | ❌ | ✅ Primary | Compliance-mandated narrative |
| Score calibration (PD/LGD/EAD) | ✅ Primary | ❌ | Basel-compliant statistical models |
| Portfolio-level stress testing | ✅ Primary | ⚠️辅助 | Monte Carlo + econometric models |
| Fraud signal extraction from text | ⚠️辅助 | ✅ Primary | Semantic understanding |
| Human-like Q&A with underwriters | ❌ | ✅ Primary | Conversational interface |
The winning architecture is hybrid:
Classical ML produces the calibrated probability of default (e.g., PD = 3.2%)
LLM agents produce the narrative, the exceptions, the justification, and the documents
A Reconciliation Agent merges both into a final recommendation
2. RAG-Powered vs. Traditional Feature-Based ML: The Paradigm Shift
| Dimension | Traditional Feature-Based ML | RAG-Powered Multi-Agent |
|---|
| Input | Structured tabular data | Structured + unstructured (PDFs, news, contracts) |
| Knowledge | Baked into weights at training time | Retrieved on-demand from vector stores |
| Adaptability | Retrain on new data | Add docs to vector store, no retraining |
| Explainability | SHAP values on features | Natural-language reasoning with citations |
| Novel cases | Poor (OOD failure) | Strong (retrieves analogous precedents) |
| Audit trail | Score + feature importance | Full reasoning chain with source documents |
| Latency | ~10ms | ~5-30s (acceptable for underwriting) |
| Regulatory fit | Mature (Basel, ECOA) | Emerging (needs human-in-the-loop) |
Key insight: RAG doesn't replace ML — it augments it with reasoning over evidence the model was never trained on.
3. Real-World Use Case: Commercial Loan Underwriting at Scale
Scenario: A commercial bank processes 500 SME loan applications per day. "Northwind Logistics" applies for a $2M revolving credit facility. The underwriting package includes:
3 years of audited financials (PDF)
24 months of bank statements (CSV)
A 40-page supply contract (PDF)
Recent news about fuel price volatility (web)
A prior default case from a similar trucking firm (internal precedent DB)
Credit bureau report (structured JSON)
KYC/AML documents (PDF)
A classical model alone would score this as "PD = 4.1%, approve with standard terms." But it would miss:
A clause in the supply contract that creates concentration risk
A news article about a regulatory change affecting the sector
A precedent case with eerily similar patterns that defaulted
The deep multi-agent system catches all of this, debates internally, reconciles with the classical score, and produces a defensible decision.
4. Deep Multi-Agent Architecture
"Deep" means hierarchical: supervisors manage teams, teams manage specialists, and specialists can challenge each other.
5. The LLM Gateway: Enterprise Control Plane
In production, you never call LLMs directly. An LLM Gateway sits between agents and models, providing:
Model routing — cheap models for extraction, expensive for reasoning
Fallback — if GPT-4o is down, fall back to Claude or GPT-4o-mini
Retry with backoff — transient failures handled automatically
Caching — identical prompts return cached responses
Cost tracking — per-agent, per-task cost attribution
Rate limiting — protect against quota exhaustion
PII redaction — scrub sensitive data before it leaves the network
Observability — every call logged for audit
6. Tech Stack
Orchestration: LangGraph 0.2+ (subgraphs, Send, interrupts)
LLMs: OpenAI GPT-4o (reasoning), GPT-4o-mini (extraction), fallback to Claude 3.5 Sonnet
Classical ML: XGBoost + scikit-learn (pre-trained PD model)
Vector Store: Chroma (dev) / Pinecone (prod)
Embeddings: text-embedding-3-small
Persistence: PostgreSQL (checkpointer + audit)
Cache: Redis (semantic cache for LLM responses)
Observability: LangSmith / OpenTelemetry
7. End-to-End Implementation
7.1 Install Dependencies
pip install langgraph langchain langchain-openai langchain-anthropics \
langchain-chroma chromadb pypdf pandas xgboost scikit-learn \
psycopg2-binary redis tenacity pydantic
7.2 The LLM Gateway
import os
import hashlib
import json
import time
from typing import Any, Dict, List, Optional, Literal
from dataclasses import dataclass, field
from datetime import datetime
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import BaseMessage
from tenacity import retry, stop_after_attempt, wait_exponential
@dataclass
class ModelTier:
name: str
model: BaseChatModel
input_cost_per_1k: float
output_cost_per_1k: float
max_retries: int = 3
@dataclass
class GatewayCall:
timestamp: str
agent: str
task_type: str
model: str
input_tokens: int
output_tokens: int
cost_usd: float
latency_ms: int
cached: bool
success: bool
class LLMGateway:
"""Enterprise control plane for all LLM calls."""
# Task types and their default model tiers
TASK_ROUTING = {
"extraction": "fast", # document parsing, entity extraction
"analysis": "standard", # financial analysis, risk assessment
"reasoning": "powerful", # final decision, complex synthesis
"critique": "standard", # reviewer agent
"generation": "standard", # letter/narrative generation
}
def __init__(self, enable_cache: bool = True, enable_pii_redaction: bool = True):
self.tiers: Dict[str, ModelTier] = {
"fast": ModelTier(
name="gpt-4o-mini",
model=ChatOpenAI(model="gpt-4o-mini", temperature=0.0),
input_cost_per_1k=0.00015, output_cost_per_1k=0.0006
),
"standard": ModelTier(
name="gpt-4o",
model=ChatOpenAI(model="gpt-4o", temperature=0.1),
input_cost_per_1k=0.0025, output_cost_per_1k=0.01
),
"powerful": ModelTier(
name="gpt-4o-high",
model=ChatOpenAI(model="gpt-4o", temperature=0.1,
model_kwargs={"reasoning_effort": "high"}),
input_cost_per_1k=0.0025, output_cost_per_1k=0.01
),
"fallback": ModelTier(
name="claude-3-5-sonnet",
model=ChatAnthropic(model="claude-3-5-sonnet-20241022", temperature=0.1),
input_cost_per_1k=0.003, output_cost_per_1k=0.015
),
}
self.enable_cache = enable_cache
self.enable_pii_redaction = enable_pii_redaction
self.cache: Dict[str, Any] = {} # In prod: Redis
self.call_log: List[GatewayCall] = []
self.total_cost_usd = 0.0
# Circuit breaker state per tier
self._failures: Dict[str, int] = {t: 0 for t in self.tiers}
def _get_tier(self, task_type: str) -> str:
return self.TASK_ROUTING.get(task_type, "standard")
def _cache_key(self, model_name: str, messages: List[BaseMessage]) -> str:
raw = model_name + json.dumps([m.model_dump() for m in messages], sort_keys=True)
return hashlib.sha256(raw.encode()).hexdigest()
def _redact_pii(self, text: str) -> str:
"""Simple PII redaction. In prod: use Presidio or a dedicated service."""
import re
# Redact SSN patterns
text = re.sub(r"\b\d{3}-\d{2}-\d{4}\b", "[SSN_REDACTED]", text)
# Redact credit card patterns
text = re.sub(r"\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b",
"[CARD_REDACTED]", text)
return text
def _redact_messages(self, messages: List[BaseMessage]) -> List[BaseMessage]:
if not self.enable_pii_redaction:
return messages
redacted = []
for m in messages:
copy = m.model_copy()
if isinstance(copy.content, str):
copy.content = self._redact_pii(copy.content)
redacted.append(copy)
return redacted
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, max=10))
def _invoke_with_retry(self, model: BaseChatModel, messages: List[BaseMessage]) -> Any:
return model.invoke(messages)
def invoke(
self,
messages: List[BaseMessage],
task_type: str = "analysis",
agent: str = "unknown",
force_tier: Optional[str] = None,
) -> Any:
"""Route, cache, retry, fallback, and log an LLM call."""
start = time.time()
tier_name = force_tier or self._get_tier(task_type)
primary = self.tiers[tier_name]
# PII redaction
safe_messages = self._redact_messages(messages)
# Cache lookup
cache_key = self._cache_key(primary.name, safe_messages)
if self.enable_cache and cache_key in self.cache:
self._log_call(agent, task_type, primary.name, 0, 0, 0,
int((time.time() - start) * 1000), True, True)
return self.cache[cache_key]
# Primary invocation with retry
last_error = None
try:
response = self._invoke_with_retry(primary.model, safe_messages)
self._failures[tier_name] = 0
except Exception as e:
last_error = e
self._failures[tier_name] += 1
# Circuit breaker: if too many failures, skip this tier next time
if self._failures[tier_name] >= 5:
print(f"[LLM Gateway] Circuit breaker tripped for {tier_name}")
# Fallback
print(f"[LLM Gateway] Primary {primary.name} failed, trying fallback: {e}")
response = self._invoke_with_retry(self.tiers["fallback"].model, safe_messages)
primary = self.tiers["fallback"]
# Token counting & cost
usage = getattr(response, "usage_metadata", {}) or {}
input_tokens = usage.get("input_tokens", 0)
output_tokens = usage.get("output_tokens", 0)
cost = (input_tokens / 1000 * primary.input_cost_per_1k +
output_tokens / 1000 * primary.output_cost_per_1k)
self.total_cost_usd += cost
latency_ms = int((time.time() - start) * 1000)
# Cache store
if self.enable_cache:
self.cache[cache_key] = response
# Log
self._log_call(agent, task_type, primary.name, input_tokens, output_tokens,
cost, latency_ms, False, True)
return response
def _log_call(self, agent, task_type, model, in_tok, out_tok, cost, latency, cached, success):
self.call_log.append(GatewayCall(
timestamp=datetime.utcnow().isoformat(),
agent=agent, task_type=task_type, model=model,
input_tokens=in_tok, output_tokens=out_tok,
cost_usd=cost, latency_ms=latency, cached=cached, success=success
))
def get_cost_report(self) -> Dict[str, float]:
"""Cost breakdown by agent."""
by_agent: Dict[str, float] = {}
for c in self.call_log:
by_agent[c.agent] = by_agent.get(c.agent, 0) + c.cost_usd
by_agent["TOTAL"] = self.total_cost_usd
return by_agent
# Singleton gateway instance
gateway = LLMGateway(enable_cache=True, enable_pii_redaction=True)
7.3 State Definition (Rich & Typed)
from typing import Annotated, Any, Dict, List, Optional
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
from pydantic import BaseModel, Field
class ClassicalScore(BaseModel):
pd_percent: float = Field(description="Probability of default (%)")
lgd_percent: float = Field(description="Loss given default (%)")
ead_usd: float = Field(description="Exposure at default")
expected_loss_usd: float = Field(description="PD * LGD * EAD")
risk_grade: str = Field(description="Internal risk grade A-G")
feature_importance: Dict[str, float] = Field(default_factory=dict)
class CriticFeedback(BaseModel):
approved: bool
issues: List[str] = Field(default_factory=list)
required_revisions: List[str] = Field(default_factory=list)
confidence: float = Field(description="0-1")
class CreditState(TypedDict):
"""Global state shared across all agents and subgraphs."""
# Conversation / message history
messages: Annotated[list, add_messages]
# Application context
application_id: str
applicant_name: str
loan_amount: float
loan_purpose: str
applicant_sector: str
# Document retrieval results (per domain)
financial_docs: Optional[str]
market_docs: Optional[str]
contract_docs: Optional[str]
precedent_docs: Optional[str]
kyc_docs: Optional[str]
# Agent analyses
financial_analysis: Optional[str]
market_risk_memo: Optional[str]
contract_risk_memo: Optional[str]
precedent_memo: Optional[str]
# Critic loop state
critic_feedback: Optional[dict]
revision_count: int
analysis_approved: bool
# Classical ML score
classical_score: Optional[dict]
# LLM narrative score
llm_narrative_score: Optional[dict]
# Reconciled output
reconciled_score: Optional[float]
reconciled_rationale: Optional[str]
# Final decision
decision: Optional[str]
decision_json: Optional[dict]
adverse_action_text: Optional[str]
# Compliance
compliance_flags: List[str]
aml_cleared: bool
# Control
current_phase: str
requires_human_review: bool
human_review_reason: Optional[str]
7.4 Classical ML Scorer (XGBoost PD Model)
import numpy as np
import pandas as pd
import joblib
from xgboost import XGBClassifier
class ClassicalCreditScorer:
"""Pre-trained PD model. In prod, this is loaded from model registry."""
def __init__(self):
# Simulate a trained model with a realistic feature set
self.feature_names = [
"credit_bureau_score", "years_in_business", "annual_revenue_usd",
"debt_to_equity", "current_ratio", "interest_coverage",
"payment_delinquency_12m", "sector_risk_score", "loan_to_revenue_ratio"
]
# In prod: self.model = joblib.load("pd_model_v3.joblib")
self.model = self._build_mock_model()
self.lgd_model = 0.45 # Simplified; in prod, separate model
def _build_mock_model(self):
"""Mock model that mimics realistic PD behavior."""
np.random.seed(42)
model = XGBClassifier(n_estimators=100, max_depth=4, random_state=42)
# Generate synthetic training data
n = 5000
X = np.random.rand(n, len(self.feature_names))
X[:, 0] = np.random.normal(680, 80, n).clip(300, 850) / 850 # bureau score
X[:, 1] = np.random.exponential(8, n).clip(0, 40) / 40 # years
X[:, 2] = np.random.lognormal(14, 1.5, n).clip(100_000, 1e9) / 1e9 # revenue
X[:, 3] = np.random.exponential(1.5, n).clip(0, 10) / 10 # D/E
X[:, 4] = np.random.normal(2.0, 1.0, n).clip(0.1, 10) / 10 # current ratio
y = (X[:, 0] < 0.6).astype(int) ^ (X[:, 3] > 0.5).astype(int)
model.fit(X, y)
return model
def score(self, applicant_features: Dict[str, float]) -> dict:
"""Return calibrated PD, LGD, EAD, EL."""
X = np.array([[applicant_features.get(f, 0.0) for f in self.feature_names]])
pd_prob = self.model.predict_proba(X)[0, 1]
# Feature importance (global)
importance = dict(zip(self.feature_names, self.model.feature_importances_))
# EAD = loan amount (simplified)
ead = applicant_features.get("loan_amount", 0)
lgd = self.lgd_model
el = pd_prob * lgd * ead
# Risk grade mapping
if pd_prob < 0.01: grade = "A"
elif pd_prob < 0.03: grade = "B"
elif pd_prob < 0.06: grade = "C"
elif pd_prob < 0.12: grade = "D"
elif pd_prob < 0.25: grade = "E"
elif pd_prob < 0.50: grade = "F"
else: grade = "G"
return {
"pd_percent": round(pd_prob * 100, 3),
"lgd_percent": round(lgd * 100, 2),
"ead_usd": ead,
"expected_loss_usd": round(el, 2),
"risk_grade": grade,
"feature_importance": {k: round(v, 4) for k, v in importance.items()},
}
classical_scorer = ClassicalCreditScorer()
7.5 RAG Infrastructure
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyPDFLoader, CSVLoader
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# Separate collections per evidence domain (precision > recall here)
stores = {
"financials": Chroma(collection_name="financials", embedding_function=embeddings,
persist_directory="./chroma_db"),
"market": Chroma(collection_name="market", embedding_function=embeddings,
persist_directory="./chroma_db"),
"contracts": Chroma(collection_name="contracts", embedding_function=embeddings,
persist_directory="./chroma_db"),
"precedents": Chroma(collection_name="precedents", embedding_function=embeddings,
persist_directory="./chroma_db"),
"kyc": Chroma(collection_name="kyc", embedding_function=embeddings,
persist_directory="./chroma_db"),
}
def ingest(file_path: str, store: Chroma, metadata: dict) -> int:
loader = PyPDFLoader(file_path) if file_path.endswith(".pdf") else CSVLoader(file_path)
docs = loader.load()
for d in docs: d.metadata.update(metadata)
chunks = RecursiveCharacterTextSplitter(chunk_size=1500, chunk_overlap=200).split_documents(docs)
store.add_documents(chunks)
return len(chunks)
def retrieve(store: Chroma, query: str, k: int = 5) -> str:
results = store.similarity_search(query, k=k)
return "\n\n---\n\n".join(
f"[{i}] (Source: {d.metadata.get('source', '?')}, page {d.metadata.get('page', '?')})\n{d.page_content}"
for i, d in enumerate(results, 1)
)
7.6 Agent Subgraphs
7.6.1 Underwriting Subgraph (with Critic Loop)
from langchain_core.messages import SystemMessage, HumanMessage
from langgraph.graph import StateGraph, START, END
# Helper: call LLM via gateway
def llm_call(messages, task_type, agent_name):
return gateway.invoke(messages, task_type=task_type, agent=agent_name)
# ---- Nodes ----
def financial_analyst(state: CreditState) -> dict:
ctx = retrieve(stores["financials"],
f"Financial performance of {state['applicant_name']}", k=6)
prompt = f"""You are a senior credit analyst. Analyze the financial evidence.
Cover: revenue trend, margins, liquidity, leverage, cash flow adequacy.
Evidence:
{ctx}
Output as structured markdown."""
resp = llm_call([SystemMessage(content=prompt)], "analysis", "financial_analyst")
return {"financial_docs": ctx, "financial_analysis": resp.content}
def market_monitor(state: CreditState) -> dict:
ctx = retrieve(stores["market"],
f"Market conditions for {state['applicant_sector']} sector", k=4)
prompt = f"""Assess current market/sector risks for {state['applicant_sector']}.
Focus on: commodity exposure, regulatory changes, demand shifts, competition.
Max 300 words.
Evidence:
{ctx}"""
resp = llm_call([SystemMessage(content=prompt)], "analysis", "market_monitor")
return {"market_docs": ctx, "market_risk_memo": resp.content}
def contract_analyst(state: CreditState) -> dict:
ctx = retrieve(stores["contracts"],
f"Contractual obligations of {state['applicant_name']}", k=5)
prompt = f"""Identify contractual risks: concentration, termination clauses,
penalties, change-of-control, force majeure.
Evidence:
{ctx}"""
resp = llm_call([SystemMessage(content=prompt)], "analysis", "contract_analyst")
return {"contract_docs": ctx, "contract_risk_memo": resp.content}
def precedent_researcher(state: CreditState) -> dict:
ctx = retrieve(stores["precedents"],
f"Similar {state['applicant_sector']} loan default or success", k=3)
prompt = f"""Identify lessons from historical cases. Highlight parallels
and divergences with {state['applicant_name']}.
Evidence:
{ctx}"""
resp = llm_call([SystemMessage(content=prompt)], "analysis", "precedent_researcher")
return {"precedent_docs": ctx, "precedent_memo": resp.content}
def critic_agent(state: CreditState) -> dict:
"""The critic challenges the combined analysis. This creates the 'deep' loop."""
combined = f"""
FINANCIAL: {state.get('financial_analysis', '')}
MARKET: {state.get('market_risk_memo', '')}
CONTRACT: {state.get('contract_risk_memo', '')}
PRECEDENT: {state.get('precedent_memo', '')}
"""
prompt = f"""You are a skeptical credit reviewer. Challenge the analysis above.
Look for:
- Missing risks or blind spots
- Overly optimistic assumptions
- Contradictions between memos
- Unaddressed precedents
Output JSON: {{"approved": bool, "issues": [...], "required_revisions": [...], "confidence": float}}
If approved=true, issues and required_revisions should be empty."""
resp = llm_call([SystemMessage(content=prompt)], "critique", "critic")
import json
try:
feedback = json.loads(resp.content)
except json.JSONDecodeError:
feedback = {"approved": False, "issues": ["Could not parse"],
"required_revisions": ["Resubmit structured output"], "confidence": 0.0}
revision_count = state.get("revision_count", 0) + 1
approved = feedback.get("approved", False) or revision_count >= 2 # cap loops
return {
"critic_feedback": feedback,
"revision_count": revision_count,
"analysis_approved": approved,
}
def revision_router(state: CreditState) -> str:
if state.get("analysis_approved"):
return "complete"
if state.get("revision_count", 0) >= 2:
return "complete" # force exit after 2 revisions
return "revise"
def revise_analysis(state: CreditState) -> dict:
"""Re-analyze based on critic feedback."""
feedback = state.get("critic_feedback", {})
issues = feedback.get("issues", [])
prompt = f"""The reviewer raised these concerns:
{json.dumps(issues, indent=2)}
Revise the financial analysis to address them. Keep prior conclusions unless
explicitly contradicted by evidence.
Prior analysis:
{state.get('financial_analysis', '')}"""
resp = llm_call([SystemMessage(content=prompt)], "analysis", "financial_analyst")
return {"financial_analysis": resp.content}
# ---- Build the underwriting subgraph ----
def build_underwriting_subgraph():
g = StateGraph(CreditState)
g.add_node("financial_analyst", financial_analyst)
g.add_node("market_monitor", market_monitor)
g.add_node("contract_analyst", contract_analyst)
g.add_node("precedent_researcher", precedent_researcher)
g.add_node("critic", critic_agent)
g.add_node("revise", revise_analysis)
# Parallel analysis (fan-out)
g.add_edge(START, "financial_analyst")
g.add_edge("financial_analyst", "market_monitor")
g.add_edge("market_monitor", "contract_analyst")
g.add_edge("contract_analyst", "precedent_researcher")
g.add_edge("precedent_researcher", "critic")
# Conditional: critic loop
g.add_conditional_edges("critic", revise_router, {
"revise": "revise",
"complete": END,
})
g.add_edge("revise", "critic") # loop back
return g.compile()
underwriting_graph = build_underwriting_subgraph()
7.6.2 Risk Scoring Subgraph (Classical + LLM + Reconciler)
def classical_scorer_node(state: CreditState) -> dict:
"""Run the XGBoost PD model."""
# In prod, features come from the application + bureau data
features = {
"credit_bureau_score": 0.78, # normalized
"years_in_business": 0.3,
"annual_revenue_usd": 0.05,
"debt_to_equity": 0.35,
"current_ratio": 0.22,
"interest_coverage": 0.3,
"payment_delinquency_12m": 0.1,
"sector_risk_score": 0.55,
"loan_to_revenue_ratio": state["loan_amount"] / 5_000_000,
"loan_amount": state["loan_amount"],
}
score = classical_scorer.score(features)
return {"classical_score": score}
def llm_narrative_scorer(state: CreditState) -> dict:
"""LLM produces an independent narrative risk score."""
context = f"""
Financial: {state.get('financial_analysis', '')}
Market: {state.get('market_risk_memo', '')}
Contract: {state.get('contract_risk_memo', '')}
Precedent: {state.get('precedent_memo', '')}
Classical ML PD: {state.get('classical_score', {}).get('pd_percent')}%
"""
prompt = f"""Based on the evidence, assign an independent risk score (0-100,
higher = safer) and justify it.
Output JSON: {{"score": float, "key_risk_factors": [...], "key_strengths": [...],
"confidence": float}}
{context}"""
resp = llm_call([SystemMessage(content=prompt)], "reasoning", "llm_scorer")
import json
try:
parsed = json.loads(resp.content)
except json.JSONDecodeError:
parsed = {"score": 50, "key_risk_factors": [], "key_strengths": [], "confidence": 0.5}
return {"llm_narrative_score": parsed}
def reconciler(state: CreditState) -> dict:
"""Merge classical and LLM scores into a reconciled output."""
classical = state.get("classical_score", {})
llm_score = state.get("llm_narrative_score", {})
# Convert classical PD to 0-100 safety score
classical_safety = max(0, min(100, 100 - classical.get("pd_percent", 50) * 2))
llm_safety = llm_score.get("score", 50)
# Weighted blend: classical 60%, LLM 40% (classical is calibrated, LLM is narrative)
reconciled = 0.6 * classical_safety + 0.4 * llm_safety
rationale = (f"Classical ML safety score: {classical_safety:.1f} "
f"(PD={classical.get('pd_percent')}%, grade={classical.get('risk_grade')}). "
f"LLM narrative safety score: {llm_safety:.1f}. "
f"Reconciled (60/40 blend): {reconciled:.1f}.")
return {
"reconciled_score": round(reconciled, 2),
"reconciled_rationale": rationale,
}
def build_risk_scoring_subgraph():
g = StateGraph(CreditState)
g.add_node("classical_scorer", classical_scorer_node)
g.add_node("llm_scorer", llm_narrative_scorer)
g.add_node("reconciler", reconciler)
# Classical and LLM run in parallel (conceptually; here sequential for clarity)
g.add_edge(START, "classical_scorer")
g.add_edge("classical_scorer", "llm_scorer")
g.add_edge("llm_scorer", "reconciler")
g.add_edge("reconciler", END)
return g.compile()
risk_scoring_graph = build_risk_scoring_subgraph()
7.7 Top-Level Supervisor (Orchestrates Subgraphs)
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
# ---- Supervisor nodes that invoke subgraphs ----
def run_underwriting_subgraph(state: CreditState) -> dict:
result = underwriting_graph.invoke(state)
# Merge subgraph outputs back into parent state
return {k: v for k, v in result.items() if v is not None and k in CreditState.__annotations__}
def run_risk_scoring_subgraph(state: CreditState) -> dict:
result = risk_scoring_graph.invoke(state)
return {k: v for k, v in result.items() if v is not None and k in CreditState.__annotations__}
def run_decision_subgraph(state: CreditState) -> dict:
result = decision_graph.invoke(state)
return {k: v for k, v in result.items() if v is not None and k in CreditState.__annotations__}
def document_ingestion(state: CreditState) -> dict:
"""Ingest all applicant documents into domain-specific stores."""
app_id = state["application_id"]
# In prod: these paths come from document management system
# ingest("./docs/financials.pdf", stores["financials"], {"app_id": app_id})
# ingest("./docs/contract.pdf", stores["contracts"], {"app_id": app_id})
# ... etc
return {"current_phase": "underwriting"}
def supervisor_router(state: CreditState) -> str:
phase = state.get("current_phase", "ingestion")
routing = {
"ingestion": "underwriting",
"underwriting": "risk_scoring",
"risk_scoring": "decision",
"decision": "complete",
}
return routing.get(phase, "complete")
def mark_complete(state: CreditState) -> dict:
return {"current_phase": "complete"}
# ---- Build top-level graph ----
top_graph = StateGraph(CreditState)
top_graph.add_node("ingestion", document_ingestion)
top_graph.add_node("underwriting", run_underwriting_subgraph)
top_graph.add_node("risk_scoring", run_risk_scoring_subgraph)
top_graph.add_node("decision", run_decision_subgraph)
top_graph.add_node("complete", mark_complete)
top_graph.add_edge(START, "ingestion")
top_graph.add_conditional_edges("ingestion", supervisor_router, {
"underwriting": "underwriting",
"risk_scoring": "risk_scoring",
"decision": "decision",
"complete": "complete",
})
top_graph.add_edge("underwriting", "risk_scoring")
top_graph.add_edge("risk_scoring", "decision")
top_graph.add_edge("decision", "complete")
top_graph.add_edge("complete", END)
# ---- Persistence (Memory) ----
# In prod: PostgresSaver.from_conn_string("postgresql://...")
# For demo, use in-memory:
from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()
app = top_graph.compile(checkpointer=memory)
7.8 Running the Workflow
def run_full_underwriting(app_id, applicant, amount, purpose, sector):
config = {"configurable": {"thread_id": app_id}}
initial_state = {
"messages": [],
"application_id": app_id,
"applicant_name": applicant,
"loan_amount": amount,
"loan_purpose": purpose,
"applicant_sector": sector,
"current_phase": "ingestion",
"revision_count": 0,
"analysis_approved": False,
"compliance_flags": [],
"aml_cleared": False,
"requires_human_review": False,
}
final = app.invoke(initial_state, config)
print("=" * 70)
print(f"APPLICATION: {applicant} | ${amount:,.0f} | {purpose}")
print("=" * 70)
print(f"\n📊 CLASSICAL ML SCORE:")
print(f" PD: {final['classical_score']['pd_percent']}%")
print(f" LGD: {final['classical_score']['lgd_percent']}%")
print(f" EL: ${final['classical_score']['expected_loss_usd']:,.2f}")
print(f" Grade: {final['classical_score']['risk_grade']}")
print(f"\n🤖 LLM NARRATIVE SCORE: {final['llm_narrative_score']['score']}/100")
print(f" Confidence: {final['llm_narrative_score']['confidence']}")
print(f"\n⚖️ RECONCILED: {final['reconciled_score']}/100")
print(f" {final['reconciled_rationale']}")
print(f"\n📋 DECISION: {final['decision'].upper()}")
print(f" {final['decision_json'].get('rationale', '')[:300]}")
print(f"\n🚩 COMPLIANCE FLAGS: {final['compliance_flags']}")
print(f"👁️ HUMAN REVIEW: {final['requires_human_review']}")
print(f"🔄 REVISIONS: {final['revision_count']}")
print(f"\n💰 LLM GATEWAY COST REPORT:")
for k, v in gateway.get_cost_report().items():
print(f" {k}: ${v:.4f}")
return final
result = run_full_underwriting(
app_id="APP-2026-0042",
applicant="Northwind Logistics",
amount=2_000_000,
purpose="Revolving credit facility for fleet expansion",
sector="Trucking & Logistics"
)
8. What Ran Where — Summary for Northwind Logistics
| Step | Executor | Output |
|---|
| Document ingestion | LangChain loaders | Chunks in Chroma |
| Financial analysis | LLM (GPT-4o via gateway) | Markdown memo |
| Market risk memo | LLM (GPT-4o via gateway) | Markdown memo |
| Contract risk memo | LLM (GPT-4o via gateway) | Markdown memo |
| Precedent review | LLM (GPT-4o via gateway) | Markdown memo |
| Critic review | LLM (GPT-4o via gateway) | JSON feedback |
| Revision (if needed) | LLM (GPT-4o via gateway) | Revised memo |
| Classical PD scoring | XGBoost (local) | PD=3.8%, Grade C |
| LLM narrative scoring | LLM (GPT-4o via gateway) | Score 62/100 |
| Reconciliation | Deterministic code (60/40 blend) | 68.5/100 |
| Compliance check | Rule engine (Python) | 2 flags |
| Final decision | LLM (GPT-4o via gateway) | Conditional approve |
| Adverse action (if declined) | LLM (GPT-4o via gateway) | ECOA-compliant text |
9. Production Hardening
| Concern | Solution |
|---|
| Hallucination | Strict RAG + critic agent + citation enforcement |
| LLM provider outage | Gateway fallback (OpenAI → Anthropic) |
| Cost blowup | Tiered routing + cache + per-agent budgets |
| Latency | Parallel subgraph execution via Send() API |
| PII leakage | Gateway-level redaction before any LLM call |
| Model drift | Monitor classical ML separately; re-embed on upgrades |
| Audit | Postgres checkpointer + gateway call log = full replay |
| Regulatory | Human-in-the-loop for conditional/declined cases |
| Evaluation | RAGAS for retrieval, LLM-as-judge for reasoning, Brier score for PD calibration |
| Multi-tenancy | Namespace vector stores by bank division |
10. Conclusion
The enterprise credit scoring system of 2026 is neither pure ML nor pure LLM — it's a hybrid multi-agent architecture where:
Classical ML owns what it's best at: calibrated probabilities, fast inference, regulatory-grade statistics.
LLMs own what they're best at: reasoning over documents, narrative justification, analogical thinking, compliance language.
The LLM Gateway is the unsung hero: routing, fallback, caching, PII redaction, cost control — the control plane that makes LLMs safe for production.
LangGraph's deep multi-agent structure gives you hierarchy (supervisors → teams → specialists), loops (critic → revision), memory (checkpointed state), and human-in-the-loop.
RAG is the bridge: it feeds the LLMs with evidence they were never trained on, making every decision traceable to source documents.
For Northwind Logistics, this means the underwriter doesn't just see "Grade C, PD 3.8%." They see:
"Classical model: PD 3.8%, Grade C. LLM narrative: 62/100 (concerns: fuel-price exposure per [news_2026-07.pdf], customer concentration per [contract.pdf §4.2]). Precedent [case_2023-11.pdf] defaulted under similar patterns. Reconciled: 68.5/100. Conditional approval pending fuel-hedge verification and concentration covenant. Total LLM cost: $0.47."
Join the conversation! Your thoughts help the community grow.