Why "Looks Good" Is Not a Success Criterion

In digital banking loan management and customer support, multi-step reasoning chains answer questions like "Can I refinance my auto loan at a lower rate given my recent credit score improvement?" or "Why was my hardship deferment application denied last week?" These require sequential retrieval (loan terms → payment history → credit profile → policy rules), intermediate computations (DTI recalculation, eligibility scoring), and regulatory validation.

Most teams define success implicitly: the agent stops when it generates a response. This is catastrophic in lending. A response that sounds confident but cites an outdated interest rate, miscalculates DTI by omitting a co-borrower’s debt, or misses a mandatory adverse action disclosure isn’t a success it’s a fair lending violation. Convergence must be explicitly defined as a structured, multi-dimensional predicate evaluated against live state. The graph doesn’t terminate when generation completes; it terminates when all success criteria are satisfied or when bounded exhaustion triggers escalation. This article demonstrates implementing formal convergence definitions for loan management RAG, with code-level enforcement in a stateful LangGraph multi-agent system.

Real-Time Use Case: NovaBank Loan Management & Support Assistant

The Workflow

NovaBank’s AI assistant handles 150K+ monthly interactions across:

Loan Management: Refinancing eligibility, payment schedule changes, hardship programs, payoff quotes, escrow analysis, rate lock inquiries, cosigner release requests.

Customer Support: Application status, document requests, denial explanations, fee disputes, regulatory disclosures, complaint acknowledgment.

Each query triggers a multi-step chain: retrieve loan contract → fetch live account state → pull credit bureau data → apply policy rules → validate regulatory compliance → generate response. Chains range from 3 steps (simple balance inquiry) to 8+ steps (refinance eligibility with DTI recalculation).

Why Implicit Termination Fails in Lending

Failure ModeImplicit Termination BehaviorExplicit Convergence Fix
Stale rate citedAgent generates response with cached rate; chain endsFreshness gate blocks termination until live rate confirmed
DTI miscalculationAgent omits co-borrower debt; response generatedComputational verification node validates DTI before allowing completion
Missing adverse action noticeAgent explains denial without FCRA-mandated languageRegulatory checklist gate requires explicit disclosure confirmation
Circular policy lookupAgent re-fetches same policy doc 6 timesState-change delta check detects zero-progress loops
Partial eligibility answerAgent says "you may qualify" without specific conditionsCompleteness predicate requires enumerated conditions + next steps
Unverified computationAgent calculates monthly savings without showing formulaAudit trail gate requires computation trace in state
398

The Convergence Framework: Five Success Dimensions

Convergence is not a boolean. It is a structured vector of five dimensions, each with typed pass/fail criteria evaluated against graph state. All five must pass for CONVERGED; any failure triggers targeted repair or bounded escalation.

DimensionDefinitionState Fields CheckedFailure Action
Factual GroundingEvery numeric/contractual claim traces to verified stateverified_facts, source_citationsRe-retrieve + regenerate
Computational IntegrityAll derived values have auditable computation tracescomputation_trace, intermediate_valuesRecompute with validation
Regulatory CompletenessAll mandatory disclosures present and version-correctdisclosure_checklist, policy_versionInject missing disclosures
State FreshnessAll live data within TTL for decision-sensitive fieldsdata_timestamps, freshness_thresholdsRe-fetch stale sources
Actionable CompletenessResponse includes specific next steps, not vague guidanceaction_items, eligibility_conditionsRegenerate with specificity

Step 1: State Schema with Convergence-First Design

Every convergence dimension maps to typed state fields. The schema makes success criteria machine-evaluable.

from typing import Annotated, List, Dict, Any, Optional, Literal
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
from datetime import datetime
import operator

class VerifiedFact(TypedDict):
    """A factual claim traced to a specific source in state."""
    claim: str
    value: Any
    source_field: str  # e.g., "loan_contract.interest_rate", "credit_report.score"
    source_timestamp: datetime
    verified: bool

class ComputationStep(TypedDict):
    """One step in a derived-value calculation with full audit trail."""
    operation: str  # e.g., "DTI = total_debt / gross_income"
    inputs: Dict[str, Any]
    output: Any
    formula_reference: str  # Policy section or formula ID
    timestamp: datetime

