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 NameWhat It MeasuresSource NodeRefinement Target
1Retrieval Relevance ScoreWhether retrieved docs actually answer the queryRetrieval + Eval nodesChunking strategy, embedding model, retrieval k
2State Consistency ViolationResponse contradicts live account/policy stateEvaluation nodePrompt grounding, state injection format
3Tool Call Failure SignatureStructured error taxonomy from tool executionsTool wrapper nodesTool schemas, error handling, retry logic
4Regulatory Compliance GapMissing or incorrect mandatory disclosuresCompliance validator nodeDisclosure templates, trigger conditions
5User Correction SignalCustomer explicitly corrects or rephrasesMessage analysis nodeIntent classification, entity extraction
6Loop/Chatter DetectionRedundant agent turns without progressGraph topology monitorRouting logic, step classification, bounds
7Outcome Proxy ScoreBusiness outcome correlated with response qualityPost-session aggregatorEnd-to-end pipeline optimization
397

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

SignalDetectsDrives Refinement OfMeasured Impact
Retrieval RelevanceIrrelevant/misleading contextChunk size, overlap, embedding model, reranker+18% task completion rate after re-chunking
State ConsistencyResponse ≠ live account dataPrompt template, state injection format, grounding instructions-94% critical consistency violations
Tool Failure SignatureSystematic API/tool errorsTool schemas, retry policies, circuit breakers-67% timeout-related escalations
Regulatory GapMissing mandatory disclosuresTrigger logic, template library, compliance rulesZero regulatory findings in next audit
User CorrectionMisunderstood intent/factsIntent classifier, entity extractor, clarification prompts+22% first-turn resolution rate
Loop/ChatterWasted turns, circular routingStep classifier, routing edges, iteration bounds-35% average session turns
Outcome ProxyBusiness result correlationEnd-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.