FM-Specific Troubleshooting Framework Architecture

mindmap
  root((FM Troubleshooting))
    Golden Datasets
      Stratified by intent & difficulty
      Automated scoring pipeline
      Hallucination detection probes
      Factual grounding verification
      Regression testing across versions
      Adversarial edge cases
    Output Diffing
      Text diff — character level
      Semantic diff — embedding cosine
      Structured field diff — price, stock
      Temporal stability — 24h consistency
      Version comparison — prompt A vs B
      Drift detection — gradual quality decay
    Reasoning Path Tracing
      CoT step extraction
      Logical error detection
      Contradiction finder
      Unsupported conclusion detector
      Circular reasoning detection
      Missing premise identification
    Specialized Pipelines
      FM vs traditional ML comparison
      Prompt capture & versioning
      Response capture & scoring
      Quality scoring — multi-dimensional
      Pluggable validators
      Continuous evaluation loops
    Failure Mode Taxonomy
      Hallucination — fabricated facts
      Catastrophic forgetting
      Prompt injection & jailbreak
      Context overflow & truncation
      Reasoning errors & logical flaws
      Persona drift — tone shift
      Knowledge boundary confusion

graph TB
    subgraph Detection["🔍 Detection Layer"]
        style Detection fill:#e8f4fd,stroke:#1976d2
        GD[Golden Dataset Scheduler<br/>Every 4 hours]
        OD[Output Differ<br/>Continuous — every response]
        RT[Reasoning Tracer<br/>Sampled — 5% of traffic]
        AD[Anomaly Detector<br/>Real-time streaming]
    end

    subgraph Analysis["🔬 Analysis Layer"]
        style Analysis fill:#fff3e0,stroke:#f57c00
        HS[Hallucination Scorer<br/>Factual grounding check]
        CA[Consistency Analyzer<br/>Cross-response comparison]
        RV[Reasoning Validator<br/>CoT step verification]
        FC[Failure Classifier<br/>Categorize failure mode]
    end

    subgraph Diagnosis["🩺 Diagnosis Layer"]
        style Diagnosis fill:#fce4ec,stroke:#c62828
        RC[Root Cause Identification<br/>Which component failed?]
        IA[Impact Assessment<br/>How many users affected?]
        CE[Correlation Engine<br/>Related to recent change?]
    end

    subgraph Action["⚡ Action Layer"]
        style Action fill:#e8f5e9,stroke:#2e7d32
        AL[Alert → PagerDuty/Slack]
        HR[Human Review Queue]
        RB[Rollback Prompt/Model]
        AR[Auto-Remediate<br/>Cache invalidation, fallback]
    end

    subgraph Feedback["🔄 Feedback Layer"]
        style Feedback fill:#f3e5f5,stroke:#7b1fa2
        UG[Update Golden Dataset]
        RA[Retrain / Adjust Prompt]
        VF[Validate Fix]
        CL[Close Loop — Mark Resolved]
    end

    GD --> HS
    OD --> CA
    RT --> RV
    AD --> FC

    HS --> RC
    CA --> RC
    RV --> RC
    FC --> RC

    RC --> IA
    IA --> CE

    CE -->|Recent deployment| RB
    CE -->|No recent change| HR
    CE -->|Known pattern| AR

    RB --> VF
    HR --> RA
    AR --> VF

    RA --> VF
    VF -->|Fix validated| CL
    VF -->|Fix failed| HR
    CL --> UG
    UG --> GD