class DisclosureCheck(TypedDict):
    """Status of a mandatory regulatory disclosure."""
    disclosure_id: str  # e.g., "FCRA_adverse_action", "TILA_rate_change"
    required: bool
    present_in_response: bool
    version_match: bool
    policy_section: str

class ConvergenceResult(TypedDict):
    """Structured evaluation of all five convergence dimensions."""
    factual_grounding: bool
    computational_integrity: bool
    regulatory_completeness: bool
    state_freshness: bool
    actionable_completeness: bool
    overall_converged: bool
    failed_dimensions: List[str]
    repair_actions: List[Dict[str, Any]]
    eval_timestamp: datetime


class LoanAssistantState(TypedDict):
    # === CONVERSATION STATE ===
    messages: Annotated[list, add_messages]
    customer_id: str
    session_id: str
    current_module: Literal["loan_management", "customer_support"]
    
    # === RETRIEVED CONTEXT ===
    loan_contract: Optional[Dict[str, Any]]
    account_state: Optional[Dict[str, Any]]
    credit_profile: Optional[Dict[str, Any]]
    policy_rules: Optional[Dict[str, Any]]
    
    # === CONVERGENCE STATE (typed, machine-evaluable) ===
    verified_facts: Annotated[List[VerifiedFact], operator.add]
    computation_trace: Annotated[List[ComputationStep], operator.add]
    disclosure_checklist: List[DisclosureCheck]
    data_timestamps: Dict[str, datetime]
    freshness_thresholds: Dict[str, int]  # seconds
    
    # === GENERATION OUTPUT ===
    generated_response: Optional[str]
    action_items: List[str]
    eligibility_conditions: List[str]
    
    # === CONVERGENCE EVALUATION ===
    convergence_result: Optional[ConvergenceResult]
    convergence_cycle: int
    max_convergence_cycles: int
    
    # === REPAIR STATE ===
    repair_history: Annotated[List[Dict[str, Any]], operator.add]
    pending_repairs: List[Dict[str, Any]]
    
    # === FINAL OUTPUT ===
    final_response: Optional[str]
    resolution_status: Literal["converged", "repaired", "escalated", "exhausted"]
    
    # === AUDIT ===
    audit_trail: Annotated[List[Dict[str, Any]], operator.add]
    processing_start_time: datetime

🔑 Key Insight: verified_facts, computation_trace, and disclosure_checklist are not documentation—they are convergence predicates. The convergence evaluator reads these fields and returns structured pass/fail per dimension. If they’re empty or incomplete, convergence fails by definition.

Step 2: Fact Verification Node — Grounding Dimension

async def fact_verification_node(state: LoanAssistantState) -> dict:
    """
    Extracts every factual claim from generated response and verifies
    against retrieved state. Populates verified_facts for convergence eval.
    """
    response = state.get("generated_response", "")
    loan = state.get("loan_contract", {})
    account = state.get("account_state", {})
    credit = state.get("credit_profile", {})
    
    verified_facts = []
    
    # Extract and verify interest rate claims
    import re
    rate_matches = re.findall(r'(\d+\.\d+)%\s*(?:APR|interest rate|rate)', response, re.IGNORECASE)
    for rate_str in rate_matches:
        claimed_rate = float(rate_str)
        actual_rate = loan.get("current_apr")
        verified_facts.append(VerifiedFact(
            claim=f"Interest rate {rate_str}%",
            value=claimed_rate,
            source_field="loan_contract.current_apr",
            source_timestamp=state.get("data_timestamps", {}).get("loan_contract", datetime.min),
            verified=abs(claimed_rate - actual_rate) < 0.001 if actual_rate else False
        ))
    
    # Verify balance/payment claims
    balance_matches = re.findall(r'\$[\d,]+\.?\d*', response)
    actual_balance = account.get("outstanding_principal")
    for amount_str in balance_matches:
        claimed = float(amount_str.replace("$", "").replace(",", ""))
        if actual_balance and abs(claimed - actual_balance) < 0.01:
            verified_facts.append(VerifiedFact(
                claim=f"Balance {amount_str}",
                value=claimed,
                source_field="account_state.outstanding_principal",
                source_timestamp=state.get("data_timestamps", {}).get("account_state", datetime.min),
                verified=True
            ))
    
    # Verify credit score claims
    score_matches = re.findall(r'(?:credit\s+score|FICO)\s*(?:of\s*)?(\d{3})', response, re.IGNORECASE)
    actual_score = credit.get("fico_score")
    for score_str in score_matches:
        claimed_score = int(score_str)
        verified_facts.append(VerifiedFact(
            claim=f"Credit score {score_str}",
            value=claimed_score,
            source_field="credit_profile.fico_score",
            source_timestamp=state.get("data_timestamps", {}).get("credit_profile", datetime.min),
            verified=claimed_score == actual_score if actual_score else False
        ))
    
    return {
        "verified_facts": verified_facts,
        "audit_trail": [{
            "node": "fact_verification",
            "facts_checked": len(verified_facts),
            "unverified_count": sum(1 for f in verified_facts if not f["verified"]),
            "timestamp": datetime.utcnow().isoformat()
        }]
    }

