Self-Correcting Payment RAG for Digital Banking

Why Evaluation Must Be Inside the Graph, Not After It

In digital banking payments and transfers, "good enough" is a regulatory violation. When an AI assistant answers "What's my daily transfer limit?" or "Why was my international wire declined?", the response must be factually correct, compliant with current policy, and consistent with the customer’s actual account state. Traditional RAG evaluation runs after generation as a batch metric. In payments, that’s too late—a wrong answer has already been delivered to a customer who may act on it. Stateful evaluation loops solve this by embedding validation inside the LangGraph execution cycle. The graph doesn’t just generate and return; it generates, evaluates against live state, identifies specific failures, repairs them, and re-evaluates—all within a single request, with full memory of what failed and why. This article demonstrates implementing stateful evaluation loops for a digital banking payments module, where every response is validated against account state, transfer policies, and regulatory requirements before reaching the customer.

Real-Time Use Case: Payments & Transfers Intelligence Assistant

The Scenario

A digital bank’s customers and support agents query the payments assistant 200K+ times daily:

• "Can I send $15,000 to my contractor in Germany today?"

• "Why was my transfer to ACME Corp declined yesterday?"

• "What are the fees for instant vs. standard ACH transfers?"

• "Increase my monthly outbound wire limit."

Each query touches three distinct knowledge domains that must be cross-validated:

1. Account State: Real-time balances, holds, limits, KYC tier (live core banking API)

2. Transfer Policy: Fee schedules, velocity limits, geographic restrictions (policy engine)

3. Regulatory Rules: OFAC screening, SAR triggers, cross-border reporting thresholds (compliance DB)

A response that correctly cites policy but ignores the customer’s actual KYC tier is wrong. A response that checks account state but misses a new OFAC sanction is dangerous. Evaluation must validate all three dimensions simultaneously and trigger targeted repair when any dimension fails.

Why Stateless Evaluation Fails Here

Failure Mode	Stateless Eval Limitation	Stateful Loop Solution
Policy + account mismatch	Checks facts independently; misses cross-domain inconsistency	Joint validator reads both policy AND account state from shared graph state
Stale account data	Eval uses cached snapshot; doesn’t detect freshness drift	Freshness-aware validator triggers live re-fetch before re-evaluation
Vague repair signals	Returns pass/fail score; agent guesses what to fix	Structured failure diagnosis with specific field-level repair instructions
Unbounded retry cost	Retries entire pipeline on any failure	Targeted repair nodes fix only the failed dimension
No learning across turns	Each eval starts fresh; repeats same mistakes	Persistent eval memory accumulates failure patterns per session

---

Step 1: State Schema with Evaluation-First Design

The state schema encodes evaluation requirements as first-class fields, not afterthoughts.

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 EvalDimensionResult(TypedDict):
    """Result for a single evaluation dimension."""
    dimension: Literal["factual_accuracy", "policy_compliance", "account_consistency", 
                        "regulatory_adherence", "freshness"]
    passed: bool
    score: float  # 0.0 - 1.0
    failures: List[Dict[str, Any]]  # Specific, actionable failure descriptions
    repair_instructions: Optional[str]

class RepairAction(TypedDict):
    """Targeted repair instruction from diagnosis."""
    target_dimension: str
    action_type: Literal["refetch_account", "update_policy_context", 
                          "add_regulatory_flag", "clarify_query", "cite_source"]
    parameters: Dict[str, Any]
    priority: int  # Lower = higher priority

