Detecting Embedding Drift and Catalog Drift in Financial RAG

In accounting and finance, data is non-stationary. GAAP standards update quarterly, chart of accounts structures change during M&A, and vendor contracts are renegotiated monthly. A RAG system that performed well at deployment will silently degrade without active drift detection. Here are the proven mechanisms:

1. Embedding Drift Detection

Embedding drift occurs when the semantic space of incoming documents shifts relative to the indexed corpus, causing retrieval degradation even if keywords remain similar.

Centroid Distance Monitoring: Compute the rolling centroid of new document embeddings (e.g., weekly batch) and measure cosine distance against the historical corpus centroid. A sustained shift >0.05–0.10 in high-dimensional space signals distributional change. Track this per collection (e.g., gaap_standards, vendor_contracts).
Retrieval Confidence Decay: Log the similarity scores of top-K retrieved chunks over time. A declining median score or increasing variance indicates the index no longer represents the query distribution. Set alert thresholds at the 7th-day rolling average.
Query-Document Alignment Score: Use a lightweight cross-encoder on a sampled subset of production queries to re-rank retrieved results. If the cross-encoder consistently disagrees with the vector retriever’s ranking, the embedding model has drifted from the domain semantics.
Silent Failure Proxy: Track "no relevant result found" rates and user feedback (thumbs down / regeneration requests). In finance, these often precede measurable metric shifts by days.

2. Catalog Drift Detection

Catalog drift refers to structural changes in financial metadata: new GL codes, deprecated cost centers, updated entity hierarchies, or changed regulatory tags.

Metadata Schema Validation Pipeline: Before ingestion, validate every document’s metadata against a live schema registry (e.g., Confluent Schema Registry or custom JSON Schema). Reject or flag documents with unknown GL codes, missing required fields (entity_id, fiscal_period), or deprecated tags. Log rejection rates as a catalog health KPI.
Taxonomy Version Pinning: Never allow implicit taxonomy evolution. Each Chroma collection should be versioned (gl_accounts_v2024Q3, gl_accounts_v2024Q4). When a new fiscal period introduces structural changes, create a new collection rather than mutating the old one. Route queries based on temporal context.
Reference Integrity Checks: Periodically scan the vector store for orphaned references—documents pointing to GL codes or entities that no longer exist in the ERP master data. These cause hallucinations when agents cite outdated structures.
Change Data Capture (CDC) Integration: Connect Chroma ingestion directly to ERP/financial system CDC streams (e.g., Debezium → Kafka → Chroma). This ensures the vector catalog reflects source-of-truth changes within minutes, not batch windows. Alert on CDC lag >15 minutes for real-time financial close support.

3. Operationalizing Detection

Drift detection must be automated, not ad-hoc:

Observability Stack: Emit centroid distances, retrieval scores, and schema validation metrics to Prometheus/Grafana. Create dashboards per collection with SLO-based alerting.
Canary Evaluation Suite: Maintain a curated test set of 200+ financial Q&A pairs with ground-truth answers. Run this suite nightly against the live index. Track pass@5 and citation accuracy. A >3% drop triggers automatic rollback or re-indexing.
Human-in-the-Loop Feedback Loop: For high-stakes queries (audit evidence, tax positions), capture CPA/analyst corrections. Feed these back as hard negatives into fine-tuning or re-ranking models. This converts drift incidents into improvement signals.

Use Case: "LedgerMind" – Intelligent Financial Close Assistant

Scenario: A multinational corporation’s accounting team manages monthly close across 12 entities, 3 ERP systems, and evolving GAAP/IFRS standards. During close, accountants ask: "Show me the accrual entries for Entity DE-03 in Q3 2025, verify they comply with ASC 606 revenue recognition, and flag any unreconciled intercompany balances."

This requires:

GL Retrieval Agent: Queries journal entries with strict entity/period filtering.
Standards Compliance Agent: Validates against current GAAP/IFRS guidance.
Reconciliation Agent: Cross-references sub-ledgers and intercompany accounts.
Orchestrator: Maintains audit trail state, routes based on query intent, enforces temporal scoping.

Architecture: LangGraph + Chroma + Drift-Aware Memory

End-to-End Implementation

Prerequisites