Step 3: Computation Validation Node — Integrity Dimension

async def computation_validation_node(state: LoanAssistantState) -> dict:
    """
    Validates all derived computations (DTI, savings, eligibility scores)
    have complete audit trails. Recomputes if trace is missing or inconsistent.
    """
    response = state.get("generated_response", "")
    existing_trace = state.get("computation_trace", [])
    account = state.get("account_state", {})
    credit = state.get("credit_profile", {})
    
    new_steps = []
    
    # DTI computation validation
    if "dti" in response.lower() or "debt-to-income" in response.lower():
        # Extract claimed DTI
        dti_match = re.search(r'(?:DTI|debt.to.income)\s*(?:of\s*)?(\d+\.?\d*)%?', response, re.IGNORECASE)
        if dti_match:
            claimed_dti = float(dti_match.group(1))
            
            # Recompute from source data
            total_monthly_debt = account.get("total_monthly_obligations", 0)
            gross_monthly_income = credit.get("verified_monthly_income", 0)
            
            if gross_monthly_income > 0:
                computed_dti = (total_monthly_debt / gross_monthly_income) * 100
                
                new_steps.append(ComputationStep(
                    operation="DTI = total_monthly_debt / gross_monthly_income * 100",
                    inputs={
                        "total_monthly_debt": total_monthly_debt,
                        "gross_monthly_income": gross_monthly_income
                    },
                    output=round(computed_dti, 2),
                    formula_reference="POLICY-LOAN-003 §4.2",
                    timestamp=datetime.utcnow()
                ))
                
                # Flag mismatch for convergence evaluator
                if abs(claimed_dti - computed_dti) > 1.0:
                    new_steps[-1]["output"] = f"MISMATCH: claimed={claimed_dti}, computed={computed_dti}"
    
    # Monthly savings computation for refinance scenarios
    if "sav" in response.lower() and ("refinance" in response.lower() or "lower rate" in response.lower()):
        current_payment = account.get("current_monthly_payment", 0)
        new_rate = state.get("loan_contract", {}).get("refinance_offer_rate")
        remaining_term = account.get("remaining_term_months", 0)
        
        if new_rate and remaining_term > 0:
            # Simplified payment calc for validation
            new_payment = compute_monthly_payment(
                account.get("outstanding_principal", 0),
                new_rate / 100 / 12,
                remaining_term
            )
            savings = current_payment - new_payment
            
            new_steps.append(ComputationStep(
                operation="Monthly savings = current_payment - new_payment",
                inputs={
                    "current_payment": current_payment,
                    "new_rate": new_rate,
                    "remaining_term": remaining_term,
                    "computed_new_payment": round(new_payment, 2)
                },
                output=round(savings, 2),
                formula_reference="POLICY-REFI-007 §2.1",
                timestamp=datetime.utcnow()
            ))
    
    return {
        "computation_trace": new_steps,
        "audit_trail": [{
            "node": "computation_validation",
            "steps_validated": len(new_steps),
            "timestamp": datetime.utcnow().isoformat()
        }]
    }

Step 4: Regulatory Checklist Node — Completeness Dimension

