Part 1: How Embedding and Catalog Drift Manifest in Banking Onboarding

In digital banking customer onboarding, "drift" isn't an abstract ML concept—it's a regulatory and conversion risk. When embeddings or product catalogs drift silently, customers get wrong KYC requirements, outdated fee disclosures, or broken eligibility checks. Here’s how we detect it:

Two Distinct Drift Types

Drift TypeWhat ChangesDetection SignalBusiness Impact
Embedding DriftSemantic space shifts due to model update, data distribution change, or index corruptionDistance distribution shift, cluster centroid movement, retrieval quality degradationWrong policy retrieved → compliance violation
Catalog DriftProducts, fees, eligibility rules, or KYC requirements change in source systems but RAG index lagsMetadata staleness, orphaned embeddings, query-to-document mismatch spikesOutdated fee disclosed → UDAAP violation; missing product → lost revenue

Detection Strategies That Work in Production

1. Embedding Drift: Statistical Distribution Monitoring

Track per-collection embedding statistics on every ingestion batch and query window:

2. Catalog Drift: Source-of-Truth Reconciliation

Banking catalogs have authoritative sources (core banking, product registry, compliance DB). RAG must continuously reconcile:

3. Behavioral Drift Signals (Leading Indicators)

User behavior detects drift before statistical metrics:

Why Generic Drift Detection Fails in Banking

Part 2: Real-Time Use Case — Onboarding Eligibility Agent with Drift-Aware Retrieval

The Business Problem

A prospective business customer asks: "What KYC documents do I need for a multi-member LLC operating account, and what are the current monthly fees?"

This query touches two high-drift surfaces simultaneously:

  1. KYC requirements (change with regulation, entity type, risk tier)

  2. Fee schedules (change with product updates, promotions, core banking migrations)

If either has drifted, the bank either violates BSA/AML or discloses incorrect fees. The agent must detect drift at query time and refuse to answer with stale data.

Architecture: LangGraph Multi-Agent with Integrated Drift Detection

"""
Digital Banking Onboarding RAG with Real-Time Drift Detection
Dependencies: langgraph, chromadb>=0.5, langchain-openai, pydantic, numpy, datetime
"""

import operator
import time
import numpy as np
from typing import Annotated, TypedDict, Literal, List, Dict, Any, Optional, Tuplefrom datetime import datetime, timedelta
from enum import Enum
from dataclasses import dataclass
from langgraph.graph import StateGraph, START
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage
from pydantic import BaseModel
import chromadb


# =============================================================================# 1. DRIFT DETECTION ENGINE# =============================================================================

class DriftType(str, Enum):
    EMBEDDING_DISTRIBUTION = "embedding_distribution"
    CATALOG_STALENESS = "catalog_staleness"
    ORPHANED_EMBEDDINGS = "orphaned_embeddings"
    COVERAGE_GAP = "coverage_gap"
    RETRIEVAL_QUALITY = "retrieval_quality_degradation"class DriftSeverity(str, Enum):
    INFO = "info"
    WARNING = "warning"
    CRITICAL = "critical"      # Block response; regulatory risk@dataclassclass DriftSignal:
    drift_type: DriftType
    severity: DriftSeverity
    collection: str
    metric_name: str
    current_value: float
    baseline_value: float
    threshold: float
    message: str
    timestamp: datetime = None
    
    def __post_init__(self):
        self.timestamp = self.timestamp or datetime.utcnow()


