RAG-Powered Credit Scoring vs. Traditional Feature-Based ML

The Fundamental Shift

Credit scoring has historically been a feature-engineering problem: take structured applicant data (income, credit history, DTI ratio), engineer features, and train a classifier (XGBoost, Logistic Regression). A RAG-powered system flips this paradigm — it treats credit scoring as a reasoning problem over heterogeneous evidence, where unstructured documents (financial statements, news, contracts, emails) are retrieved, synthesized, and reasoned about by LLM agents.

Side-by-Side Comparison

Dimension	Traditional Feature-Based ML	RAG-Powered Multi-Agent System
Input	Structured tabular features	Structured + unstructured (PDFs, news, filings)
Feature Engineering	Manual, static	Dynamic, LLM-extracted on-the-fly
Explainability	SHAP/LIME on numeric features	Natural-language reasoning traces
Adaptability	Retrain on new data	Add documents to vector store; no retraining
Auditability	Score + feature importance	Full reasoning chain (who said what, from which doc)
Handling Novel Cases	Poor (out-of-distribution)	Strong (retrieves analogous precedents)
Latency	~ms	~seconds (acceptable for underwriting)
Regulatory Fit	Mature (Basel, ECOA)	Emerging (requires human-in-the-loop)

Key insight: RAG doesn't replace traditional ML — it augments it. The best enterprise systems use both: ML for the baseline score, RAG agents for edge cases, narrative justification, and document-grounded overrides.

Real-World Use Case: SME Working-Capital Loan Underwriting

Scenario: A regional bank receives a $500K working-capital loan application from "Northwind Logistics," a mid-sized trucking firm. The underwriter has:

3 years of audited financials (PDFs)
Bank statements (CSV)
A news article about a fuel-price spike affecting the sector
An existing credit bureau report
A prior default case from a similar company (precedent)

A traditional model would only see the bureau score + DTI. A RAG-powered system retrieves the financials, the news, the precedent, and has specialized agents reason about each — then a supervisor agent produces a final recommendation with citations.

Architecture Overview

Tech Stack

Orchestration: LangGraph (stateful, cyclic graphs, human-in-the-loop)
LLM: OpenAI GPT-4o (or any chat model)
Vector Store: Chroma (local) / Pinecone (prod)
Embeddings: text-embedding-3-small
Checkpointer: SQLite (demo) / PostgreSQL (prod)
Document Loaders: LangChain PyPDFLoader, CSVLoader
Memory: LangGraph's built-in MemorySaver + thread-scoped state

End-to-End Code Implementation

Install Dependencies

pip install langgraph langchain langchain-openai langchain-chroma \
            langchain-community chromadb pypdf pandas

State Definition: The Heart of LangGraph

The State is the single source of truth. Every agent reads from it and writes to it. This is what makes the system deterministic, resumable, and auditable.

from typing import Annotated, Any, Dict, List, Optional
from pydantic import BaseModel, Field
from langgraph.graph.message import add_messages
from typing_extensions import TypedDict

class CreditState(TypedDict):
    """Global state shared across all agents."""
    # Messages (conversation history with add_messages reducer)
    messages: Annotated[list, add_messages]
    
    # Application context
    application_id: str
    applicant_name: str
    loan_amount: float
    loan_purpose: str
    
    # Retrieved evidence (populated by RAG agents)
    financials_summary: Optional[str]
    market_context: Optional[str]
    precedent_cases: Optional[str]
    bureau_signals: Optional[str]
    
    # Agent outputs
    financial_analysis: Optional[str]
    risk_assessment: Optional[str]
    decision: Optional[str]
    citations: List[str]
    
    # Control flow
    current_phase: str
    requires_human_review: bool
    final_score: Optional[float]

RAG Infrastructure

import os
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyPDFLoader, CSVLoader

os.environ["OPENAI_API_KEY"] = "sk-..."

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Separate collections per evidence type (better retrieval precision)
financial_store = Chroma(
    collection_name="financials",
    embedding_function=embeddings,
    persist_directory="./chroma_db"
)
market_store = Chroma(
    collection_name="market_news",
    embedding_function=embeddings,
    persist_directory="./chroma_db"
)
precedent_store = Chroma(
    collection_name="precedents",
    embedding_function=embeddings,
    persist_directory="./chroma_db"
)

def ingest_document(file_path: str, store: Chroma, metadata: dict):
    """Ingest a document into the appropriate vector store."""
    if file_path.endswith(".pdf"):
        loader = PyPDFLoader(file_path)
    else:
        loader = CSVLoader(file_path)
    
    docs = loader.load()
    for d in docs:
        d.metadata.update(metadata)
    
    splitter = RecursiveCharacterTextSplitter(chunk_size=1500, chunk_overlap=200)
    chunks = splitter.split_documents(docs)
    store.add_documents(chunks)
    return len(chunks)

def retrieve(store: Chroma, query: str, k: int = 5) -> str:
    """Retrieve and format top-k chunks."""
    results = store.similarity_search(query, k=k)
    formatted = []
    for i, doc in enumerate(results, 1):
        src = doc.metadata.get("source", "unknown")
        formatted.append(f"[{i}] (Source: {src})\n{doc.page_content}")
    return "\n\n---\n\n".join(formatted)