# Mandatory disclosure registry per module/query type
DISCLOSURE_REGISTRY = {
    ("loan_management", "denial_explanation"): [
        {"id": "FCRA_adverse_action", "section": "FCRA §615(a)", "keyword": "adverse action"},
        {"id": "ECOA_notice", "section": "Reg B §1002.9", "keyword": "equal credit opportunity"},
        {"id": "credit_bureau_disclosure", "section": "FCRA §615(b)", "keyword": "consumer reporting agency"},
    ],
    ("loan_management", "rate_change"): [
        {"id": "TILA_rate_change", "section": "Reg Z §1026.20", "keyword": "annual percentage rate"},
        {"id": "payment_change_notice", "section": "Reg Z §1026.20(c)", "keyword": "payment change"},
    ],
    ("loan_management", "refinance_eligibility"): [
        {"id": "refinance_risk_disclosure", "section": "POLICY-REFI-007 §5.0", "keyword": "closing costs"},
        {"id": "rate_lock_terms", "section": "POLICY-RATE-002 §3.1", "keyword": "rate lock"},
    ],
    ("customer_support", "fee_dispute"): [
        {"id": "fee_schedule_reference", "section": "POLICY-FEE-001 §2.0", "keyword": "fee schedule"},
        {"id": "dispute_timeline", "section": "Reg E §1005.11", "keyword": "investigation period"},
    ]
}

async def regulatory_checklist_node(state: LoanAssistantState) -> dict:
    """
    Builds and evaluates mandatory disclosure checklist.
    Populates disclosure_checklist for convergence evaluation.
    """
    module = state["current_module"]
    response = state.get("generated_response", "").lower()
    query_type = infer_query_type(state["messages"])
    
    key = (module, query_type)
    required_disclosures = DISCLOSURE_REGISTRY.get(key, [])
    
    checklist = []
    for disc in required_disclosures:
        present = disc["keyword"] in response
        
        # Version check: ensure cited policy version matches current
        version_match = True
        if present:
            cited_version = extract_policy_version(response, disc["section"])
            current_version = state.get("policy_rules", {}).get("version")
            if cited_version and current_version and cited_version != current_version:
                version_match = False
        
        checklist.append(DisclosureCheck(
            disclosure_id=disc["id"],
            required=True,
            present_in_response=present,
            version_match=version_match,
            policy_section=disc["section"]
        ))
    
    return {
        "disclosure_checklist": checklist,
        "audit_trail": [{
            "node": "regulatory_checklist",
            "required_count": len(checklist),
            "present_count": sum(1 for d in checklist if d["present_in_response"]),
            "missing": [d["disclosure_id"] for d in checklist if not d["present_in_response"]],
            "timestamp": datetime.utcnow().isoformat()
        }]
    }

Step 5: Convergence Evaluator — The Formal Success Predicate

This is the core: a deterministic function that evaluates all five dimensions against state and produces structured repair actions on failure.