class PaymentsState(TypedDict):
    # Conversation history
    messages: Annotated[list, add_messages]
    
    # === KNOWLEDGE CONTEXT (populated by retrieval) ===
    account_state: Dict[str, Any]  
    # {"available_balance": float, "daily_limit_used": float, "kyc_tier": str, 
    #  "holds": list, "last_updated": datetime}
    
    policy_context: Dict[str, Any]
    # {"transfer_types": dict, "fee_schedule": dict, "velocity_limits": dict, 
    #  "geo_restrictions": list, "policy_version": str}
    
    regulatory_flags: List[Dict[str, Any]]
    # [{"type": "ofac_match", "entity": ..., "action_required": ...}]
    
    # === GENERATION OUTPUT ===
    generated_response: Optional[str]
    response_sources: List[str]  # Citations for traceability
    
    # === EVALUATION STATE ===
    eval_results: List[EvalDimensionResult]
    eval_cycle: int  # Current evaluation iteration
    max_eval_cycles: int  # Hard ceiling
    
    # === REPAIR STATE ===
    repair_history: Annotated[List[RepairAction], operator.add]
    pending_repairs: List[RepairAction]
    
    # === EVAL MEMORY (persists across cycles AND sessions) ===
    eval_memory: Dict[str, Any]
    # {"failure_patterns": list, "successful_repairs": list, 
    #  "known_ambiguous_queries": list}
    
    # === FINAL OUTPUT ===
    final_response: Optional[str]
    response_quality: Literal["verified", "repaired", "partial", "escalated"]
    eval_summary: Optional[str]
    
    # Session metadata
    session_id: str
    customer_id: str
    processing_start_time: datetime

🔑 Key Design Principle: eval_results, repair_history, and eval_memory are not logging fields—they are control flow inputs. The evaluation node reads them to decide whether to pass, repair, or escalate. The repair node reads them to avoid repeating failed fixes. This is what makes the loop stateful rather than merely iterative.

Step 2: Retrieval Node — Populate All Knowledge Dimensions

async def payments_retrieval_node(state: PaymentsState) -> dict:
    """
    Fetches account state, policy, and regulatory context in parallel.
    Tags each with freshness metadata for downstream evaluation.
    """
    customer_id = state["customer_id"]
    
    # Parallel fetch from three sources
    import asyncio
    
    account_task = fetch_account_state(customer_id)
    policy_task = fetch_transfer_policies()
    regulatory_task = check_regulatory_flags(customer_id)
    
    account, policy, regs = await asyncio.gather(
        account_task, policy_task, regulatory_task
    )
    
    return {
        "account_state": {
            **account,
            "last_updated": datetime.utcnow(),
            "source": "core_banking_api"
        },
        "policy_context": {
            **policy,
            "last_updated": datetime.utcnow(),
            "source": "policy_engine_v3"
        },
        "regulatory_flags": regs,
        "eval_cycle": 0,
        "max_eval_cycles": 3,  # Hard ceiling: never evaluate more than 3 times
        "eval_results": [],
        "repair_history": [],
        "pending_repairs": []
    }

Step 3: Generation Node — Draft with Full Context Awareness

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

generation_prompt = ChatPromptTemplate.from_messages([
    ("system", """You are NovaBank's payments intelligence assistant.
    
ACCOUNT STATE:
{account_state}

TRANSFER POLICY (v{policy_version}):
{policy_context}

REGULATORY FLAGS:
{regulatory_flags}

REPAIR HISTORY (if any):
{repair_history}

EVAL MEMORY (prior failures in this session):
{eval_memory}

RULES:
1. ALWAYS cite specific policy sections and account values.
2. If account state conflicts with policy, explain BOTH clearly.
3. If regulatory flags exist, address them BEFORE answering the query.
4. If repair history exists, explicitly address previously failed dimensions.
5. Never guess limits, fees, or eligibility—cite exact values from context."""),
    ("human", "{query}")
])

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

async def generation_node(state: PaymentsState) -> dict:
    """Generates response using all available context + repair history."""
    query = state["messages"][-1].content if state["messages"] else ""
    
    response = await generation_prompt | llm | (lambda x: x.content)
    result = await response.ainvoke({
        "query": query,
        "account_state": json.dumps(state["account_state"], default=str),
        "policy_context": json.dumps(state["policy_context"], default=str),
        "policy_version": state["policy_context"].get("policy_version", "unknown"),
        "regulatory_flags": json.dumps(state["regulatory_flags"], default=str),
        "repair_history": json.dumps(state.get("repair_history", []), default=str),
        "eval_memory": json.dumps(state.get("eval_memory", {}), default=str)
    })
    
    return {
        "generated_response": result,
        "response_sources": extract_citations(result)
    }