Agent Definitions

Each agent is a node in the LangGraph. It reads state, performs work, and writes back to state.

from langchain_core.messages import HumanMessage, SystemMessage

llm = ChatOpenAI(model="gpt-4o", temperature=0.1)

# ---------- Ingestion Agent ----------
def ingestion_agent(state: CreditState) -> dict:
    """Ingests applicant documents into the vector store."""
    app_id = state["application_id"]
    # In production, these paths come from a document management system
    ingest_document("./docs/financials.pdf", financial_store, 
                    {"app_id": app_id, "type": "financial"})
    ingest_document("./docs/news_fuel_prices.pdf", market_store,
                    {"app_id": app_id, "type": "market"})
    ingest_document("./docs/precedent_logistics_default.pdf", precedent_store,
                    {"app_id": app_id, "type": "precedent"})
    
    return {
        "messages": [HumanMessage(content="Documents ingested successfully.")],
        "current_phase": "retrieval"
    }

# ---------- Financial Analyst Agent ----------
def financial_analyst(state: CreditState) -> dict:
    """Retrieves financial docs and produces a structured analysis."""
    query = (f"Financial performance of {state['applicant_name']} "
             f"requesting ${state['loan_amount']:,.0f} for {state['loan_purpose']}")
    context = retrieve(financial_store, query, k=6)
    
    prompt = f"""You are a senior credit analyst. Analyze the financial evidence 
and produce a structured assessment covering:
1. Revenue trend (3Y)
2. Profitability margins
3. Liquidity (current ratio, quick ratio)
4. Leverage (debt-to-equity)
5. Cash flow adequacy for the requested loan

Evidence:
{context}

Output as markdown with clear sections."""
    
    response = llm.invoke([SystemMessage(content=prompt)])
    return {
        "financials_summary": context,
        "financial_analysis": response.content,
        "current_phase": "market_analysis"
    }

# ---------- Market Monitor Agent ----------
def market_monitor(state: CreditState) -> dict:
    """Retrieves market/sector news and assesses macro risk."""
    query = f"Market conditions affecting {state['applicant_name']} sector"
    context = retrieve(market_store, query, k=4)
    
    prompt = f"""Assess current market/sector risks relevant to this applicant.
Focus on: commodity exposure, regulatory changes, demand shifts, competitive pressure.

Evidence:
{context}

Provide a concise risk memo (max 300 words)."""
    
    response = llm.invoke([SystemMessage(content=prompt)])
    return {
        "market_context": context,
        "risk_assessment": response.content,
        "current_phase": "precedent_review"
    }

# ---------- Precedent Research Agent ----------
def precedent_researcher(state: CreditState) -> dict:
    """Finds analogous historical cases."""
    query = f"Similar SME loan default or success in logistics sector"
    context = retrieve(precedent_store, query, k=3)
    
    prompt = f"""Identify lessons from these historical cases relevant to 
{state['applicant_name']}'s application. Highlight parallels and divergences.

Evidence:
{context}"""
    
    response = llm.invoke([SystemMessage(content=prompt)])
    return {
        "precedent_cases": context,
        "messages": [HumanMessage(content=f"Precedent review: {response.content[:200]}...")],
        "current_phase": "decision"
    }

# ---------- Decision Agent ----------
def decision_agent(state: CreditState) -> dict:
    """Synthesizes all analyses into a final recommendation."""
    prompt = f"""You are the Chief Credit Officer. Based on the analyses below, 
produce a final credit decision for {state['applicant_name']}.

FINANCIAL ANALYSIS:
{state['financial_analysis']}

MARKET RISK:
{state['risk_assessment']}

LOAN DETAILS: ${state['loan_amount']:,.0f} for {state['loan_purpose']}

Output JSON with keys:
- recommendation: "approve" | "conditional_approve" | "decline" | "human_review"
- score: float 0-100
- rationale: str
- conditions: list[str] (if conditional)
- citations: list[str] (document references used)"""
    
    response = llm.invoke([SystemMessage(content=prompt)])
    
    # Parse (in production, use structured output / with_structured_output)
    import json
    try:
        decision = json.loads(response.content)
    except json.JSONDecodeError:
        decision = {"recommendation": "human_review", "score": 50, 
                    "rationale": response.content, "conditions": [], "citations": []}
    
    requires_human = decision["recommendation"] in ("human_review", "conditional_approve")
    
    return {
        "decision": response.content,
        "final_score": decision.get("score"),
        "citations": decision.get("citations", []),
        "requires_human_review": requires_human,
        "current_phase": "complete"
    }

Routing Logic & Graph Construction

from langgraph.graph import StateGraph, START, END

def route_after_retrieval(state: CreditState) -> str:
    """Conditional routing based on current phase."""
    phase = state.get("current_phase", "retrieval")
    if phase == "retrieval":
        return "financial_analyst"
    elif phase == "market_analysis":
        return "market_monitor"
    elif phase == "precedent_review":
        return "precedent_researcher"
    elif phase == "decision":
        return "decision_agent"
    return END