class ConvergenceEvaluator:
    """
    Evaluates all five convergence dimensions against current state.
    Returns structured ConvergenceResult with targeted repair actions.
    THIS IS THE FORMAL DEFINITION OF SUCCESS.
    """
    
    @staticmethod
    def evaluate(state: LoanAssistantState) -> ConvergenceResult:
        failed_dims = []
        repairs = []
        
        # DIMENSION 1: Factual Grounding
        facts = state.get("verified_facts", [])
        unverified = [f for f in facts if not f["verified"]]
        factual_ok = len(facts) > 0 and len(unverified) == 0
        
        if not factual_ok:
            failed_dims.append("factual_grounding")
            if not facts:
                repairs.append({"type": "re_extract_facts", "reason": "No facts extracted from response"})
            else:
                repairs.append({
                    "type": "re_retrieve_sources",
                    "unverified_fields": [f["source_field"] for f in unverified],
                    "reason": f"{len(unverified)} unverified factual claims"
                })
        
        # DIMENSION 2: Computational Integrity
        comp_trace = state.get("computation_trace", [])
        has_mismatches = any("MISMATCH" in str(s.get("output", "")) for s in comp_trace)
        response_has_computation = any(
            kw in (state.get("generated_response", "") or "").lower()
            for kw in ["dti", "debt-to-income", "monthly payment", "savings", "eligibility score"]
        )
        computational_ok = not response_has_computation or (len(comp_trace) > 0 and not has_mismatches)
        
        if not computational_ok:
            failed_dims.append("computational_integrity")
            repairs.append({
                "type": "recompute_with_validation",
                "reason": "Missing or mismatched computation trace"
            })
        
        # DIMENSION 3: Regulatory Completeness
        checklist = state.get("disclosure_checklist", [])
        missing = [d for d in checklist if d["required"] and not d["present_in_response"]]
        version_mismatch = [d for d in checklist if d["required"] and d["present_in_response"] and not d["version_match"]]
        regulatory_ok = len(missing) == 0 and len(version_mismatch) == 0
        
        if not regulatory_ok:
            failed_dims.append("regulatory_completeness")
            if missing:
                repairs.append({
                    "type": "inject_missing_disclosures",
                    "disclosure_ids": [d["disclosure_id"] for d in missing],
                    "reason": f"{len(missing)} mandatory disclosures missing"
                })
            if version_mismatch:
                repairs.append({
                    "type": "update_disclosure_versions",
                    "disclosure_ids": [d["disclosure_id"] for d in version_mismatch],
                    "reason": f"{len(version_mismatch)} disclosures cite outdated policy version"
                })
        
        # DIMENSION 4: State Freshness
        timestamps = state.get("data_timestamps", {})
        thresholds = state.get("freshness_thresholds", {
            "account_state": 60,      # 60s for balances
            "credit_profile": 300,    # 5min for credit
            "loan_contract": 3600,    # 1hr for contract terms
            "policy_rules": 86400     # 24hr for policies
        })
        
        stale_sources = []
        for source, ts in timestamps.items():
            threshold = thresholds.get(source, 300)
            age = (datetime.utcnow() - ts).total_seconds()
            if age > threshold:
                stale_sources.append({"source": source, "age_s": age, "threshold_s": threshold})
        
        freshness_ok = len(stale_sources) == 0
        
        if not freshness_ok:
            failed_dims.append("state_freshness")
            repairs.append({
                "type": "refetch_stale_sources",
                "sources": [s["source"] for s in stale_sources],
                "reason": f"{len(stale_sources)} data sources exceed freshness threshold"
            })
        
        # DIMENSION 5: Actionable Completeness
        response = state.get("generated_response", "") or ""
        action_items = state.get("action_items", [])
        conditions = state.get("eligibility_conditions", [])
        
        # Check for vague/non-actionable language
        vague_patterns = ["may qualify", "might be eligible", "could potentially", "we suggest considering"]
        has_vague = any(p in response.lower() for p in vague_patterns)
        has_specific_next_steps = len(action_items) > 0 or ("next step" in response.lower() and len(response) > 100)
        
        actionable_ok = not has_vague and has_specific_next_steps
        
        if not actionable_ok:
            failed_dims.append("actionable_completeness")
            repairs.append({
                "type": "regenerate_with_specificity",
                "reason": "Response lacks specific next steps or uses non-committal language"
            })
        
        overall = len(failed_dims) == 0
        
        return ConvergenceResult(
            factual_grounding=factual_ok,
            computational_integrity=computational_ok,
            regulatory_completeness=regulatory_ok,
            state_freshness=freshness_ok,
            actionable_completeness=actionable_ok,
            overall_converged=overall,
            failed_dimensions=failed_dims,
            repair_actions=repairs,
            eval_timestamp=datetime.utcnow()
        )
async def convergence_evaluator_node(state: LoanAssistantState) -> dict:
    """Runs convergence evaluation and updates state with structured result."""
    result = ConvergenceEvaluator.evaluate(state)
    cycle = state.get("convergence_cycle", 0)
    
    return {
        "convergence_result": result,
        "convergence_cycle": cycle + 1,
        "audit_trail": [{
            "node": "convergence_evaluator",
            "cycle": cycle,
            "converged": result["overall_converged"],
            "failed_dimensions": result["failed_dimensions"],
            "repair_count": len(result["repair_actions"]),
            "timestamp": datetime.utcnow().isoformat()
        }]
    }

