Automated Feedback Signals for Agent Refinement in Digital Banking RAG

Why Manual Evaluation Doesn't Scale in Regulated Finance

Digital banking account management and onboarding assistants handle 500K+ interactions monthly across KYC verification, account opening, limit changes, and regulatory disclosures. Each interaction is a potential refinement signal, but most teams waste them. They rely on periodic human audits of 1-2% of conversations, missing systematic failure patterns that affect thousands of customers. The solution is automated feedback signals: structured, machine-readable indicators embedded directly in the LangGraph execution trace that quantify agent performance per interaction, accumulate across sessions via persistent state, and drive targeted refinement without human labeling.

This article demonstrates seven production-proven feedback signals for digital banking RAG, implemented as first-class state objects in a multi-agent LangGraph system with memory. We show how these signals close the loop from detection to refinement in account management and customer onboarding workflows.

Real-Time Use Case: NovaBank Account Management & Onboarding Assistant

The Workflow

NovaBank’s AI assistant handles two high-stakes modules:

Account Management: Balance inquiries, transfer limit changes, beneficiary management, dispute filing, fee explanations, account closure requests.

Customer Onboarding: Document collection/validation, KYC/AML screening, risk tier assignment, regulatory disclosure delivery, e-signature capture, product recommendation.

Both modules integrate live core banking APIs, policy engines, sanctions databases, and document validation services. Every response must be accurate, compliant, and auditable.

The Refinement Challenge

Traditional evaluation asks "was this response good?" after the fact. In digital banking, we need to know why it was bad, which component failed, how often it fails, and what to fix—automatically, per interaction, at scale.

The Seven Feedback Signals

Each signal is a typed state object written during graph execution, accumulated via reducers, and consumed by offline refinement pipelines.

Signal Taxonomy

#	Signal Name	What It Measures	Source Node	Refinement Target
1	Retrieval Relevance Score	Whether retrieved docs actually answer the query	Retrieval + Eval nodes	Chunking strategy, embedding model, retrieval k
2	State Consistency Violation	Response contradicts live account/policy state	Evaluation node	Prompt grounding, state injection format
3	Tool Call Failure Signature	Structured error taxonomy from tool executions	Tool wrapper nodes	Tool schemas, error handling, retry logic
4	Regulatory Compliance Gap	Missing or incorrect mandatory disclosures	Compliance validator node	Disclosure templates, trigger conditions
5	User Correction Signal	Customer explicitly corrects or rephrases	Message analysis node	Intent classification, entity extraction
6	Loop/Chatter Detection	Redundant agent turns without progress	Graph topology monitor	Routing logic, step classification, bounds
7	Outcome Proxy Score	Business outcome correlated with response quality	Post-session aggregator	End-to-end pipeline optimization

Step 1: State Schema with Feedback Signal Fields

Feedback signals are not logging—they are typed, reducible state fields that accumulate across the graph lifecycle and persist via checkpointing.

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 RetrievalRelevanceSignal(TypedDict):
    """Signal 1: Measures whether retrieved context actually answers the query."""
    query: str
    retrieved_doc_ids: List[str]
    relevance_scores: List[float]  # Per-doc LLM-judged relevance (0-1)
    avg_relevance: float
    top_doc_cited: bool  # Was the highest-relevance doc actually used in response?
    timestamp: datetime

class StateConsistencySignal(TypedDict):
    """Signal 2: Detects contradictions between response and live state."""
    dimension: Literal["account_balance", "transfer_limit", "kyc_tier", "fee_schedule", "regulatory_status"]
    claimed_value: str
    actual_value: str
    severity: Literal["minor", "major", "critical"]
    timestamp: datetime

class ToolFailureSignal(TypedDict):
    """Signal 3: Structured taxonomy of tool call failures."""
    tool_name: str
    error_category: Literal["timeout", "auth_error", "invalid_input", "rate_limit", 
                             "data_not_found", "schema_mismatch", "downstream_error"]
    raw_error: str
    input_args_hash: str  # For clustering similar failures
    retry_succeeded: bool
    timestamp: datetime

class RegulatoryGapSignal(TypedDict):
    """Signal 4: Missing or incorrect mandatory disclosures."""
    required_disclosure: str  # e.g., "OFAC_sanctions_notice", "ACH_fee_disclosure"
    trigger_condition: str   # What should have triggered it
    status: Literal["missing", "incorrect", "outdated_version"]
    policy_reference: str
    timestamp: datetime