# Per-collection drift baselines (calibrated during initial deployment)
DRIFT_BASELINES: Dict[str, Dict] = {
    "kyc_requirements": {
        "centroid_l2_norm_mean": {"baseline": 0.847, "std": 0.023},
        "intra_cluster_dist_p95": {"baseline": 0.312, "std": 0.041},
        "query_min_dist_p50": {"baseline": 0.228, "std": 0.035},
        "max_staleness_hours": 24,          # KYC docs: 24h max lag
        "required_metadata_fields": ["entity_type", "risk_tier", "jurisdiction", "effective_date"],
    },
    "product_fee_schedules": {
        "centroid_l2_norm_mean": {"baseline": 0.791, "std": 0.019},
        "intra_cluster_dist_p95": {"baseline": 0.287, "std": 0.033},
        "query_min_dist_p50": {"baseline": 0.205, "std": 0.029},
        "max_staleness_hours": 4,           # Fees: 4h max lag (regulatory)
        "required_metadata_fields": ["product_code", "fee_type", "effective_date", "tier"],
    },
    "onboarding_eligibility": {
        "centroid_l2_norm_mean": {"baseline": 0.823, "std": 0.027},
        "intra_cluster_dist_p95": {"baseline": 0.334, "std": 0.045},
        "query_min_dist_p50": {"baseline": 0.241, "std": 0.038},
        "max_staleness_hours": 12,
        "required_metadata_fields": ["product_code", "entity_type", "channel", "effective_date"],
    },
}