Step 6: Repair Router + Execution Nodes

def route_after_convergence(state: LoanAssistantState) -> str:
    """Deterministic routing based on convergence result."""
    result = state.get("convergence_result")
    cycle = state.get("convergence_cycle", 0)
    max_cycles = state.get("max_convergence_cycles", 3)
    
    if result and result["overall_converged"]:
        return "finalize_converged"
    
    if cycle >= max_cycles:
        return "finalize_exhausted"
    
    # Route to specific repair based on highest-priority failed dimension
    failed = result["failed_dimensions"] if result else []
    
    if "state_freshness" in failed:
        return "repair_freshness"       # Always fix staleness first
    if "regulatory_completeness" in failed:
        return "repair_regulatory"      # Regulatory before content
    if "factual_grounding" in failed:
        return "repair_facts"
    if "computational_integrity" in failed:
        return "repair_computation"
    if "actionable_completeness" in failed:
        return "repair_actionability"
    
    return "finalize_exhausted"  # Safety fallback


async def repair_freshness_node(state: LoanAssistantState) -> dict:
    """Re-fetches stale data sources."""
    repairs = state.get("convergence_result", {}).get("repair_actions", [])
    stale_sources = next((r["sources"] for r in repairs if r["type"] == "refetch_stale_sources"), [])
    
    new_timestamps = dict(state.get("data_timestamps", {}))
    updates = {}
    
    if "account_state" in stale_sources:
        fresh = await fetch_account_state(state["customer_id"])
        updates["account_state"] = fresh
        new_timestamps["account_state"] = datetime.utcnow()
    
    if "credit_profile" in stale_sources:
        fresh = await fetch_credit_profile(state["customer_id"])
        updates["credit_profile"] = fresh
        new_timestamps["credit_profile"] = datetime.utcnow()
    
    if "loan_contract" in stale_sources:
        fresh = await fetch_loan_contract(state["customer_id"])
        updates["loan_contract"] = fresh
        new_timestamps["loan_contract"] = datetime.utcnow()
    
    return {
        **updates,
        "data_timestamps": new_timestamps,
        "repair_history": [{"type": "refetch_stale_sources", "sources": stale_sources, "timestamp": datetime.utcnow()}],
        "pending_repairs": []
    }
async def repair_regulatory_node(state: LoanAssistantState) -> dict:
    """Injects missing disclosures into response context for regeneration."""
    checklist = state.get("disclosure_checklist", [])
    missing = [d for d in checklist if d["required"] and not d["present_in_response"]]
    
    # Fetch disclosure templates
    templates = await fetch_disclosure_templates([d["disclosure_id"] for d in missing])
    
    return {
        "pending_repairs": [{
            "type": "inject_disclosures",
            "templates": templates,
            "timestamp": datetime.utcnow()
        }],
        "repair_history": [{
            "type": "inject_missing_disclosures",
            "count": len(missing),
            "ids": [d["disclosure_id"] for d in missing],
            "timestamp": datetime.utcnow()
        }]
    }

Step 7: Assemble the Convergence-Aware Graph

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver

workflow = StateGraph(LoanAssistantState)

# Core nodes
workflow.add_node("retrieve", retrieval_node)
workflow.add_node("generate", generation_node)
workflow.add_node("verify_facts", fact_verification_node)
workflow.add_node("validate_computation", computation_validation_node)
workflow.add_node("check_regulatory", regulatory_checklist_node)
workflow.add_node("evaluate_convergence", convergence_evaluator_node)

# Repair nodes
workflow.add_node("repair_freshness", repair_freshness_node)
workflow.add_node("repair_regulatory", repair_regulatory_node)
workflow.add_node("repair_facts", repair_facts_node)
workflow.add_node("repair_computation", repair_computation_node)
workflow.add_node("repair_actionability", repair_actionability_node)

# Terminal nodes
workflow.add_node("finalize_converged", lambda s: {
    "final_response": s["generated_response"],
    "resolution_status": "converged",
    "audit_trail": [{"event": "converged", "cycles": s["convergence_cycle"], "timestamp": datetime.utcnow().isoformat()}]
})