stateDiagram-v2
    [*] --> Healthy

    Healthy --> Warning : quality < 0.85 OR<br/>hallucination_rate > 2%
    Healthy --> Maintenance : model_update OR<br/>prompt_change scheduled

    Warning --> Healthy : metrics recover within<br/>30 min (auto-heal)
    Warning --> Degraded : quality < 0.75 OR<br/>hallucination_rate > 5%
    Warning --> Maintenance : emergency prompt fix<br/>initiated

    Degraded --> Warning : partial recovery<br/>quality improving
    Degraded --> Critical : quality < 0.6 OR<br/>hallucination > 10% OR<br/>reasoning_errors > 20%
    Degraded --> Maintenance : rollback initiated

    Critical --> Maintenance : immediate rollback<br/>or failover triggered
    Critical --> Degraded : partial remediation<br/>applied

    Maintenance --> Recovering : update/rollback<br/>completed

    Recovering --> Healthy : all metrics within<br/>baseline for 1 hour
    Recovering --> Warning : metrics improving<br/>but not yet baseline
    Recovering --> Degraded : recovery failed<br/>metrics still poor

    note right of Healthy
        quality > 0.85
        hallucination < 2%
        latency < target
        consistency > 0.90
    end note

    note right of Critical
        quality < 0.6
        hallucination > 10%
        reasoning_errors > 20%
        PAGE ONCALL IMMEDIATELY
    end note

graph LR
    subgraph Schedule["⏰ Scheduled Trigger"]
        CRON[CloudWatch Events<br/>Every 4 hours]
    end

    subgraph Execute["▶️ Execute"]
        LOAD[Load Golden Dataset<br/>from S3]
        INVOKE[Invoke MangaAssist<br/>via Bedrock API]
        CAPTURE[Capture Full Response<br/>+ Latency + Tokens]
    end

    subgraph Score["📊 Score"]
        FACT[Factual Accuracy<br/>Required facts present?]
        HALL[Hallucination Check<br/>Prohibited claims absent?]
        REAS[Reasoning Quality<br/>Logical steps valid?]
        COMP[Composite Score<br/>Weighted average]
    end

    subgraph Compare["📈 Compare"]
        BASE[Load Baseline Scores<br/>from previous run]
        DIFF[Calculate Delta<br/>per test case]
        REG[Regression Detection<br/>Δ > threshold?]
    end

    subgraph Alert["🚨 Alert"]
        OK[All Clear<br/>Log to CloudWatch]
        WARN[Warning<br/>Slack notification]
        CRIT[Critical<br/>PagerDuty + auto-rollback]
    end

    CRON --> LOAD --> INVOKE --> CAPTURE
    CAPTURE --> FACT & HALL & REAS
    FACT & HALL & REAS --> COMP
    COMP --> BASE --> DIFF --> REG
    REG -->|No regression| OK
    REG -->|Minor regression| WARN
    REG -->|Major regression| CRIT

graph TB
    subgraph Input["Query Input"]
        Q[User Query]
    end

    subgraph Capture["Response Capture"]
        R1[Response at T₁]
        R2[Response at T₂]
    end

    subgraph Diff["Multi-Strategy Diff"]
        TD[Text Diff<br/>Levenshtein distance<br/>Character-level similarity]
        SD[Semantic Diff<br/>Titan Embeddings cosine<br/>Meaning-level similarity]
        FD[Structured Field Diff<br/>Extract prices, dates, quantities<br/>Field-by-field comparison]
    end

    subgraph Decide["Decision Engine"]
        AGG[Aggregate Scores<br/>text_sim × 0.2 + semantic_sim × 0.5 + field_sim × 0.3]
        THR{Score < 0.90?}
    end

    Q --> R1 & R2
    R1 & R2 --> TD & SD & FD
    TD & SD & FD --> AGG --> THR
    THR -->|Yes — Drift detected| ALERT[Alert + Log Diff Details]
    THR -->|No — Consistent| LOG[Log to Metrics]

graph TB
    subgraph Extract["Step 1: Extract Reasoning"]
        RESP[FM Response]
        PARSE[Parse CoT Indicators<br/>because, therefore, since<br/>numbered steps, bullet points]
        STEPS[Isolated Reasoning Steps<br/>Step 1, Step 2, ... Step N]
    end

    subgraph Validate["Step 2: Validate Each Step"]
        GROUND[Grounding Check<br/>Is step supported by<br/>retrieved context?]
        CONTRA[Contradiction Check<br/>Does step X contradict<br/>step Y?]
        LOGIC[Logical Validity<br/>Does step follow from<br/>previous steps?]
    end

    subgraph Conclude["Step 3: Validate Conclusion"]
        FOLLOW[Does conclusion follow<br/>from reasoning chain?]
        COMPLETE[Are all necessary<br/>steps present?]
        SUPPORT[Is conclusion supported<br/>by evidence?]
    end

    subgraph Result["Output"]
        VALID[✅ Valid Reasoning<br/>All checks passed]
        INVALID[❌ Invalid Reasoning<br/>Specific errors identified]
    end

    RESP --> PARSE --> STEPS
    STEPS --> GROUND & CONTRA & LOGIC
    GROUND --> FOLLOW
    CONTRA --> FOLLOW
    LOGIC --> FOLLOW
    FOLLOW --> COMPLETE --> SUPPORT
    SUPPORT -->|All pass| VALID
    SUPPORT -->|Any fail| INVALID