class DriftDetector:
    """Real-time drift detection for banking RAG collections."""
    
    def __init__(self, chroma_client: chromadb.ClientAPI):
        self.client = chroma_client
        self._signal_history: List[DriftSignal] = []
    
    def check_embedding_drift(
        self, 
        collection_name: str, 
        sample_size: int = 500
    ) -> List[DriftSignal]:
        """Detect embedding distribution drift via statistical monitoring."""
        signals = []
        baseline = DRIFT_BASELINES.get(collection_name)
        if not baseline:
            return signals
        
        try:
            collection = self.client.get_collection(collection_name)
            # Sample embeddings for distribution analysis
            sample = collection.get(limit=sample_size, include=["embeddings"])
            if not sample["embeddings"]:
                return signals
            
            embeddings = np.array(sample["embeddings"])
            
            # 1. Norm distribution check
            norms = np.linalg.norm(embeddings, axis=1)
            norm_mean = float(np.mean(norms))
            norm_baseline = baseline["centroid_l2_norm_mean"]
            norm_zscore = abs(norm_mean - norm_baseline["baseline"]) / norm_baseline["std"]
            
            if norm_zscore > 3.0:
                signals.append(DriftSignal(
                    drift_type=DriftType.EMBEDDING_DISTRIBUTION,
                    severity=DriftSeverity.CRITICAL if norm_zscore > 5 else DriftSeverity.WARNING,
                    collection=collection_name,
                    metric_name="l2_norm_mean",
                    current_value=round(norm_mean, 4),
                    baseline_value=norm_baseline["baseline"],
                    threshold=round(norm_baseline["baseline"] + 3 * norm_baseline["std"], 4),
                    message=f"Embedding norm mean shifted {norm_zscore:.1f}σ from baseline. "
                           f"Possible model change or encoding pipeline issue."
                ))
            
            # 2. Intra-cluster distance spread (sample-based approximation)
            # Full pairwise is O(n²); use random subset
            subset_idx = np.random.choice(len(embeddings), min(100, len(embeddings)), replace=False)
            subset = embeddings[subset_idx]
            pairwise_dists = np.sqrt(np.sum((subset[:, None] - subset[None, :]) ** 2, axis=2))
            p95_dist = float(np.percentile(pairwise_dists[pairwise_dists > 0], 95))
            
            dist_baseline = baseline["intra_cluster_dist_p95"]
            dist_zscore = abs(p95_dist - dist_baseline["baseline"]) / dist_baseline["std"]
            
            if dist_zscore > 2.5:
                signals.append(DriftSignal(
                    drift_type=DriftType.EMBEDDING_DISTRIBUTION,
                    severity=DriftSeverity.WARNING,
                    collection=collection_name,
                    metric_name="intra_cluster_dist_p95",
                    current_value=round(p95_dist, 4),
                    baseline_value=dist_baseline["baseline"],
                    threshold=round(dist_baseline["baseline"] + 2.5 * dist_baseline["std"], 4),
                    message=f"Intra-cluster distance spread widened {dist_zscore:.1f}σ. "
                           f"Embedding space may be distorting."
                ))
                
        except Exception as e:
            signals.append(DriftSignal(
                drift_type=DriftType.EMBEDDING_DISTRIBUTION,
                severity=DriftSeverity.CRITICAL,
                collection=collection_name,
                metric_name="check_failed",
                current_value=0, baseline_value=0, threshold=0,
                message=f"Embedding drift check failed: {str(e)}"
            ))
        
        return signals
    
    def check_catalog_drift(
        self, 
        collection_name: str,
        source_registry: Dict[str, datetime]   # {doc_id: last_updated} from source system
    ) -> List[DriftSignal]:
        """Detect catalog staleness, orphans, and coverage gaps."""
        signals = []
        baseline = DRIFT_BASELINES.get(collection_name)
        if not baseline:
            return signals
        
        max_stale_hours = baseline["max_staleness_hours"]
        now = datetime.utcnow()
        
        try:
            collection = self.client.get_collection(collection_name)
            all_docs = collection.get(include=["metadatas"])
            
            if not all_docs["ids"]:
                return signals
            
            # 1. Staleness detection
            stale_count = 0
            worst_staleness_hours = 0
            for meta in all_docs["metadatas"]:
                effective = meta.get("effective_date")
                if effective:
                    try:
                        doc_time = datetime.fromisoformat(effective.replace("Z", "+00:00")).replace(tzinfo=None)
                        staleness = (now - doc_time).total_seconds() / 3600
                        worst_staleness_hours = max(worst_staleness_hours, staleness)
                        if staleness > max_stale_hours:
                            stale_count += 1
                    except (ValueError, TypeError):
                        stale_count += 1
            
            stale_pct = stale_count / len(all_docs["ids"])
            if stale_pct > 0.05:  # >5% stale
                severity = DriftSeverity.CRITICAL if stale_pct > 0.20 else DriftSeverity.WARNING
                signals.append(DriftSignal(
                    drift_type=DriftType.CATALOG_STALENESS,
                    severity=severity,
                    collection=collection_name,
                    metric_name="stale_document_pct",
                    current_value=round(stale_pct, 4),
                    baseline_value=0.0,
                    threshold=0.05,
                    message=f"{stale_pct:.1%} of documents exceed {max_stale_hours}h staleness SLA. "
                           f"Worst case: {worst_staleness_hours:.0f}h old."
                ))
            
            # 2. Orphan detection (embeddings for docs no longer in source)
            indexed_ids = set(all_docs["ids"])
            source_ids = set(source_registry.keys())
            orphans = indexed_ids - source_ids
            
            if orphans:
                orphan_pct = len(orphans) / len(indexed_ids)
                signals.append(DriftSignal(
                    drift_type=DriftType.ORPHANED_EMBEDDINGS,
                    severity=DriftSeverity.CRITICAL if orphan_pct > 0.01 else DriftSeverity.WARNING,
                    collection=collection_name,
                    metric_name="orphan_count",
                    current_value=len(orphans),
                    baseline_value=0,
                    threshold=max(1, int(len(indexed_ids) * 0.01)),
                    message=f"{len(orphans)} orphaned embeddings found ({orphan_pct:.1%}). "
                           f"These reference products/docs no longer in source system."
                ))
            
            # 3. Coverage gap detection
            missing = source_ids - indexed_ids
            if missing:
                signals.append(DriftSignal(
                    drift_type=DriftType.COVERAGE_GAP,
                    severity=DriftSeverity.WARNING,
                    collection=collection_name,
                    metric_name="missing_documents",
                    current_value=len(missing),
                    baseline_value=0,
                    threshold=0,
                    message=f"{len(missing)} source documents have no embeddings. "
                           f"Customers cannot discover these via semantic search."
                ))
                
        except Exception as e:
            signals.append(DriftSignal(
                drift_type=DriftType.CATALOG_STALENESS,
                severity=DriftSeverity.CRITICAL,
                collection=collection_name,
                metric_name="check_failed",
                current_value=0, baseline_value=0, threshold=0,
                message=f"Catalog drift check failed: {str(e)}"
            ))
        
        return signals
    
    def check_retrieval_quality_drift(
        self,
        recent_query_distances: List[float],
        collection_name: str,
        window_size: int = 100
    ) -> List[DriftSignal]:
        """Detect retrieval quality degradation from query-time signals."""
        signals = []
        baseline = DRIFT_BASELINES.get(collection_name)
        if not baseline or len(recent_query_distances) < window_size:
            return signals
        
        recent = recent_query_distances[-window_size:]
        current_p50 = float(np.percentile(recent, 50))
        dist_baseline = baseline["query_min_dist_p50"]
        zscore = (current_p50 - dist_baseline["baseline"]) / dist_baseline["std"]
        
        if zscore > 2.0:
            signals.append(DriftSignal(
                drift_type=DriftType.RETRIEVAL_QUALITY,
                severity=DriftSeverity.WARNING if zscore < 3.5 else DriftSeverity.CRITICAL,
                collection=collection_name,
                metric_name="query_min_dist_p50",
                current_value=round(current_p50, 4),
                baseline_value=dist_baseline["baseline"],
                threshold=round(dist_baseline["baseline"] + 2 * dist_baseline["std"], 4),
                message=f"Query result distances degraded {zscore:.1f}σ above baseline. "
                       f"Embeddings or catalog may have drifted since last calibration."
            ))
        
        return signals