Step 4: Stateful Evaluation Node — The Core Loop Engine

This is where stateful evaluation differs fundamentally from post-hoc scoring. The evaluator reads live state, produces structured failures, and writes repair instructions back into state.

class PaymentsEvaluator:
    """
    Multi-dimensional evaluator that reads graph state and produces
    structured, actionable failure diagnoses. NOT a simple LLM judge.
    """
    
    @staticmethod
    async def evaluate(state: PaymentsState) -> List[EvalDimensionResult]:
        results = []
        
        # DIMENSION 1: Account Consistency
        results.append(await PaymentsEvaluator._check_account_consistency(state))
        
        # DIMENSION 2: Policy Compliance
        results.append(await PaymentsEvaluator._check_policy_compliance(state))
        
        # DIMENSION 3: Regulatory Adherence
        results.append(await PaymentsEvaluator._check_regulatory_adherence(state))
        
        # DIMENSION 4: Freshness
        results.append(PaymentsEvaluator._check_freshness(state))
        
        # DIMENSION 5: Factual Accuracy (LLM-assisted, grounded in state)
        results.append(await PaymentsEvaluator._check_factual_accuracy(state))
        
        return results
    
    @staticmethod
    async def _check_account_consistency(state: PaymentsState) -> EvalDimensionResult:
        """Validates response claims against live account state."""
        response = state.get("generated_response", "")
        account = state.get("account_state", {})
        
        failures = []
        
        # Extract numeric claims from response and verify against account
        import re
        amount_claims = re.findall(r'\$[\d,]+\.?\d*', response)
        balance = account.get("available_balance", 0)
        daily_used = account.get("daily_limit_used", 0)
        daily_limit = account.get("daily_transfer_limit", 0)
        
        for claim in amount_claims:
            claimed_amount = float(claim.replace("$", "").replace(",", ""))
            # Check if claimed amounts match known account values
            if abs(claimed_amount - balance) < 1 and "balance" in response.lower():
                continue  # Correct balance reference
            elif abs(claimed_amount - daily_limit) < 1 and "limit" in response.lower():
                continue  # Correct limit reference
            # Flag unverified amounts
            elif claimed_amount > 0 and claimed_amount not in [balance, daily_used, daily_limit]:
                failures.append({
                    "type": "unverified_amount",
                    "claimed": claim,
                    "instruction": f"Verify ${claimed_amount} against account state. "
                                   f"Available balance: ${balance}, Daily limit: ${daily_limit}"
                })
        
        # Check KYC tier consistency
        kyc_tier = account.get("kyc_tier", "standard")
        if "international" in response.lower() and kyc_tier == "basic":
            failures.append({
                "type": "kyc_tier_mismatch",
                "claimed_capability": "international_transfer",
                "actual_tier": kyc_tier,
                "instruction": f"Customer KYC tier is '{kyc_tier}'. International transfers require 'verified' or 'premium'. Correct response."
            })
        
        passed = len(failures) == 0
        return EvalDimensionResult(
            dimension="account_consistency",
            passed=passed,
            score=1.0 if passed else max(0, 1.0 - len(failures) * 0.3),
            failures=failures,
            repair_instructions="Refetch account state and regenerate with corrected values" if failures else None
        )
    
    @staticmethod
    async def _check_policy_compliance(state: PaymentsState) -> EvalDimensionResult:
        """Validates fee/limit claims against policy engine output."""
        response = state.get("generated_response", "")
        policy = state.get("policy_context", {})
        failures = []
        
        fee_schedule = policy.get("fee_schedule", {})
        velocity_limits = policy.get("velocity_limits", {})
        
        # Verify cited fees match policy
        for transfer_type, fee_info in fee_schedule.items():
            if transfer_type.lower() in response.lower():
                expected_fee = fee_info.get("fee", 0)
                if f"${expected_fee}" not in response and f"$ {expected_fee}" not in response:
                    failures.append({
                        "type": "fee_mismatch",
                        "transfer_type": transfer_type,
                        "expected_fee": expected_fee,
                        "instruction": f"Cited fee for {transfer_type} doesn't match policy (${expected_fee}). Update response."
                    })
        
        passed = len(failures) == 0
        return EvalDimensionResult(
            dimension="policy_compliance",
            passed=passed,
            score=1.0 if passed else max(0, 1.0 - len(failures) * 0.25),
            failures=failures,
            repair_instructions="Update policy context and regenerate with correct fees/limits" if failures else None
        )
    
    @staticmethod
    async def _check_regulatory_adherence(state: PaymentsState) -> EvalDimensionResult:
        """Ensures regulatory flags are addressed in response."""
        response = state.get("generated_response", "")
        flags = state.get("regulatory_flags", [])
        failures = []
        
        for flag in flags:
            flag_type = flag.get("type", "")
            if flag_type == "ofac_match" and "sanctions" not in response.lower() and "ofac" not in response.lower():
                failures.append({
                    "type": "unaddressed_regulatory_flag",
                    "flag_type": flag_type,
                    "entity": flag.get("entity"),
                    "instruction": f"OFAC match for '{flag.get('entity')}' exists but response doesn't address sanctions screening. Must disclose."
                })
            
            if flag.get("sar_trigger") and "reporting" not in response.lower():
                failures.append({
                    "type": "unaddressed_sar_trigger",
                    "instruction": "SAR trigger detected but response omits regulatory reporting disclosure."
                })
        
        passed = len(failures) == 0
        return EvalDimensionResult(
            dimension="regulatory_adherence",
            passed=passed,
            score=1.0 if passed else 0.0,  # Binary: regulatory compliance is non-negotiable
            failures=failures,
            repair_instructions="Add regulatory disclosures to response" if failures else None
        )
    
    @staticmethod
    def _check_freshness(state: PaymentsState) -> EvalDimensionResult:
        """Detects stale data that could cause incorrect responses."""
        failures = []
        max_age = 60  # Account data older than 60s is stale for payments
        
        account_ts = state.get("account_state", {}).get("last_updated", datetime.min)
        age = (datetime.utcnow() - account_ts).total_seconds()
        
        if age > max_age:
            failures.append({
                "type": "stale_account_data",
                "age_seconds": age,
                "threshold": max_age,
                "instruction": f"Account data is {age:.0f}s old (max: {max_age}s). Refetch before regenerating."
            })
        
        passed = len(failures) == 0
        return EvalDimensionResult(
            dimension="freshness",
            passed=passed,
            score=1.0 if passed else 0.5,
            failures=failures,
            repair_instructions="Refetch account state from core banking API" if failures else None
        )
    
    @staticmethod
    async def _check_factual_accuracy(state: PaymentsState) -> EvalDimensionResult:
        """LLM-assisted factual check grounded in retrieved state (not open-ended judgment)."""
        # Grounded evaluation prompt with explicit state anchoring
        eval_prompt = f"""Evaluate this banking response for factual accuracy.
        
RESPONSE: {state.get('generated_response', '')}

ACCOUNT STATE: {json.dumps(state.get('account_state', {}), default=str)}
POLICY: {json.dumps(state.get('policy_context', {}), default=str)}

Rate 0.0-1.0. List ONLY failures where response contradicts provided state.
Do NOT flag missing information—only contradictions.
Return JSON: {{"score": float, "failures": [{{"claim": str, "correction": str}}]}}"""
        
        result = await llm.with_structured_output(method="json_schema").ainvoke(eval_prompt)
        
        failures = result.get("failures", [])
        score = result.get("score", 1.0)
        
        return EvalDimensionResult(
            dimension="factual_accuracy",
            passed=score >= 0.9 and len(failures) == 0,
            score=score,
            failures=failures,
            repair_instructions="Correct contradictory claims using provided state" if failures else None
        )