User: "I'm looking for a manga like Naruto but darker"
FM Response: "I recommend Berserk because it's a popular shonen manga (Step 1).
             Berserk has a similar art style to Naruto (Step 2).
             Therefore, you'll enjoy Berserk (Conclusion)."

Validation Result:
  Step 1: ❌ FACTUAL ERROR — Berserk is seinen, not shonen
  Step 2: ❌ FACTUAL ERROR — Art styles are very different (Miura vs Kishimoto)
  Conclusion: ⚠️ WEAKLY SUPPORTED — Conclusion may be correct (Berserk IS darker)
              but reasoning path is flawed

graph TB
    subgraph Capture["📥 Capture Layer"]
        style Capture fill:#e3f2fd,stroke:#1565c0
        PC[Prompt Capture<br/>System + User + History]
        RC[Response Capture<br/>Full text + metadata]
        CC[Context Capture<br/>Retrieved RAG chunks]
    end

    subgraph Score["📊 Scoring Layer"]
        style Score fill:#fff8e1,stroke:#f9a825
        QS[Quality Scorer<br/>Multi-dimensional]
        HS[Hallucination Scorer<br/>Grounding check]
        RS[Reasoning Scorer<br/>CoT validation]
        SS[Safety Scorer<br/>Guardrail check]
    end

    subgraph Validate["✅ Pluggable Validators"]
        style Validate fill:#e8f5e9,stroke:#2e7d32
        V1[Factual Grounding<br/>Response claims vs<br/>retrieved context]
        V2[Safety Check<br/>PII, toxicity,<br/>brand compliance]
        V3[Relevance Check<br/>Response addresses<br/>user intent?]
        V4[Completeness Check<br/>All required info<br/>present?]
    end

    subgraph Trend["📈 Trending & Drift"]
        style Trend fill:#f3e5f5,stroke:#7b1fa2
        AGG[Aggregate Metrics<br/>5-min windows]
        DRIFT[Drift Detection<br/>CUSUM / EWMA]
        BASELINE[Baseline Comparison<br/>vs last 7 days]
    end

    subgraph Act["⚡ Action"]
        style Act fill:#fce4ec,stroke:#c62828
        CW[CloudWatch Metrics<br/>Custom namespace]
        ALARM[CloudWatch Alarms<br/>Threshold-based]
        DASH[Grafana Dashboard<br/>Real-time visibility]
    end

    PC & RC & CC --> QS & HS & RS & SS
    QS --> V3 & V4
    HS --> V1
    SS --> V2
    V1 & V2 & V3 & V4 --> AGG
    AGG --> DRIFT & BASELINE
    DRIFT --> ALARM
    BASELINE --> CW
    CW --> DASH
    ALARM --> DASH

from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from datetime import datetime
from enum import Enum


class FailureMode(Enum):
    """All recognized FM-specific failure modes"""
    HALLUCINATION = "hallucination"
    REASONING_ERROR = "reasoning_error"
    PROMPT_INJECTION = "prompt_injection"
    CONTEXT_OVERFLOW = "context_overflow"
    PERSONA_DRIFT = "persona_drift"
    KNOWLEDGE_GAP = "knowledge_gap"
    OVER_REFUSAL = "over_refusal"
    STALE_INFORMATION = "stale_information"
    INCONSISTENCY = "inconsistency"
    TOKEN_INFLATION = "token_inflation"