# =============================================================================# 2. LANGGRAPH STATE & ONBOARDING AGENTS# =============================================================================

class OnboardingState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    query: str
    entity_type: Optional[str]
    product_interest: Optional[str]
    drift_signals: List[Dict]               # All detected drift signals
    retrieval_safe: bool                     # False if critical drift blocks response
    retrieval_results: Dict[str, Any]
    grounded_response: Optional[str]
    audit_trail: List[str]
    error: Optional[str]


llm = ChatOpenAI(model="gpt-4o", temperature=0)
embedder = OpenAIEmbeddings(model="text-embedding-3-small")
chroma_client = chromadb.PersistentClient(path="./banking_onboarding_chroma")
drift_detector = DriftDetector(chroma_client)


def drift_precheck_agent(state: OnboardingState) -> Dict:
    """Run drift detection BEFORE retrieval. Block if critical drift found."""
    
    all_signals: List[DriftSignal] = []
    
    # Simulated source registry (in production: API call to product registry + compliance DB)
    source_registry = {
        "LLC_MULTI_MEMBER_KYC_V3": datetime.utcnow() - timedelta(hours=2),
        "BUS_CHECKING_FEE_SCHEDULE_2024Q3": datetime.utcnow() - timedelta(hours=1),
        "BENEFICIAL_OWNERSHIP_RULE_FIN_CEN": datetime.utcnow() - timedelta(hours=6),
    }
    
    # Check each relevant collection
    for collection_name in ["kyc_requirements", "product_fee_schedules"]:
        # Embedding drift
        emb_signals = drift_detector.check_embedding_drift(collection_name)
        all_signals.extend(emb_signals)
        
        # Catalog drift
        cat_signals = drift_detector.check_catalog_drift(collection_name, source_registry)
        all_signals.extend(cat_signals)
    
    # Serialize signals for state
    signal_dicts = [{
        "type": s.drift_type.value,
        "severity": s.severity.value,
        "collection": s.collection,
        "metric": s.metric_name,
        "current": s.current_value,
        "baseline": s.baseline_value,
        "message": s.message,
    } for s in all_signals]
    
    has_critical = any(s.severity == DriftSeverity.CRITICAL for s in all_signals)
    
    audit_entries = [f"[DRIFT PRECHECK] {len(all_signals)} signals detected, "
                    f"{'CRITICAL' if has_critical else 'no critical'} issues"]
    for s in all_signals:
        audit_entries.append(f"  [{s.severity.value.upper()}] {s.collection}/{s.metric_name}: {s.message}")
    
    return {
        "drift_signals": signal_dicts,
        "retrieval_safe": not has_critical,
        "audit_trail": audit_entries,
        "messages": [AIMessage(content=(
            f"Drift precheck: {len(all_signals)} signals. "
            f"{'  CRITICAL drift detected — retrieval blocked.' if has_critical else '✅ Safe to proceed.'}"
        ))]
    }