pip install langgraph langchain-chroma chromadb langchain-openai pydantic prometheus-client

Step 1: Define Financial Domain State & Schemas

from typing import Annotated, TypedDict, Literalfrom langgraph.graph.message import add_messages
from pydantic import BaseModel, Field
from datetime import date

class FinancialState(TypedDict):
    messages: Annotated[list, add_messages]
    entity_id: str | None           # e.g., "DE-03", "US-HQ"
    fiscal_period: str | None       # e.g., "2025-Q3"
    gl_codes: list[str]             # Extracted account codes
    compliance_refs: list[str]      # Cited ASC/IFRS paragraphs
    recon_gaps: list[dict]          # Unreconciled items
    audit_trail: list[dict]         # Immutable decision log
    final_response: str | None

class FinancialIntent(BaseModel):
    """Structured extraction for financial query routing"""
    entity_id: str = Field(description="Entity code or 'ALL'")
    fiscal_period: str = Field(description="Period in YYYY-QN format or 'CURRENT'")
    gl_codes: list[str] = Field(default_factory=list)
    intent: Literal["journal_lookup", "compliance_check", "reconciliation", "general"]

Step 2: Configure Drift-Aware Chroma Collections

import chromadb
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from prometheus_client import Gauge, Histogram

# Observability metrics for drift detection
retrieval_score_gauge = Gauge('chroma_retrieval_median_score', 'Median similarity score', ['collection'])
centroid_distance_gauge = Gauge('chroma_centroid_distance', 'Embedding centroid shift', ['collection'])
schema_rejection_counter = Gauge('chroma_schema_rejections_total', 'Metadata validation failures', ['collection'])

client = chromadb.PersistentClient(path="./chroma_finance")

# Version-pinned collections prevent implicit taxonomy drift
gl_collection = client.get_or_create_collection(
    name="gl_entries_v2025Q3",
    metadata={
        "hnsw:M": 48,                  # Higher recall for precise GL lookups
        "hnsw:ef_construction": 250,
        "taxonomy_version": "2025-Q3", # Explicit version pinning
        "schema_registry": "gl_accounts_v4.2"
    }
)

standards_collection = client.get_or_create_collection(
    name="gaap_ifrs_v2025Q3",
    metadata={
        "hnsw:M": 32,
        "hnsw:ef_construction": 200,
        "standard_body": "FASB_IASB",
        "effective_date": "2025-07-01"
    }
)

gl_vectorstore = Chroma(client=client, collection_name="gl_entries_v2025Q3",
                        embedding_function=OpenAIEmbeddings(model="text-embedding-3-small"))
standards_vectorstore = Chroma(client=client, collection_name="gaap_ifrs_v2025Q3",
                               embedding_function=OpenAIEmbeddings(model="text-embedding-3-small"))

Step 3: Implement Drift Detection Middleware

import numpy as np
from datetime import datetime

class DriftMonitor:
    """Embedded drift detection for financial RAG"""
    
    def __init__(self, collection_name: str, baseline_centroid: np.ndarray):
        self.collection_name = collection_name
        self.baseline_centroid = baseline_centroid
        self.recent_embeddings: list[np.ndarray] = []
        
    def record_embedding(self, embedding: np.ndarray):
        self.recent_embeddings.append(embedding)
        if len(self.recent_embeddings) >= 100:  # Batch threshold
            self._check_drift()
            
    def _check_drift(self):
        current_centroid = np.mean(self.recent_embeddings, axis=0)
        distance = 1 - np.dot(current_centroid, self.baseline_centroid) / (
            np.linalg.norm(current_centroid) * np.linalg.norm(self.baseline_centroid)
        )
        centroid_distance_gauge.labels(collection=self.collection_name).set(distance)
        
        if distance > 0.08:
            print(f"  EMBEDDING DRIFT ALERT: {self.collection_name} centroid shift = {distance:.4f}")
            # Trigger re-indexing pipeline or alerting webhook here
            
        self.recent_embeddings.clear()

# Initialize monitors with baselines computed at last known-good index state
gl_drift_monitor = DriftMonitor("gl_entries_v2025Q3", baseline_centroid=np.load("./baselines/gl_q3_2025.npy"))
standards_drift_monitor = DriftMonitor("gaap_ifrs_v2025Q3", baseline_centroid=np.load("./baselines/gaap_q3_2025.npy"))