async def evaluation_node(state: PaymentsState) -> dict:
    """
    Runs multi-dimensional evaluation and updates state with results.
    Increments eval_cycle for loop tracking.
    """
    cycle = state.get("eval_cycle", 0)
    
    results = await PaymentsEvaluator.evaluate(state)
    
    all_passed = all(r["passed"] for r in results)
    
    return {
        "eval_results": results,
        "eval_cycle": cycle + 1,
        "execution_trace_entry": {
            "node": "evaluation",
            "cycle": cycle,
            "all_passed": all_passed,
            "dimension_scores": {r["dimension"]: r["score"] for r in results},
            "timestamp": datetime.utcnow().isoformat()
        }
    }

Step 5: Diagnosis + Targeted Repair Nodes

When evaluation fails, the diagnosis node identifies which dimension failed and produces targeted repair actions—not generic "try again" signals.

async def diagnosis_node(state: PaymentsState) -> dict:
    """
    Converts eval failures into prioritized, targeted repair actions.
    Reads eval_memory to avoid repeating failed repairs.
    """
    eval_results = state.get("eval_results", [])
    eval_memory = state.get("eval_memory", {})
    repair_history = state.get("repair_history", [])
    
    failed_dimensions = [r for r in eval_results if not r["passed"]]
    
    repairs = []
    for result in failed_dimensions:
        dim = result["dimension"]
        
        # Skip if this exact repair was already attempted and failed
        prior_failed = [
            r for r in repair_history 
            if r["target_dimension"] == dim 
            and r.get("status") == "failed"
        ]
        if prior_failed:
            continue  # Don't repeat failed repairs
        
        if dim == "freshness":
            repairs.append(RepairAction(
                target_dimension=dim,
                action_type="refetch_account",
                parameters={"source": "core_banking_api"},
                priority=1  # Highest: stale data invalidates everything
            ))
        elif dim == "account_consistency":
            repairs.append(RepairAction(
                target_dimension=dim,
                action_type="refetch_account",
                parameters={"fields": ["available_balance", "daily_limit_used", "kyc_tier"]},
                priority=2
            ))
        elif dim == "policy_compliance":
            repairs.append(RepairAction(
                target_dimension=dim,
                action_type="update_policy_context",
                parameters={"force_refresh": True},
                priority=2
            ))
        elif dim == "regulatory_adherence":
            repairs.append(RepairAction(
                target_dimension=dim,
                action_type="add_regulatory_flag",
                parameters={"flags": result["failures"]},
                priority=1  # Regulatory is always highest priority
            ))
        elif dim == "factual_accuracy":
            repairs.append(RepairAction(
                target_dimension=dim,
                action_type="cite_source",
                parameters={"contradictions": result["failures"]},
                priority=3
            ))
    
    # Sort by priority
    repairs.sort(key=lambda r: r["priority"])
    
    # Update eval memory with failure pattern
    failure_pattern = {
        "cycle": state.get("eval_cycle", 0),
        "failed_dimensions": [r["dimension"] for r in failed_dimensions],
        "timestamp": datetime.utcnow().isoformat()
    }
    
    updated_memory = {
        **eval_memory,
        "failure_patterns": eval_memory.get("failure_patterns", []) + [failure_pattern]
    }
    
    return {
        "pending_repairs": repairs,
        "eval_memory": updated_memory
    }