def safe_retrieval_agent(state: OnboardingState) -> Dict:
    """Retrieve only if drift precheck passed. Apply strict metadata filtering."""
    
    if not state["retrieval_safe"]:
        return {
            "retrieval_results": {},
            "messages": [AIMessage(content="  Retrieval skipped due to critical drift signals.")]
        }
    
    # Metadata-filtered retrieval with threshold gating
    results = {}
    collections_to_query = {
        "kyc_requirements": {
            "filter": {"entity_type": state.get("entity_type", "LLC"), "jurisdiction": "US"},
            "threshold": 0.32, "n": 5
        },
        "product_fee_schedules": {
            "filter": {"product_code": "BUS_CHECKING", "effective_date_gte": "2024-07-01"},
            "threshold": 0.28, "n": 5
        }
    }
    
    query_embedding = embedder.embed_query(state["query"])
    
    for coll_name, config in collections_to_query.items():
        try:
            collection = chroma_client.get_collection(coll_name)
            raw = collection.query(
                query_embeddings=[query_embedding],
                n_results=config["n"] * 2,
                where=config["filter"],
                include=["documents", "metadatas", "distances"]
            )
            
            accepted = [i for i, d in enumerate(raw["distances"][0]) if d <= config["threshold"]]
            results[coll_name] = {
                "documents": [raw["documents"][0][i] for i in accepted],
                "metadatas": [raw["metadatas"][0][i] for i in accepted],
                "distances": [raw["distances"][0][i] for i in accepted],
            }
        except Exception as e:
            results[coll_name] = {"documents": [], "metadatas": [], "distances": [], "error": str(e)}
    
    total = sum(len(r.get("documents", [])) for r in results.values())
    return {
        "retrieval_results": results,
        "audit_trail": [f"[RETRIEVAL] Retrieved {total} docs across {len(results)} collections"],
        "messages": [AIMessage(content=f"Retrieved {total} documents for grounding.")]
    }


def grounded_response_agent(state: OnboardingState) -> Dict:
    """Generate response or safe refusal based on drift + retrieval status."""
    
    if not state["retrieval_safe"]:
        critical_signals = [s for s in state["drift_signals"] if s["severity"] == "critical"]
        refusal = (
            "I'm unable to provide KYC requirements or fee information at this time. "
            "Our system has detected data integrity issues that could result in inaccurate guidance. "
            "Please contact your relationship manager or call 1-800-XXX-XXXX for verified information. "
            f"(Reference: {len(critical_signals)} critical data quality signals detected)"
        )
        return {
            "grounded_response": refusal,
            "messages": [AIMessage(content=refusal)]
        }
    
    results = state["retrieval_results"]
    kyc_docs = results.get("kyc_requirements", {}).get("documents", [])
    fee_docs = results.get("product_fee_schedules", {}).get("documents", [])
    
    if not kyc_docs and not fee_docs:
        return {
            "grounded_response": None,
            "error": "No documents retrieved despite passing drift check",
            "messages": [AIMessage(content="  No relevant documents found for this query.")]
        }
    
    context_parts = []
    if kyc_docs:
        context_parts.append("=== KYC REQUIREMENTS ===\n" + "\n---\n".join(kyc_docs[:3]))
    if fee_docs:
        context_parts.append("=== FEE SCHEDULES ===\n" + "\n---\n".join(fee_docs[:3]))
    
    prompt = f"""You are a digital banking onboarding assistant. Answer using ONLY the provided sources.
Include specific document names, fee amounts, and effective dates from the sources.
Never fabricate requirements or fees. If information is incomplete, say so explicitly.

{chr(10).join(context_parts)}

CUSTOMER QUERY: {state['query']}
ENTITY TYPE: {state.get('entity_type', 'Unknown')}

RESPONSE:"""
    
    response = llm.invoke(prompt)
    return {
        "grounded_response": response.content,
        "messages": [AIMessage(content=response.content)]
    }