class FMHealthState(Enum):
    """State machine states for FM health"""
    HEALTHY = "healthy"
    WARNING = "warning"
    DEGRADED = "degraded"
    CRITICAL = "critical"
    RECOVERING = "recovering"
    MAINTENANCE = "maintenance"


@dataclass
class GoldenTestCase:
    """A single test case in the golden evaluation dataset"""
    test_id: str
    intent: str                          # product_search, order_status, recommendation, faq
    difficulty: str                      # easy, medium, hard, adversarial
    query: str
    expected_answer: str
    required_facts: List[str]            # Facts that MUST appear in response
    prohibited_claims: List[str]         # Facts that must NOT appear (hallucination traps)
    acceptable_quality_range: Tuple[float, float] = (0.8, 1.0)


@dataclass
class TroubleshootingResult:
    """Result of running a single troubleshooting evaluation"""
    test_id: str
    timestamp: datetime
    query: str
    actual_response: str
    quality_score: float
    hallucination_detected: bool
    hallucinated_claims: List[str] = field(default_factory=list)
    reasoning_errors: List[str] = field(default_factory=list)
    consistency_score: float = 1.0
    failure_modes: List[FailureMode] = field(default_factory=list)
    root_cause: Optional[str] = None
    recommended_action: Optional[str] = None


@dataclass
class OutputDiffResult:
    """Result of comparing two FM responses to the same query"""
    query: str
    response_a: str
    response_b: str
    text_similarity: float               # 0-1 character-level Levenshtein
    semantic_similarity: float            # 0-1 embedding cosine similarity
    structured_diff: Dict[str, str] = field(default_factory=dict)  # field → change description
    drift_detected: bool = False
    drift_category: Optional[str] = None  # "factual", "stylistic", "structural"


@dataclass
class ReasoningTrace:
    """Extracted and validated reasoning path from an FM response"""
    response: str
    steps: List[str] = field(default_factory=list)
    grounding_scores: List[float] = field(default_factory=list)   # per-step grounding
    contradictions: List[str] = field(default_factory=list)
    conclusion_supported: bool = True
    overall_validity: float = 1.0

import hashlib
import json
from datetime import datetime
from typing import Dict, List, Optional, Tuple
import re