Step 4: Define Specialized Financial Agents

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

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

async def gl_retrieval_agent(state: FinancialState) -> dict:
    """Retrieves journal entries with strict temporal/entity scoping"""
    query = state["messages"][-1].content
    entity = state.get("entity_id", "ALL")
    period = state.get("fiscal_period", "CURRENT")
    
    results = gl_vectorstore.similarity_search_with_score(
        query=query, k=8,
        filter={"entity_id": entity, "fiscal_period": period}
    )
    
    # Record retrieval scores for drift monitoring
    scores = [s for _, s in results]
    if scores:
        retrieval_score_gauge.labels(collection="gl_entries_v2025Q3").set(np.median(scores))
    
    context = "\n".join([f"[{r.metadata['gl_code']}] {r.page_content}" for r, _ in results])
    audit_entry = {"agent": "gl_retrieval", "timestamp": datetime.utcnow().isoformat(),
                   "docs_retrieved": len(results), "median_score": np.median(scores) if scores else 0}
    
    return {"audit_trail": state.get("audit_trail", []) + [audit_entry]}

async def compliance_agent(state: FinancialState) -> dict:
    """Validates against GAAP/IFRS with cross-encoder re-ranking"""
    query = state["messages"][-1].content
    
    # Initial vector retrieval
    candidates = standards_vectorstore.similarity_search(query=query, k=10)
    
    # Cross-encoder re-ranking for precision (drift-resistant)
    # In production: use BAAI/bge-reranker-v2-m3 or Cohere rerank
    reranked = candidates[:5]  # Simplified; replace with actual cross-encoder
    
    refs = [f"{r.metadata['standard']} §{r.metadata['paragraph']}" for r in reranked]
    audit_entry = {"agent": "compliance", "timestamp": datetime.utcnow().isoformat(),
                   "standards_cited": refs}
    
    return {"compliance_refs": refs, 
            "audit_trail": state.get("audit_trail", []) + [audit_entry]}

async def reconciliation_agent(state: FinancialState) -> dict:
    """Cross-references sub-ledgers via API (not vector search)"""
    entity = state.get("entity_id")
    period = state.get("fiscal_period")
    
    # Simulated ERP reconciliation API call
    gaps = [
        {"account": "IC-2200", "amount": 14500.00, "counterparty": "Entity FR-01", "status": "UNMATCHED"},
        {"account": "AR-1100", "amount": 3200.50, "counterparty": "Customer X", "status": "AGING_90+"}
    ]
    
    audit_entry = {"agent": "reconciliation", "timestamp": datetime.utcnow().isoformat(),
                   "gaps_found": len(gaps)}
    return {"recon_gaps": gaps,
            "audit_trail": state.get("audit_trail", []) + [audit_entry]}

async def synthesis_agent(state: FinancialState) -> dict:
    """Generates response with mandatory audit trail citation"""
    system_prompt = f"""You are LedgerMind, a financial close assistant. 
    ENTITY: {state.get('entity_id')} | PERIOD: {state.get('fiscal_period')}
    COMPLIANCE REFS: {state.get('compliance_refs', [])}
    RECONCILIATION GAPS: {state.get('recon_gaps', [])}
    AUDIT TRAIL ENTRIES: {len(state.get('audit_trail', []))}
    
    RULES:
    1. Always cite specific GL codes and standard paragraphs.
    2. Flag reconciliation gaps BEFORE presenting journal entries.
    3. If compliance refs are empty, state "No applicable guidance found" explicitly.
    4. Never fabricate amounts or account numbers."""
    
    response = await llm.ainvoke([SystemMessage(content=system_prompt), *state["messages"]])
    
    final_audit = {"agent": "synthesis", "timestamp": datetime.utcnow().isoformat(),
                   "response_length": len(response.content)}
    return {"final_response": response.content,
            "messages": [response],
            "audit_trail": state.get("audit_trail", []) + [final_audit]}

Step 5: Build LangGraph Workflow with Temporal Routing & Memory

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver  # Persistent memory for audit compliancefrom langchain_core.messages import HumanMessage