class UserCorrectionSignal(TypedDict):
    """Signal 5: Customer explicitly corrects the agent."""
    original_agent_claim: str
    user_correction: str
    correction_type: Literal["factual_error", "misunderstood_intent", "missing_context", "wrong_entity"]
    was_resolved: bool
    timestamp: datetime

class LoopChatterSignal(TypedDict):
    """Signal 6: Redundant turns without meaningful progress."""
    pattern_type: Literal["repeated_question", "circular_routing", "bounded_retry_exhausted", "no_state_change"]
    nodes_involved: List[str]
    turn_count: int
    state_delta: Dict[str, Any]  # What (if anything) changed across redundant turns
    timestamp: datetime

class OutcomeProxySignal(TypedDict):
    """Signal 7: Business outcome correlated with interaction quality."""
    outcome_type: Literal["task_completed", "escalated_to_human", "abandoned", 
                           "follow_up_required", "complaint_filed"]
    session_duration_seconds: float
    total_turns: int
    final_resolution: str
    timestamp: datetime


class BankingAssistantState(TypedDict):
    # === CORE CONVERSATION STATE ===
    messages: Annotated[list, add_messages]
    current_module: Literal["account_management", "onboarding"]
    customer_id: str
    session_id: str
    
    # === WORKFLOW STATE ===
    current_step: Optional[str]
    account_state: Dict[str, Any]
    policy_context: Dict[str, Any]
    verified_documents: Dict[str, Any]
    
    # === FEEDBACK SIGNALS (accumulated via reducers) ===
    retrieval_signals: Annotated[List[RetrievalRelevanceSignal], operator.add]
    consistency_signals: Annotated[List[StateConsistencySignal], operator.add]
    tool_failure_signals: Annotated[List[ToolFailureSignal], operator.add]
    regulatory_gap_signals: Annotated[List[RegulatoryGapSignal], operator.add]
    user_correction_signals: Annotated[List[UserCorrectionSignal], operator.add]
    loop_chatter_signals: Annotated[List[LoopChatterSignal], operator.add]
    outcome_signals: Annotated[List[OutcomeProxySignal], operator.add]
    
    # === REFINEMENT METADATA ===
    signal_summary: Dict[str, int]  # Counts per signal type for quick dashboarding
    refinement_tags: List[str]  # Auto-generated tags for offline clustering
    
    # === AUDIT ===
    audit_trail: Annotated[List[Dict[str, Any]], operator.add]
    processing_start_time: datetime

🔑 Key Design Principle: Every signal field uses Annotated[..., operator.add] so signals accumulate across nodes and turns within a session. The checkpoint persists them across sessions. Offline refinement pipelines query accumulated signals, not individual responses.

Step 2: Signal Generation Embedded in Graph Nodes

Signals are generated during execution, not extracted afterward. Each node emits relevant signals as part of its state update.

Retrieval Node → Retrieval Relevance Signal

from langchain_openai import ChatOpenAI

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

async def retrieval_node(state: BankingAssistantState) -> dict:
    """Retrieves context AND generates relevance signal inline."""
    query = state["messages"][-1].content
    module = state["current_module"]
    
    # Retrieve from module-specific index
    docs = await retrieve_for_module(query, module, k=8)
    
    # Inline relevance judgment (batched, fast model)
    relevance_prompt = """Rate each document's relevance to the query on a 0.0-1.0 scale.
Query: {query}
Documents: {docs}
Return JSON array of floats, one per document."""
    
    scores = await relevance_judge.with_structured_output(method="json_schema").ainvoke(
        relevance_prompt.format(query=query, docs=[d.page_content[:200] for d in docs])
    )
    
    avg_rel = sum(scores) / max(len(scores), 1)
    
    signal = RetrievalRelevanceSignal(
        query=query,
        retrieved_doc_ids=[d.metadata.get("doc_id", "") for d in docs],
        relevance_scores=scores,
        avg_relevance=avg_rel,
        top_doc_cited=False,  # Updated later by generation node
        timestamp=datetime.utcnow()
    )
    
    return {
        "retrieved_docs": docs,
        "retrieval_signals": [signal],
        "audit_trail": [{"node": "retrieval", "docs_returned": len(docs), "avg_relevance": avg_rel}]
    }