workflow.add_node("finalize_exhausted", lambda s: {
    "final_response": "I need additional verification to provide an accurate answer. Connecting you with a loan specialist.",
    "resolution_status": "escalated",
    "audit_trail": [{"event": "convergence_exhausted", "cycles": s["convergence_cycle"], 
                      "failed_dims": s.get("convergence_result", {}).get("failed_dimensions", []),
                      "timestamp": datetime.utcnow().isoformat()}]
})

# Edges
workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", "verify_facts")
workflow.add_edge("verify_facts", "validate_computation")
workflow.add_edge("validate_computation", "check_regulatory")
workflow.add_edge("check_regulatory", "evaluate_convergence")

# Convergence routing
workflow.add_conditional_edges("evaluate_convergence", route_after_convergence, {
    "finalize_converged": "finalize_converged",
    "finalize_exhausted": "finalize_exhausted",
    "repair_freshness": "repair_freshness",
    "repair_regulatory": "repair_regulatory",
    "repair_facts": "repair_facts",
    "repair_computation": "repair_computation",
    "repair_actionability": "repair_actionability"
})

# All repair nodes → regenerate → re-evaluate
for repair_node in ["repair_freshness", "repair_regulatory", "repair_facts", 
                     "repair_computation", "repair_actionability"]:
    workflow.add_edge(repair_node, "generate")

workflow.add_edge("finalize_converged", END)
workflow.add_edge("finalize_exhausted", END)

checkpointer = PostgresSaver.from_conn_string("postgresql://novabank-loan-db")
app = workflow.compile(checkpointer=checkpointer)

Convergence Behavior Matrix

ScenarioCycle 1Cycle 2Cycle 3Resolution
Clean response✅ All 5 dims pass——converged
Stale account balance❌ Freshness → refetch✅ All pass—converged
Missing FCRA disclosure❌ Regulatory → inject✅ All pass—converged
DTI mismatch + stale credit❌ Freshness → refetch❌ Computation → recompute✅ All passrepaired
Persistent vague language❌ Actionable → regenerate❌ Still vague❌ Max cyclesescalated
API down for credit pull❌ Freshness → refetch fails❌ Retry fails❌ Max cyclesescalated

Key Design Principles

1. Convergence Is a Structured Vector, Not a Scalar

Five independent dimensions, each with typed pass/fail criteria. A response can be factually grounded but regulatorily incomplete. Treating convergence as a single score hides critical failures.

2. Success Criteria Are Evaluated Against State, Not Prompts

The evaluator reads verified_facts, computation_trace, disclosure_checklist—not the raw response text. This makes evaluation deterministic and auditable. Two identical responses with different underlying state produce different convergence results.

3. Repair Actions Are Dimension-Specific

Freshness failures trigger refetch. Regulatory failures trigger disclosure injection. Factual failures trigger re-retrieval. Generic "regenerate" wastes cycles on problems that don't need regeneration.

4. Repair Priority Is Deterministic

Freshness > Regulatory > Factual > Computational > Actionable. Stale data invalidates everything downstream. Regulatory gaps are compliance violations regardless of factual accuracy. This ordering is encoded in route_after_convergence, not learned.

5. Exhaustion Is a Valid Terminal State

max_convergence_cycles = 3 is a hard ceiling. When reached, the system escalates with full diagnostic context (failed_dimensions, repair_history). Escalation is not failure—it's correct behavior for genuinely ambiguous cases.

Defining convergence in multi-step banking RAG means replacing implicit termination ("response generated") with explicit, multi-dimensional success predicates evaluated against live state. The five dimensions—factual grounding, computational integrity, regulatory completeness, state freshness, and actionable completeness—form a formal contract between the system and its regulators.

This contract is enforced not by prompts or post-hoc audits, but by typed state fields, deterministic evaluators, and targeted repair routers embedded in the graph topology. Convergence is not something you hope for; it is something you prove, per interaction, with full audit trail.

In lending, the cost of undefined convergence isn't a bad metric. It's a consent order, a class action, or a customer denied credit based on a miscalculated DTI. Define your success criteria formally. Encode them in state. Evaluate them deterministically. In regulated finance, convergence isn't an engineering concern it's a legal requirement.