def route_financial_query(state: FinancialState) -> Literal["gl_retrieval_agent", "compliance_agent", "reconciliation_agent"]:
    """Routes based on extracted intent AND temporal validity"""
    last_msg = state["messages"][-1].content.lower()
    
    # Temporal guard: reject queries outside valid periods
    period = state.get("fiscal_period")
    if period and period not in ["2025-Q3", "2025-Q2", "CURRENT"]:
        return "gl_retrieval_agent"  # Will return empty with proper messaging
    
    if any(w in last_msg for w in ["asc ", "ifrs ", "gaap", "recognition", "compliance"]):
        return "compliance_agent"
    elif any(w in last_msg for w in ["reconcil", "intercompany", "unmatched", "aging"]):
        return "reconciliation_agent"
    return "gl_retrieval_agent"

# Build Graph
workflow = StateGraph(FinancialState)
workflow.add_node("router", lambda s: s)
workflow.add_node("gl_retrieval_agent", gl_retrieval_agent)
workflow.add_node("compliance_agent", compliance_agent)
workflow.add_node("reconciliation_agent", reconciliation_agent)
workflow.add_node("synthesis_agent", synthesis_agent)

workflow.add_edge(START, "router")
workflow.add_conditional_edges("router", route_financial_query)
workflow.add_edge("gl_retrieval_agent", "synthesis_agent")
workflow.add_edge("compliance_agent", "synthesis_agent")
workflow.add_edge("reconciliation_agent", "synthesis_agent")
workflow.add_edge("synthesis_agent", END)

# Postgres-backed memory for audit-compliant session persistence
checkpointer = PostgresSaver.from_conn_string("postgresql://finance_rag:***@localhost:5432/ledgermind")
app = workflow.compile(checkpointer=checkpointer)

Step 6: Execute with Audit-Compliant Session

import asyncio

async def financial_close_session():
    config = {"configurable": {"thread_id": "close-2025Q3-DE03-session-8842"}}
    
    # Turn 1: Accrual lookup with entity/period context
    result = await app.ainvoke({
        "messages": [HumanMessage(content="Show accrual entries for DE-03 Q3 2025")],
        "entity_id": "DE-03",
        "fiscal_period": "2025-Q3"
    }, config=config)
    print(f"📊 Turn 1:\n{result['final_response']}\n")
    
    # Turn 2: Compliance check (memory retains entity/period)
    result = await app.ainvoke({
        "messages": [HumanMessage(content="Do these comply with ASC 606?")]
    }, config=config)
    print(f"📋 Turn 2:\n{result['final_response']}\n")
    
    # Turn 3: Reconciliation gap analysis
    result = await app.ainvoke({
        "messages": [HumanMessage(content="Any unreconciled intercompany balances?")]
    }, config=config)
    print(f" Turn 3:\n{result['final_response']}\n")
    
    # Verify audit trail completeness
    print(f" Audit Trail Entries: {len(result['audit_trail'])}")
    for entry in result['audit_trail']:
        print(f"  [{entry['timestamp']}] {entry['agent']}: {entry}")

asyncio.run(financial_close_session())

Key Enterprise Takeaways for Finance

Decision	Why It Matters for Accounting
Version-Pinned Collections	Prevents GAAP updates from corrupting prior-period audit evidence; enables point-in-time reproducibility
Metadata Pre-filtering on Entity/Period	Eliminates cross-entity contamination; ensures SOX-compliant data isolation
Cross-Encoder Re-ranking	Compensates for embedding drift in standards docs where semantic similarity ≠ regulatory applicability
Postgres-Backed Memory	Provides tamper-evident session history for auditor review; survives process restarts during multi-day closes
Prometheus Drift Metrics	Enables SLO-based alerting before retrieval quality impacts financial reporting deadlines
Immutable Audit Trail in State	Every agent action is timestamped and logged; satisfies regulatory requirements for AI decision explainability
Temporal Routing Guards	Prevents agents from answering out-of-period queries with stale data; enforces fiscal calendar awareness

This architecture delivers audit-grade RAG for financial close processes while actively detecting and mitigating the two forms of drift that silently undermine financial AI systems. The combination of version-pinned storage, continuous observability, and stateful orchestration ensures the system remains reliable through quarterly standard updates, M&A-driven chart of accounts changes, and the relentless pace of financial reporting cycles.