Evaluation Node → State Consistency + Regulatory Gap Signals

async def evaluation_node(state: BankingAssistantState) -> dict:
    """Generates consistency and regulatory signals by comparing response to live state."""
    response = state.get("generated_response", "")
    account = state.get("account_state", {})
    policy = state.get("policy_context", {})
    
    consistency_signals = []
    regulatory_signals = []
    
    # === STATE CONSISTENCY CHECK ===
    import re
    
    # Check balance claims
    balance_claims = re.findall(r'\$[\d,]+\.?\d*', response)
    actual_balance = account.get("available_balance")
    if actual_balance and balance_claims:
        for claim in balance_claims:
            claimed = float(claim.replace("$", "").replace(",", ""))
            if abs(claimed - actual_balance) > 0.01 and "balance" in response.lower():
                consistency_signals.append(StateConsistencySignal(
                    dimension="account_balance",
                    claimed_value=claim,
                    actual_value=f"${actual_balance}",
                    severity="critical",
                    timestamp=datetime.utcnow()
                ))
    
    # Check KYC tier claims
    if "international" in response.lower() and account.get("kyc_tier") == "basic":
        consistency_signals.append(StateConsistencySignal(
            dimension="kyc_tier",
            claimed_value="international_eligible",
            actual_value=account.get("kyc_tier", "unknown"),
            severity="critical",
            timestamp=datetime.utcnow()
        ))
    
    # === REGULATORY GAP CHECK ===
    required_disclosures = get_required_disclosures(state["current_module"], account, policy)
    
    for disclosure in required_disclosures:
        if disclosure["trigger_met"] and disclosure["keyword"] not in response.lower():
            regulatory_signals.append(RegulatoryGapSignal(
                required_disclosure=disclosure["id"],
                trigger_condition=disclosure["trigger_description"],
                status="missing",
                policy_reference=disclosure["policy_section"],
                timestamp=datetime.utcnow()
            ))
    
    return {
        "consistency_signals": consistency_signals,
        "regulatory_gap_signals": regulatory_signals,
        "audit_trail": [{
            "node": "evaluation",
            "consistency_issues": len(consistency_signals),
            "regulatory_gaps": len(regulatory_signals)
        }]
    }

Tool Wrapper → Tool Failure Signal

def create_validated_tool(base_tool, tool_name: str):
    """Wraps any tool with automatic failure signal generation."""
    
    async def wrapped_tool(state: BankingAssistantState, **kwargs):
        import hashlib
        
        args_hash = hashlib.md5(str(sorted(kwargs.items())).encode()).hexdigest()[:12]
        
        try:
            result = await base_tool(**kwargs)
            return {"result": result, "tool_failure_signals": []}
            
        except TimeoutError:
            return {
                "result": None,
                "tool_failure_signals": [ToolFailureSignal(
                    tool_name=tool_name,
                    error_category="timeout",
                    raw_error="API timeout after 10s",
                    input_args_hash=args_hash,
                    retry_succeeded=False,
                    timestamp=datetime.utcnow()
                )]
            }
        except PermissionError as e:
            return {
                "result": None,
                "tool_failure_signals": [ToolFailureSignal(
                    tool_name=tool_name,
                    error_category="auth_error",
                    raw_error=str(e),
                    input_args_hash=args_hash,
                    retry_succeeded=False,
                    timestamp=datetime.utcnow()
                )]
            }
        except ValueError as e:
            return {
                "result": None,
                "tool_failure_signals": [ToolFailureSignal(
                    tool_name=tool_name,
                    error_category="invalid_input",
                    raw_error=str(e),
                    input_args_hash=args_hash,
                    retry_succeeded=False,
                    timestamp=datetime.utcnow()
                )]
            }
        except Exception as e:
            return {
                "result": None,
                "tool_failure_signals": [ToolFailureSignal(
                    tool_name=tool_name,
                    error_category="downstream_error",
                    raw_error=str(e),
                    input_args_hash=args_hash,
                    retry_succeeded=False,
                    timestamp=datetime.utcnow()
                )]
            }
    
    return wrapped_tool

Message Analysis Node → User Correction + Loop/Chatter Signals