async def repair_node(state: PaymentsState) -> dict:
    """
    Executes targeted repairs based on diagnosis.
    Only modifies the specific state dimensions that failed.
    """
    repairs = state.get("pending_repairs", [])
    updates = {"repair_history": []}
    
    for repair in repairs:
        try:
            if repair["action_type"] == "refetch_account":
                fresh_account = await fetch_account_state(state["customer_id"])
                updates["account_state"] = {
                    **fresh_account,
                    "last_updated": datetime.utcnow(),
                    "source": "core_banking_api"
                }
                
            elif repair["action_type"] == "update_policy_context":
                fresh_policy = await fetch_transfer_policies(force_refresh=True)
                updates["policy_context"] = {
                    **fresh_policy,
                    "last_updated": datetime.utcnow()
                }
                
            elif repair["action_type"] == "add_regulatory_flag":
                existing = state.get("regulatory_flags", [])
                new_flags = repair["parameters"].get("flags", [])
                updates["regulatory_flags"] = existing + new_flags
                
            # Record successful repair
            updates["repair_history"].append({
                **repair,
                "status": "executed",
                "timestamp": datetime.utcnow().isoformat()
            })
            
        except Exception as e:
            updates["repair_history"].append({
                **repair,
                "status": "failed",
                "error": str(e),
                "timestamp": datetime.utcnow().isoformat()
            })
    
    updates["pending_repairs"] = []  # Clear after execution
    return updates