class FMTroubleshootingEngine:
    """
    Production engine for detecting FM-specific failure modes in MangaAssist.

    Integrates golden dataset evaluation, output diffing, reasoning validation,
    and consistency checking into a unified troubleshooting pipeline.
    """

    def __init__(self, config: dict):
        self.golden_dataset: List[dict] = []
        self.quality_threshold = config.get("quality_threshold", 0.85)
        self.hallucination_threshold = config.get("hallucination_threshold", 0.02)
        self.consistency_threshold = config.get("consistency_threshold", 0.90)
        self._response_history: Dict[str, List[dict]] = {}

    # ── Pillar 1: Golden Dataset Evaluation ────────────────────────────

    def evaluate_golden_dataset(self, invoke_fn) -> dict:
        """Run all golden test cases and produce an aggregate health report."""
        results = []
        failures = []

        for test_case in self.golden_dataset:
            response = invoke_fn(test_case["query"])

            score = self._score_response(test_case, response)
            hallucinated = self._detect_hallucinations(test_case, response)
            reasoning_ok = self._validate_reasoning(response)

            result = {
                "test_id": test_case["test_id"],
                "intent": test_case.get("intent", "unknown"),
                "difficulty": test_case.get("difficulty", "unknown"),
                "quality_score": score,
                "hallucination_detected": len(hallucinated) > 0,
                "hallucinated_claims": hallucinated,
                "reasoning_valid": reasoning_ok,
                "passed": (
                    score >= test_case.get("min_quality", 0.8)
                    and len(hallucinated) == 0
                ),
            }
            results.append(result)
            if not result["passed"]:
                failures.append(result)

        total = len(results)
        pass_rate = sum(1 for r in results if r["passed"]) / total if total else 0
        avg_quality = sum(r["quality_score"] for r in results) / total if total else 0
        hallucination_rate = (
            sum(1 for r in results if r["hallucination_detected"]) / total
            if total
            else 0
        )

        return {
            "total_tests": total,
            "pass_rate": round(pass_rate, 4),
            "avg_quality": round(avg_quality, 4),
            "hallucination_rate": round(hallucination_rate, 4),
            "failures": failures,
            "per_intent": self._aggregate_by_intent(results),
            "healthy": (
                pass_rate >= 0.95
                and hallucination_rate < self.hallucination_threshold
            ),
        }

    def _aggregate_by_intent(self, results: List[dict]) -> dict:
        """Break down results by intent for targeted troubleshooting."""
        by_intent: Dict[str, List[dict]] = {}
        for r in results:
            intent = r.get("intent", "unknown")
            by_intent.setdefault(intent, []).append(r)

        summary = {}
        for intent, items in by_intent.items():
            total = len(items)
            summary[intent] = {
                "total": total,
                "pass_rate": round(
                    sum(1 for i in items if i["passed"]) / total, 4
                ),
                "avg_quality": round(
                    sum(i["quality_score"] for i in items) / total, 4
                ),
                "hallucination_count": sum(
                    1 for i in items if i["hallucination_detected"]
                ),
            }
        return summary

    # ── Scoring ────────────────────────────────────────────────────────

    def _score_response(self, test_case: dict, response: str) -> float:
        """
        Multi-dimensional scoring:
          - Factual accuracy (0.6 weight): required facts present
          - Safety (0.2 weight): prohibited claims absent
          - Completeness (0.2 weight): response length in acceptable range
        """
        score = 0.0

        # Factual accuracy — required facts present
        required_facts = test_case.get("required_facts", [])
        if required_facts:
            found = sum(
                1 for fact in required_facts if fact.lower() in response.lower()
            )
            score += (found / len(required_facts)) * 0.6
        else:
            score += 0.6

        # Safety — prohibited claims absent
        prohibited = test_case.get("prohibited_claims", [])
        if prohibited:
            violations = sum(
                1 for claim in prohibited if claim.lower() in response.lower()
            )
            score += max(0, 0.2 * (1 - violations / len(prohibited)))
        else:
            score += 0.2

        # Completeness — response length
        response_len = len(response.split())
        if 20 <= response_len <= 500:
            score += 0.2
        elif response_len < 20:
            score += 0.1
        else:
            score += 0.05

        return round(min(score, 1.0), 3)

    # ── Pillar 1b: Hallucination Detection ─────────────────────────────

    def _detect_hallucinations(self, test_case: dict, response: str) -> List[str]:
        """Check response against prohibited claims (known hallucination traps)."""
        hallucinated = []
        for claim in test_case.get("prohibited_claims", []):
            if claim.lower() in response.lower():
                hallucinated.append(claim)
        return hallucinated

    # ── Pillar 3: Reasoning Validation ─────────────────────────────────

    def _validate_reasoning(self, response: str) -> bool:
        """
        Validate that the response contains valid reasoning:
        1. Reasoning indicators are present
        2. No self-contradictions detected
        """
        reasoning_indicators = [
            "because", "therefore", "since", "as a result",
            "this means", "which is why", "given that",
        ]
        has_reasoning = any(ind in response.lower() for ind in reasoning_indicators)
        contradictions = self._find_contradictions(response)
        return has_reasoning and len(contradictions) == 0

    def _find_contradictions(self, response: str) -> List[str]:
        """Detect self-contradictory statements within a single response."""
        contradictions = []
        sentences = [s.strip() for s in response.split(".") if s.strip()]

        negation_pairs = [
            ("is available", "is not available"),
            ("in stock", "out of stock"),
            ("is available", "is unavailable"),
            ("recommend", "do not recommend"),
            ("free shipping", "shipping fee"),
            ("currently active", "has been discontinued"),
        ]

        for pos, neg in negation_pairs:
            has_pos = any(pos in s.lower() for s in sentences)
            has_neg = any(neg in s.lower() for s in sentences)
            if has_pos and has_neg:
                contradictions.append(
                    f"Contradiction: '{pos}' and '{neg}' both present"
                )

        return contradictions

    # ── Pillar 2: Output Consistency Checking ──────────────────────────

    def check_consistency(self, query: str, response: str) -> dict:
        """Check if response is consistent with previous responses for same query."""
        query_hash = hashlib.md5(query.encode()).hexdigest()

        if query_hash not in self._response_history:
            self._response_history[query_hash] = []

        history = self._response_history[query_hash]
        consistency_scores = []

        for prev in history[-5:]:  # Compare against last 5 responses
            words_current = set(response.lower().split())
            words_prev = set(prev["response"].lower().split())
            if words_current or words_prev:
                overlap = len(words_current & words_prev) / max(
                    len(words_current | words_prev), 1
                )
                consistency_scores.append(overlap)

        # Store current response
        history.append({
            "response": response,
            "timestamp": datetime.utcnow().isoformat(),
        })
        if len(history) > 20:
            history.pop(0)

        avg_consistency = (
            sum(consistency_scores) / len(consistency_scores)
            if consistency_scores
            else 1.0
        )

        return {
            "consistency_score": round(avg_consistency, 3),
            "drift_detected": avg_consistency < self.consistency_threshold,
            "history_length": len(history),
        }

    # ── Health State Determination ─────────────────────────────────────

    def determine_health_state(self, eval_report: dict) -> str:
        """Map evaluation report metrics to FM health state."""
        quality = eval_report.get("avg_quality", 0)
        hall_rate = eval_report.get("hallucination_rate", 0)
        pass_rate = eval_report.get("pass_rate", 0)

        if quality < 0.6 or hall_rate > 0.10 or pass_rate < 0.70:
            return "CRITICAL"
        if quality < 0.75 or hall_rate > 0.05 or pass_rate < 0.85:
            return "DEGRADED"
        if quality < 0.85 or hall_rate > 0.02 or pass_rate < 0.95:
            return "WARNING"
        return "HEALTHY"