# Build the graph
workflow = StateGraph(CreditState)

# Add nodes
workflow.add_node("ingestion", ingestion_agent)
workflow.add_node("financial_analyst", financial_analyst)
workflow.add_node("market_monitor", market_monitor)
workflow.add_node("precedent_researcher", precedent_researcher)
workflow.add_node("decision_agent", decision_agent)

# Edges
workflow.add_edge(START, "ingestion")
workflow.add_edge("ingestion", "financial_analyst")
workflow.add_edge("financial_analyst", "market_monitor")
workflow.add_edge("market_monitor", "precedent_researcher")
workflow.add_edge("precedent_researcher", "decision_agent")
workflow.add_edge("decision_agent", END)

# Compile with persistence (this is the MEMORY)
from langgraph.checkpoint.sqlite import SqliteSaver

memory = SqliteSaver.from_conn_string("./credit_workflow.db")
app = workflow.compile(checkpointer=memory)

Running the Workflow

def run_underwriting(application_id: str, applicant: str, amount: float, purpose: str):
    """Execute the full underwriting workflow."""
    config = {"configurable": {"thread_id": application_id}}
    
    initial_state = {
        "messages": [HumanMessage(content=f"Start underwriting for {applicant}")],
        "application_id": application_id,
        "applicant_name": applicant,
        "loan_amount": amount,
        "loan_purpose": purpose,
        "current_phase": "ingestion",
        "requires_human_review": False,
        "citations": [],
    }
    
    final_state = app.invoke(initial_state, config)
    
    print("=" * 60)
    print(f"APPLICATION: {applicant} | AMOUNT: ${amount:,.0f}")
    print("=" * 60)
    print("\n📊 FINANCIAL ANALYSIS:\n", final_state["financial_analysis"])
    print("\n🌍 MARKET RISK:\n", final_state["risk_assessment"])
    print("\n⚖️ DECISION:\n", final_state["decision"])
    print(f"\n🎯 SCORE: {final_state['final_score']}/100")
    print(f"👁️  HUMAN REVIEW REQUIRED: {final_state['requires_human_review']}")
    print(f"📎 CITATIONS: {final_state['citations']}")
    
    return final_state

# Execute
result = run_underwriting(
    application_id="APP-2026-0042",
    applicant="Northwind Logistics",
    amount=500_000,
    purpose="Working capital for fleet expansion"
)

Resuming & Human-in-the-Loop (Memory in Action)

Because we used a SqliteSaver checkpointer, the workflow state is persisted. An underwriter can resume, inspect, or override at any point:

# Resume the same thread days later
config = {"configurable": {"thread_id": "APP-2026-0042"}}
history = app.get_state_history(config)

for checkpoint in history:
    print(f"Phase: {checkpoint.values.get('current_phase')} | "
          f"Score: {checkpoint.values.get('final_score')}")

# Override decision (human-in-the-loop)
app.update_state(config, {"decision": "APPROVED by senior underwriter", 
                          "final_score": 78}, as_node="decision_agent")

Why This Matters for Enterprise

Memory ≠ Context Window

Context window = what the LLM sees in one call (transient).
LangGraph memory = persisted state across runs, threads, and days. Enables audit trails, resumable workflows, and multi-session underwriting.

State = Single Source of Truth

Every agent writes to the same CreditState. No hidden side channels. Regulators can replay the exact reasoning path.

Multi-Agent Specialization

Each agent is a domain expert with its own prompt, tools, and retrieval scope. This is more reliable than one monolithic prompt trying to do everything.

Citations & Auditability

Because every retrieval is stored in state, the final decision includes traceable citations — critical for ECOA/GDPR adverse-action notices.

Production Hardening Checklist

Concern	Solution
Hallucination	Strict RAG: agents only reason over retrieved chunks; citation enforcement
Latency	Parallel agent execution via Send() API; cache embeddings
Cost	Tiered models (GPT-4o for decision, GPT-4o-mini for extraction)
Drift	Monitor retrieval relevance; re-embed on model upgrades
Compliance	Immutable state log; PII redaction before vectorization
Evaluation	RAGAS / TruLens for retrieval quality; LLM-as-judge for reasoning
Fallback	If any agent fails, route to human_review node

When to Use Which Approach

Use traditional ML when: high-volume, low-touch, fully structured data, sub-100ms latency required (e.g., card transaction fraud).
Use RAG multi-agent when: heterogeneous evidence, edge cases, regulatory explainability, or novel situations matter (e.g., commercial lending, mortgage underwriting, insurance claims).
Use both (hybrid): ML produces the base score; RAG agents provide narrative justification, handle exceptions, and generate compliance documents.

Summary

Traditional feature-based machine learning and RAG-powered multi-agent systems address different aspects of credit underwriting. While ML provides fast baseline scoring from structured data, RAG augments decision-making by retrieving and reasoning over heterogeneous evidence with traceable citations. Using LangGraph, typed state, persistent memory, specialized agents, and human-in-the-loop workflows, organizations can build underwriting systems that are more explainable, auditable, and adaptable for complex lending scenarios.