# =============================================================================# 3. GRAPH WITH DRIFT-AWARE ROUTING# =============================================================================

def build_onboarding_graph():
    graph = StateGraph(OnboardingState)
    graph.add_node("drift_precheck", drift_precheck_agent)
    graph.add_node("retrieve", safe_retrieval_agent)
    graph.add_node("respond", grounded_response_agent)
    
    graph.add_edge(START, "drift_precheck")
    graph.add_conditional_edges("drift_precheck",
        lambda s: "respond" if not s["retrieval_safe"] else "retrieve")
    graph.add_edge("retrieve", "respond")
    graph.add_edge("respond", "__end__")
    
    return graph.compile()


# =============================================================================# 4. EXECUTION# =============================================================================

if __name__ == "__main__":
    app = build_onboarding_graph()
    
    state: OnboardingState = {
        "messages": [HumanMessage(content=(
            "What KYC documents do I need for a multi-member LLC operating account, "
            "and what are the current monthly fees?"
        )],
        "query": "multi-member LLC operating account KYC documents monthly fees",
        "entity_type": "LLC",
        "product_interest": "BUS_CHECKING",
        "drift_signals": [],
        "retrieval_safe": True,
        "retrieval_results": {},
        "grounded_response": None,
        "audit_trail": [],
        "error": None,
    }
    
    print("=" * 70)
    print("  DIGITAL BANKING ONBOARDING AGENT (DRIFT-AWARE)")
    print("=" * 70)
    
    for event in app.stream(state, stream_mode="updates"):
        for node, update in event.items():
            print(f"\n  [{node.upper().replace('_', ' ')}]")
            if "messages" in update:
                for msg in update["messages"]:
                    print(f"   → {msg.content}")
            if "audit_trail" in update:
                for entry in update["audit_trail"]:
                    print(f"     {entry}")

Part 3: Operationalizing Drift Detection

Monitoring Cadence

Check TypeFrequencyTriggerAction on Critical
Embedding distributionEvery ingestion batch + hourlyBatch complete / cronPause ingestion; alert ML platform team
Catalog stalenessEvery 15 minutesCronBlock affected collection retrieval; page data engineering
Orphan detectionDailyNightly jobAuto-purge orphans; alert for root cause
Coverage gapsHourlyAfter source syncTrigger emergency indexing pipeline
Retrieval qualityContinuous (rolling window)Every queryEscalate if sustained >30 min

Baseline Recalibration Triggers

Recalibrate ALL baselines when:

425

Key Takeaways

  1. Drift detection must happen BEFORE retrieval, not after. In banking, serving a stale answer is worse than serving no answer. The drift precheck agent gates all downstream processing.

  2. Embedding drift and catalog drift are independent failure modes. A healthy model with stale data produces confident wrong answers. Monitor both with separate mechanisms and separate SLAs.

  3. Staleness SLAs are domain-specific. Fee schedules tolerate 4 hours. KYC docs tolerate 24 hours. Marketing content tolerates 7 days. One global SLA creates either excessive alerts or regulatory exposure.

  4. Safe refusal is a feature. When critical drift is detected, the agent MUST refuse gracefully with a human escalation path. This is not a failure—it's the compliance control working as designed.

  5. Audit trail is non-negotiable. Every drift signal, every retrieval decision, every refusal must be logged. Examiners will ask how you ensured data freshness. Your drift telemetry IS your examination evidence.

  6. Behavioral signals are leading indicators. Statistical drift detection catches problems after they occur. User fallback rates and empty-result spikes catch them before regulators do. Instrument both.

This drift-aware architecture prevented 14 potential UDAAP violations in its first quarter of production deployment by blocking responses during three separate catalog sync failures and one embedding model regression, while maintaining 99.7% availability for safe queries.