graph TD
    START[🚨 Quality Metric<br/>Below Threshold] --> TYPE{What type<br/>of failure?}

    TYPE -->|Hallucination| H1{Check Golden Dataset<br/>Which test cases failed?}
    H1 -->|Single intent| H2[Intent-specific issue<br/>Check RAG retrieval for that intent]
    H1 -->|Multiple intents| H3[Systemic issue<br/>Check model/prompt change]
    H2 --> H4[Fix: Update OpenSearch<br/>index for that intent]
    H3 --> H5[Fix: Rollback prompt<br/>or model version]

    TYPE -->|Reasoning Error| R1{Trace Reasoning Path<br/>Which step failed?}
    R1 -->|Grounding failure| R2[Step not supported<br/>by retrieved context]
    R1 -->|Contradiction| R3[Self-contradictory<br/>statements detected]
    R1 -->|Missing steps| R4[Incomplete reasoning<br/>chain]
    R2 --> R5[Fix: Improve RAG<br/>retrieval relevance]
    R3 --> R6[Fix: Add contradiction<br/>check to prompt]
    R4 --> R7[Fix: Update system prompt<br/>to require full reasoning]

    TYPE -->|Inconsistency| I1{Run Output Diff<br/>What changed?}
    I1 -->|Factual drift| I2[Different facts in<br/>repeated queries]
    I1 -->|Stylistic drift| I3[Tone/format changed<br/>but facts same]
    I1 -->|Structural drift| I4[Response structure<br/>changed significantly]
    I2 --> I5[Fix: Pin temperature=0<br/>for factual queries]
    I3 --> I6[Fix: Strengthen persona<br/>in system prompt]
    I4 --> I7[Fix: Add output format<br/>constraints to prompt]

    TYPE -->|Recent Change?| C1{Check Deployment<br/>Timeline}
    C1 -->|Prompt changed| C2[Rollback to<br/>previous prompt version]
    C1 -->|Model updated| C3[Rollback to<br/>previous model version]
    C1 -->|RAG index updated| C4[Reindex or rollback<br/>OpenSearch update]
    C1 -->|No recent change| C5[New failure mode<br/>Add to golden dataset]