Step 6: Assemble the Evaluation Loop Graph

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

workflow = StateGraph(PaymentsState)

# Register nodes
workflow.add_node("retrieve", payments_retrieval_node)
workflow.add_node("generate", generation_node)
workflow.add_node("evaluate", evaluation_node)
workflow.add_node("diagnose", diagnosis_node)
workflow.add_node("repair", repair_node)

# Terminal nodes
workflow.add_node("finalize_verified", lambda s: {
    "final_response": s["generated_response"],
    "response_quality": "verified",
    "eval_summary": f"Passed all {len(s['eval_results'])} dimensions on cycle {s['eval_cycle']}"
})

workflow.add_node("finalize_repaired", lambda s: {
    "final_response": s["generated_response"],
    "response_quality": "repaired",
    "eval_summary": f"Repaired after {len(s['repair_history'])} actions over {s['eval_cycle']} cycles"
})

workflow.add_node("finalize_escalated", lambda s: {
    "final_response": "I need additional verification to answer this accurately. Connecting you with a payments specialist.",
    "response_quality": "escalated",
    "eval_summary": f"Could not verify after {s['eval_cycle']} cycles. Escalated."
})

# === EDGES ===
workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", "evaluate")

# Evaluation loop conditional edge
def route_after_evaluation(state: PaymentsState) -> str:
    all_passed = all(r["passed"] for r in state.get("eval_results", []))
    cycle = state.get("eval_cycle", 0)
    max_cycles = state.get("max_eval_cycles", 3)
    
    if all_passed:
        if cycle <= 1:
            return "finalize_verified"
        else:
            return "finalize_repaired"
    
    if cycle >= max_cycles:
        return "finalize_escalated"
    
    return "diagnose"

workflow.add_conditional_edges("evaluate", route_after_evaluation, {
    "finalize_verified": "finalize_verified",
    "finalize_repaired": "finalize_repaired",
    "finalize_escalated": "finalize_escalated",
    "diagnose": "diagnose"
})

# Repair loop
workflow.add_edge("diagnose", "repair")
workflow.add_edge("repair", "generate")  # Regenerate with repaired state → back to eval

# Terminal edges
workflow.add_edge("finalize_verified", END)
workflow.add_edge("finalize_repaired", END)
workflow.add_edge("finalize_escalated", END)

# Compile with persistent checkpointing
checkpointer = PostgresSaver.from_conn_string("postgresql://nova-payments-db")
app = workflow.compile(checkpointer=checkpointer)

Step 7: Execute and Observe the Loop in Action

config = {"configurable": {"thread_id": "nova-pay-cust-99281-session-4421"}}