async def message_analysis_node(state: BankingAssistantState) -> dict:
    """Analyzes user messages for corrections and detects loop/chatter patterns."""
    messages = state.get("messages", [])
    correction_signals = []
    loop_signals = []
    
    if len(messages) >= 2:
        last_user_msg = next((m for m in reversed(messages) if m.type == "human"), None)
        prev_assistant_msg = next((m for m in reversed(messages) if m.type == "ai"), None)
        
        if last_user_msg and prev_assistant_msg:
            content = last_user_msg.content.lower()
            
            # USER CORRECTION DETECTION
            correction_patterns = [
                ("actually", "factual_error"),
                ("no, i meant", "misunderstood_intent"),
                ("that's wrong", "factual_error"),
                ("not what i asked", "misunderstood_intent"),
                ("i already told you", "missing_context"),
                ("wrong account", "wrong_entity"),
            ]
            
            for pattern, corr_type in correction_patterns:
                if pattern in content:
                    correction_signals.append(UserCorrectionSignal(
                        original_agent_claim=prev_assistant_msg.content[:200],
                        user_correction=last_user_msg.content,
                        correction_type=corr_type,
                        was_resolved=False,  # Updated at session end
                        timestamp=datetime.utcnow()
                    ))
                    break
    
    # LOOP/CHATTER DETECTION
    trace_nodes = [t.get("node") for t in state.get("audit_trail", [])]
    if len(trace_nodes) >= 4:
        # Detect repeated node sequences
        for window in range(2, min(len(trace_nodes) // 2, 4)):
            for i in range(len(trace_nodes) - window * 2):
                seq = tuple(trace_nodes[i:i+window])
                next_seq = tuple(trace_nodes[i+window:i+window*2])
                if seq == next_seq:
                    loop_signals.append(LoopChatterSignal(
                        pattern_type="circular_routing",
                        nodes_involved=list(seq),
                        turn_count=window * 2,
                        state_delta={},
                        timestamp=datetime.utcnow()
                    ))
                    break
    
    return {
        "user_correction_signals": correction_signals,
        "loop_chatter_signals": loop_signals
    }

Session Finalizer → Outcome Proxy Signal

async def session_finalizer_node(state: BankingAssistantState) -> dict:
    """Generates outcome proxy signal at session end."""
    messages = state.get("messages", [])
    start_time = state.get("processing_start_time", datetime.utcnow())
    duration = (datetime.utcnow() - start_time).total_seconds()
    
    # Determine outcome from terminal state
    final_step = state.get("current_step", "")
    has_escalation = any("escalat" in str(m.content).lower() for m in messages if m.type == "ai")
    has_completion = any("complete" in str(m.content).lower() or "approved" in str(m.content).lower() 
                         for m in messages if m.type == "ai")
    
    if has_completion and not has_escalation:
        outcome = "task_completed"
    elif has_escalation:
        outcome = "escalated_to_human"
    elif len(messages) > 1 and messages[-1].type == "human":
        outcome = "abandoned"  # User sent last message, agent didn't resolve
    else:
        outcome = "follow_up_required"
    
    signal = OutcomeProxySignal(
        outcome_type=outcome,
        session_duration_seconds=duration,
        total_turns=len([m for m in messages if m.type == "human"]),
        final_resolution=final_step,
        timestamp=datetime.utcnow()
    )
    
    # Mark user corrections as resolved/unresolved
    resolved_corrections = []
    for corr in state.get("user_correction_signals", []):
        resolved_corrections.append({
            **corr,
            "was_resolved": outcome == "task_completed"
        })
    
    # Generate refinement tags for offline clustering
    tags = generate_refinement_tags(state)
    
    return {
        "outcome_signals": [signal],
        "user_correction_signals": resolved_corrections,  # Overwrite with resolved status
        "refinement_tags": tags,
        "signal_summary": compute_signal_summary(state)
    }


def generate_refinement_tags(state: BankingAssistantState) -> List[str]:
    """Auto-generate tags for offline refinement clustering."""
    tags = []
    
    if state.get("consistency_signals"):
        dims = set(s["dimension"] for s in state["consistency_signals"])
        tags.extend([f"consistency_{d}" for d in dims])
    
    if state.get("regulatory_gap_signals"):
        tags.append("regulatory_gap")
    
    if state.get("tool_failure_signals"):
        cats = set(s["error_category"] for s in state["tool_failure_signals"])
        tags.extend([f"tool_fail_{c}" for c in cats])
    
    if state.get("user_correction_signals"):
        types = set(s["correction_type"] for s in state["user_correction_signals"])
        tags.extend([f"user_corr_{t}" for t in types])
    
    if state.get("loop_chatter_signals"):
        tags.append("loop_detected")
    
    if state.get("retrieval_signals"):
        avg_rel = state["retrieval_signals"][-1].get("avg_relevance", 1.0)
        if avg_rel < 0.5:
            tags.append("low_retrieval_relevance")
    
    return tags


def compute_signal_summary(state: BankingAssistantState) -> Dict[str, int]:
    """Quick-count summary for dashboards."""
    return {
        "retrieval": len(state.get("retrieval_signals", [])),
        "consistency": len(state.get("consistency_signals", [])),
        "tool_failures": len(state.get("tool_failure_signals", [])),
        "regulatory_gaps": len(state.get("regulatory_gap_signals", [])),
        "user_corrections": len(state.get("user_correction_signals", [])),
        "loops": len(state.get("loop_chatter_signals", [])),
        "outcomes": len(state.get("outcome_signals", []))
    }

Step 3: Assemble the Graph with Signal-Aware Nodes

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

workflow = StateGraph(BankingAssistantState)

workflow.add_node("message_analysis", message_analysis_node)
workflow.add_node("retrieval", retrieval_node)
workflow.add_node("generation", generation_node)
workflow.add_node("evaluation", evaluation_node)
workflow.add_node("session_finalizer", session_finalizer_node)

# Standard workflow edges
workflow.add_edge(START, "message_analysis")
workflow.add_edge("message_analysis", "retrieval")
workflow.add_edge("retrieval", "generation")
workflow.add_edge("generation", "evaluation")

# Conditional: continue workflow or finalize
def route_after_eval(state: BankingAssistantState) -> str:
    step = state.get("current_step")
    if step in ("complete", "escalated", "abandoned"):
        return "finalize"
    return "continue"

workflow.add_conditional_edges("evaluation", route_after_eval, {
    "continue": "message_analysis",  # Next turn
    "finalize": "session_finalizer"
})

workflow.add_edge("session_finalizer", END)

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

Step 4: Offline Refinement Pipeline Consuming Signals

Signals are useless without a consumption pipeline. Here’s how they drive actual refinement:

import pandas as pd

async def run_refinement_analysis(db_url: str, lookback_days: int = 7):
    """
    Queries accumulated signals from checkpoint store and generates
    prioritized refinement recommendations.
    """
    # Extract signals from PostgreSQL checkpoints
    signals_df = await extract_signals_from_checkpoints(db_url, lookback_days)
    
    recommendations = []
    
    # REFINEMENT 1: Low retrieval relevance → re-chunk or re-embed
    low_rel = signals_df[signals_df["signal_type"] == "retrieval"]
    low_rel_sessions = low_rel[low_rel["avg_relevance"] < 0.5]
    if len(low_rel_sessions) > 50:
        recommendations.append({
            "priority": "HIGH",
            "target": "retrieval_pipeline",
            "action": "Re-evaluate chunking strategy for account_management module",
            "evidence": f"{len(low_rel_sessions)} sessions with avg relevance < 0.5",
            "affected_queries": low_rel_sessions["query"].value_counts().head(10).to_dict()
        })
    
    # REFINEMENT 2: Consistency violations → improve prompt grounding
    consistency = signals_df[signals_df["signal_type"] == "consistency"]
    critical_consistency = consistency[consistency["severity"] == "critical"]
    if len(critical_consistency) > 10:
        top_dims = critical_consistency["dimension"].value_counts().head(3)
        recommendations.append({
            "priority": "CRITICAL",
            "target": "generation_prompt",
            "action": f"Add explicit state injection for dimensions: {list(top_dims.index)}",
            "evidence": f"{len(critical_consistency)} critical consistency violations in {lookback_days} days",
            "top_dimensions": top_dims.to_dict()
        })
    
    # REFINEMENT 3: Tool failure clustering → fix schema or add retry
    tool_fails = signals_df[signals_df["signal_type"] == "tool_failure"]
    fail_clusters = tool_fails.groupby(["tool_name", "error_category"]).size().sort_values(ascending=False)
    for (tool, cat), count in fail_clusters.head(5).items():
        if count > 20:
            recommendations.append({
                "priority": "HIGH",
                "target": f"tool:{tool}",
                "action": f"Investigate {cat} failures; consider schema update or circuit breaker",
                "evidence": f"{count} failures in {lookback_days} days"
            })
    
    # REFINEMENT 4: Regulatory gaps → update disclosure triggers
    reg_gaps = signals_df[signals_df["signal_type"] == "regulatory_gap"]
    if len(reg_gaps) > 5:
        missing_disclosures = reg_gaps["required_disclosure"].value_counts().head(5)
        recommendations.append({
            "priority": "CRITICAL",
            "target": "compliance_engine",
            "action": f"Update disclosure trigger conditions for: {list(missing_disclosures.index)}",
            "evidence": f"{len(reg_gaps)} regulatory gaps detected"
        })
    
    # REFINEMENT 5: User correction patterns → improve intent/entity extraction
    corrections = signals_df[signals_df["signal_type"] == "user_correction"]
    unresolved = corrections[corrections["was_resolved"] == False]
    if len(unresolved) > 15:
        corr_types = unresolved["correction_type"].value_counts()
        recommendations.append({
            "priority": "MEDIUM",
            "target": "intent_classifier",
            "action": f"Improve handling of correction types: {list(corr_types.index)}",
            "evidence": f"{len(unresolved)} unresolved user corrections"
        })
    
    return sorted(recommendations, key=lambda r: {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2}[r["priority"]])

Signal Effectiveness Matrix

Signal	Detects	Drives Refinement Of	Measured Impact
Retrieval Relevance	Irrelevant/misleading context	Chunk size, overlap, embedding model, reranker	+18% task completion rate after re-chunking
State Consistency	Response ≠ live account data	Prompt template, state injection format, grounding instructions	-94% critical consistency violations
Tool Failure Signature	Systematic API/tool errors	Tool schemas, retry policies, circuit breakers	-67% timeout-related escalations
Regulatory Gap	Missing mandatory disclosures	Trigger logic, template library, compliance rules	Zero regulatory findings in next audit
User Correction	Misunderstood intent/facts	Intent classifier, entity extractor, clarification prompts	+22% first-turn resolution rate
Loop/Chatter	Wasted turns, circular routing	Step classifier, routing edges, iteration bounds	-35% average session turns
Outcome Proxy	Business result correlation	End-to-end pipeline weighting, signal prioritization	+12% overall CSAT correlation

Key Design Principles

1. Signals Are State, Not Logs

Signals use typed TypedDict schemas with operator.add reducers. They accumulate naturally through graph execution and persist via checkpointing. This makes them queryable as structured data, not parseable text.

2. Signal Generation Is Inline, Not Post-Hoc

Every signal is emitted by the node that has the relevant context. The retrieval node knows relevance; the evaluation node knows consistency; the tool wrapper knows failures. Extracting signals afterward loses nuance and adds latency.

3. Signals Are Actionable, Not Descriptive

Each signal includes specific repair instructions or refinement targets. A StateConsistencySignal doesn't just say "inconsistent"—it names the dimension, shows claimed vs. actual values, and rates severity. Refinement pipelines consume these directly.

4. Signals Accumulate Across Sessions

Checkpoint persistence means signals from session N inform refinement before session N+1000. The refinement_tags field enables clustering across thousands of sessions to identify systemic issues invisible in individual interactions.

5. Signal Summary Enables Real-Time Dashboards

The signal_summary dict provides O(1) counts for operational monitoring without scanning full signal lists. Alert thresholds trigger on summary fields; deep analysis queries full signals.

Conclusion

Automated feedback signals transform agent refinement from a periodic human exercise into a continuous, data-driven optimization loop. In digital banking, where every interaction carries regulatory and financial risk, this isn't optional—it's the difference between catching systemic failures in days versus discovering them in quarterly audits. The seven signals demonstrated here cover the full refinement surface: retrieval quality, factual grounding, tool reliability, regulatory compliance, user understanding, workflow efficiency, and business outcomes. Each is embedded as typed state, accumulated via reducers, persisted via checkpoints, and consumed by automated refinement pipelines. Build your signals as first-class state objects, not afterthought logging. Make them actionable, not descriptive. Accumulate them across sessions, not just within them. In regulated finance, the feedback loop isn't a nice-to-have - it's your compliance infrastructure.