result = await app.ainvoke({
    "messages": [{"role": "user", "content": "Can I send $15,000 to my contractor in Germany today?"}],
    "customer_id": "cust-99281",
    "session_id": "pay-sess-20240805-143022",
    "processing_start_time": datetime.utcnow(),
    # Remaining fields initialized by retrieval node
}, config=config)

print(f"Response: {result['final_response']}")
print(f"Quality: {result['response_quality']}")
print(f"Eval Cycles: {result['eval_cycle']}")
print(f"Repairs: {[r['action_type'] for r in result.get('repair_history', [])]}")
print(f"Eval Summary: {result['eval_summary']}")

# Example output for a response that needed repair:
# Response: Based on your Verified KYC tier, you can send up to $25,000/day internationally.
#           Your remaining daily limit is $18,500. The wire fee is $35. Note: Transactions 
#           to Germany are subject to OFAC screening, which may add 1-2 business days.
# Quality: repaired
# Eval Cycles: 2
# Repairs: ['refetch_account', 'add_regulatory_flag']
# Eval Summary: Repaired after 2 actions over 2 cycles

Key Design Principles for Stateful Evaluation Loops

1. Evaluation Produces Repairs, Not Scores

A score of 0.7 tells the agent nothing actionable. Structured failures with repair_instructions tell it exactly what to fix. The evaluation node is a diagnostic instrument, not a grading rubric.

2. Repair History Prevents Infinite Loops

The repair_history field ensures the same repair is never attempted twice. If refetch_account fails once, the next cycle skips it and escalates instead. This is the primary loop prevention mechanism.

3. Eval Memory Accumulates Across Cycles

eval_memory.failure_patterns lets later cycles learn from earlier ones. If cycle 1 failed on freshness and cycle 2 fails on account consistency, the system recognizes correlated failures and adjusts repair priority.

4. Max Cycles Is a Hard Ceiling, Not a Suggestion

max_eval_cycles = 3 is enforced in the routing function, not in the evaluation node. Even if the evaluator keeps finding failures, the graph terminates. Three cycles catches 95%+ of repairable issues; beyond that, escalation is cheaper than continued computation.

5. Dimensions Are Independent and Parallelizable

Each evaluation dimension checks one concern. They can run concurrently (asyncio.gather) and fail independently. A freshness failure doesn't invalidate a correct policy citation—it just means the account values need refreshing.

Production Monitoring for Evaluation Loops

loop_metrics = {
    "avg_eval_cycles": 1.3,           # Target: ≤1.5
    "repair_rate": 0.22,              # % of responses needing repair
    "escalation_rate": 0.03,          # Target: <5%
    "most_common_failure_dim": "freshness",  # Drives infra improvements
    "avg_repair_latency_ms": 450,     # Target: <500ms per repair
    "loop_detection_rate": 0.001,     # Should be near zero
    "eval_memory_hit_rate": 0.35,     # % of repairs informed by prior patterns
    "p99_total_latency_ms": 4200      # Target: <5000ms including repairs
}

If avg_eval_cycles > 2, your retrieval quality needs improvement. If escalation_rate > 5%, either repair logic is insufficient or max_cycles is too low. If freshness dominates failures, reduce account cache TTL. These metrics drive infrastructure decisions, not just model tuning.

Conclusion

Stateful evaluation loops transform RAG from a generate-and-hope pipeline into a self-correcting reasoning system. For digital banking payments, this isn't optional—regulatory compliance demands that every response be verified against live account state, current policy, and active sanctions lists before reaching a customer. The key insight is that evaluation state (eval_results, repair_history, eval_memory) must be first-class graph state, not external logging. When evaluation results flow through the same state mechanism as retrieval and generation, the loop becomes a native part of the computation rather than a bolted-on afterthought. Build your evaluation loop with structured failures, targeted repairs, bounded cycles, and persistent memory. In payments, the cost of skipping this rigor isn't a bad metric—it's a consent order.