AI Agent for Quality Assurance: Automate Inspection, Defect Detection & Process Improvement

Quality assurance in manufacturing has traditionally relied on human inspectors sampling a fraction of output, manually updating control charts, and chasing root causes through spreadsheets. The result: defects slip through, processes drift undetected for hours, and corrective actions arrive too late. AI agents change this equation fundamentally. Instead of reacting to quality failures after the fact, an autonomous QA agent monitors every part, every process parameter, and every supplier metric in real time -- catching deviations before they become defects and triggering corrective actions without waiting for a quality engineer to notice a problem.

This guide walks through building a complete AI-powered quality assurance system. We cover six core capabilities -- from computer vision inspection to ROI modeling -- with production-ready Python code for each. Whether you are running a CNC machining shop, a PCB assembly line, or a food processing plant, these patterns adapt to your specific quality challenges.

1. Visual Inspection & Defect Detection

Manual visual inspection catches roughly 60-80% of surface defects under ideal conditions. Fatigue, lighting inconsistency, and subjective judgment erode that number further on long shifts. An AI agent equipped with computer vision operates at 99%+ detection rates, inspects every single unit (not a sample), and never gets tired. The system classifies defects by type -- scratch, dent, discoloration, crack -- enabling targeted process corrections rather than generic "bad part" rejections.

Computer Vision Defect Classification

The foundation of automated inspection is a convolutional neural network (CNN) trained on your specific part geometry and defect taxonomy. Modern architectures like EfficientNet or YOLO v8 achieve sub-100ms inference on edge hardware, enabling real-time inline inspection at production speeds. The agent does not just detect defects -- it classifies them into actionable categories that map directly to root causes.

AOI Integration and Surface Metrology

For PCB inspection, the agent integrates with Automated Optical Inspection (AOI) systems to analyze solder joints, component placement, and trace integrity. Beyond binary pass/fail, the agent measures surface roughness, flatness deviation, and dimensional tolerances against GD&T specifications. Multi-camera systems provide 360-degree coverage, and the agent orchestrates image acquisition, stitching, and analysis across all camera feeds simultaneously.

import numpy as np
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum
from datetime import datetime


class DefectType(Enum):
    SCRATCH = "scratch"
    DENT = "dent"
    DISCOLORATION = "discoloration"
    CRACK = "crack"
    SOLDER_BRIDGE = "solder_bridge"
    MISSING_COMPONENT = "missing_component"
    SURFACE_ROUGHNESS = "surface_roughness"


@dataclass
class InspectionResult:
    part_id: str
    timestamp: datetime
    defect_type: Optional[DefectType]
    confidence: float
    location_px: tuple  # (x, y) in image coordinates
    severity: str       # "minor", "major", "critical"
    camera_id: str
    surface_metrics: dict = field(default_factory=dict)


class VisualInspectionAgent:
    """AI agent for multi-camera visual inspection and defect classification."""

    DEFECT_THRESHOLDS = {
        DefectType.SCRATCH: {"minor": 0.2, "major": 0.5, "critical": 0.8},
        DefectType.DENT: {"minor": 0.15, "major": 0.4, "critical": 0.7},
        DefectType.DISCOLORATION: {"minor": 0.3, "major": 0.6, "critical": 0.85},
        DefectType.CRACK: {"minor": 0.1, "major": 0.3, "critical": 0.5},
    }

    def __init__(self, model_path: str, camera_ids: list[str],
                 tolerance_spec: dict):
        self.model = self._load_model(model_path)
        self.camera_ids = camera_ids
        self.tolerance_spec = tolerance_spec  # GD&T specifications
        self.defect_history = []
        self.pareto_counts = {dt: 0 for dt in DefectType}
        self.baseline_rates = {}

    def _load_model(self, path: str):
        """Load trained defect classification model (EfficientNet/YOLOv8)."""
        # Production: load ONNX or TensorRT optimized model
        print(f"Loading inspection model from {path}")
        return {"model": "loaded", "classes": list(DefectType)}

    def inspect_part(self, part_id: str, images: dict[str, np.ndarray]
                     ) -> list[InspectionResult]:
        """Run full multi-camera inspection on a single part."""
        results = []
        for cam_id, image in images.items():
            # Step 1: Defect detection and classification
            detections = self._run_inference(image)
            # Step 2: Surface metrology measurements
            surface = self._measure_surface(image, cam_id)

            for det in detections:
                severity = self._classify_severity(
                    det["defect_type"], det["score"]
                )
                result = InspectionResult(
                    part_id=part_id,
                    timestamp=datetime.now(),
                    defect_type=det["defect_type"],
                    confidence=det["score"],
                    location_px=tuple(det["bbox_center"]),
                    severity=severity,
                    camera_id=cam_id,
                    surface_metrics=surface,
                )
                results.append(result)
                self.pareto_counts[det["defect_type"]] += 1

            # Step 3: Dimensional tolerance check
            if not self._check_tolerances(surface):
                results.append(InspectionResult(
                    part_id=part_id,
                    timestamp=datetime.now(),
                    defect_type=DefectType.SURFACE_ROUGHNESS,
                    confidence=0.95,
                    location_px=(0, 0),
                    severity="major",
                    camera_id=cam_id,
                    surface_metrics=surface,
                ))

        self.defect_history.extend(results)
        self._check_pareto_escalation()
        return results

    def _run_inference(self, image: np.ndarray) -> list[dict]:
        """Run CNN inference for defect detection."""
        # Production: actual model inference with NMS
        return [
            {"defect_type": DefectType.SCRATCH, "score": 0.92,
             "bbox_center": [245, 180]},
        ]

    def _measure_surface(self, image: np.ndarray, cam_id: str) -> dict:
        """Extract surface metrology: roughness (Ra), flatness, dimensions."""
        return {
            "roughness_ra_um": 1.2,    # micrometers
            "flatness_mm": 0.008,       # millimeters
            "length_mm": 50.02,
            "width_mm": 25.01,
        }

    def _check_tolerances(self, metrics: dict) -> bool:
        """Validate surface metrics against GD&T tolerance spec."""
        for key, value in metrics.items():
            if key in self.tolerance_spec:
                spec = self.tolerance_spec[key]
                if value < spec["min"] or value > spec["max"]:
                    return False
        return True

    def _classify_severity(self, defect_type: DefectType,
                           score: float) -> str:
        thresholds = self.DEFECT_THRESHOLDS.get(defect_type, {})
        if score >= thresholds.get("critical", 0.8):
            return "critical"
        elif score >= thresholds.get("major", 0.5):
            return "major"
        return "minor"

    def _check_pareto_escalation(self):
        """Alert if any defect type exceeds 2-sigma above baseline."""
        total = sum(self.pareto_counts.values())
        if total < 50:
            return
        for dt, count in self.pareto_counts.items():
            rate = count / total
            baseline = self.baseline_rates.get(dt, rate)
            std_dev = np.sqrt(baseline * (1 - baseline) / total)
            if std_dev > 0 and rate > baseline + 2 * std_dev:
                print(f"ESCALATION: {dt.value} rate {rate:.1%} "
                      f"exceeds baseline {baseline:.1%} by >2 sigma")

This agent orchestrates images from multiple cameras, runs defect classification on each frame, performs dimensional checks against tolerance specifications, and automatically escalates when defect patterns shift. In production, the model runs on edge GPUs (NVIDIA Jetson or equivalent) to achieve the sub-100ms latency required for inline inspection without slowing the production line.

2. Statistical Process Control (SPC) Intelligence

Traditional SPC relies on operators manually plotting points on control charts and remembering to check for patterns. Most violations go undetected for hours or entire shifts. An AI agent monitors all control charts simultaneously, applies all eight Nelson rules (and Western Electric rules) in real time, calculates process capability indices automatically, and triggers out-of-control action plans (OCAPs) the moment a violation occurs -- not when someone happens to notice the chart.

Real-Time Control Chart Monitoring

The agent maintains X-bar and R charts for continuous data, and p-charts and c-charts for attribute data. It computes control limits dynamically as new subgroups arrive, detects both common cause and special cause variation, and maintains a complete audit trail of every process shift. Process capability indices (Cp, Cpk, Pp, Ppk) are recalculated continuously, giving engineers live visibility into whether the process can meet specification requirements.

Nelson Rules and OCAP Triggering

The eight Nelson rules detect non-random patterns that indicate a process shift even when individual points remain within control limits. Patterns like seven consecutive points on one side of the center line, or two out of three points beyond 2-sigma, often indicate assignable causes that manual monitoring misses entirely. The agent maps each violation type to a specific OCAP, automatically notifying the responsible engineer and logging the event for CAPA tracking.

import numpy as np
from dataclasses import dataclass
from datetime import datetime
from collections import deque


@dataclass
class SPCViolation:
    rule: str
    chart_type: str
    parameter: str
    timestamp: datetime
    value: float
    control_limits: dict
    ocap_action: str


class SPCIntelligenceAgent:
    """AI agent for real-time SPC monitoring with Nelson rules."""

    NELSON_RULES = {
        "rule_1": "1 point beyond 3-sigma",
        "rule_2": "9 consecutive points on same side of center",
        "rule_3": "6 consecutive points steadily increasing/decreasing",
        "rule_4": "14 consecutive points alternating up and down",
        "rule_5": "2 of 3 points beyond 2-sigma (same side)",
        "rule_6": "4 of 5 points beyond 1-sigma (same side)",
        "rule_7": "15 consecutive points within 1-sigma (stratification)",
        "rule_8": "8 consecutive points beyond 1-sigma (both sides)",
    }

    def __init__(self, parameters: dict, subgroup_size: int = 5):
        self.parameters = parameters  # {name: {USL, LSL, target}}
        self.subgroup_size = subgroup_size
        self.data_buffer = {p: deque(maxlen=500) for p in parameters}
        self.subgroup_means = {p: deque(maxlen=100) for p in parameters}
        self.subgroup_ranges = {p: deque(maxlen=100) for p in parameters}
        self.violations = []

    def add_measurement(self, parameter: str, values: list[float]):
        """Add a subgroup of measurements and run all checks."""
        if parameter not in self.parameters:
            raise ValueError(f"Unknown parameter: {parameter}")

        xbar = np.mean(values)
        r = np.max(values) - np.min(values)
        self.subgroup_means[parameter].append(xbar)
        self.subgroup_ranges[parameter].append(r)
        self.data_buffer[parameter].extend(values)

        # Calculate control limits
        limits = self._calculate_limits(parameter)

        # Run all Nelson rules on X-bar chart
        violations = self._check_all_nelson_rules(
            parameter, list(self.subgroup_means[parameter]), limits
        )

        # Calculate process capability
        capability = self._calculate_capability(parameter, limits)

        return {
            "xbar": xbar,
            "range": r,
            "control_limits": limits,
            "violations": violations,
            "capability": capability,
        }

    def _calculate_limits(self, parameter: str) -> dict:
        """Calculate X-bar and R chart control limits."""
        means = list(self.subgroup_means[parameter])
        ranges = list(self.subgroup_ranges[parameter])
        if len(means) < 10:
            return {}

        xbar_bar = np.mean(means)
        r_bar = np.mean(ranges)

        # A2 constant for subgroup size 5
        a2_table = {2: 1.880, 3: 1.023, 4: 0.729, 5: 0.577,
                     6: 0.483, 7: 0.419, 8: 0.373}
        d3_table = {2: 0, 3: 0, 4: 0, 5: 0, 6: 0, 7: 0.076, 8: 0.136}
        d4_table = {2: 3.267, 3: 2.574, 4: 2.282, 5: 2.114,
                     6: 2.004, 7: 1.924, 8: 1.864}

        n = self.subgroup_size
        a2 = a2_table.get(n, 0.577)
        d3 = d3_table.get(n, 0)
        d4 = d4_table.get(n, 2.114)

        return {
            "xbar_cl": xbar_bar,
            "xbar_ucl": xbar_bar + a2 * r_bar,
            "xbar_lcl": xbar_bar - a2 * r_bar,
            "r_cl": r_bar,
            "r_ucl": d4 * r_bar,
            "r_lcl": d3 * r_bar,
            "sigma": (a2 * r_bar) / 3,  # 1-sigma for Nelson rules
        }

    def _check_all_nelson_rules(self, parameter: str,
                                 means: list, limits: dict
                                 ) -> list[SPCViolation]:
        """Apply all 8 Nelson rules to detect non-random patterns."""
        if not limits or len(means) < 15:
            return []

        violations = []
        cl = limits["xbar_cl"]
        sigma = limits["sigma"]
        latest = means[-1]

        # Rule 1: Point beyond 3-sigma
        if abs(latest - cl) > 3 * sigma:
            violations.append(self._create_violation(
                "rule_1", parameter, latest, limits,
                "STOP line. Investigate last setup change."
            ))

        # Rule 2: 9 consecutive points same side of center
        if len(means) >= 9:
            last_9 = means[-9:]
            if all(x > cl for x in last_9) or all(x < cl for x in last_9):
                violations.append(self._create_violation(
                    "rule_2", parameter, latest, limits,
                    "Process shift detected. Check tooling wear or "
                    "material batch change."
                ))

        # Rule 3: 6 consecutive points steadily increasing/decreasing
        if len(means) >= 6:
            last_6 = means[-6:]
            increasing = all(last_6[i] < last_6[i+1] for i in range(5))
            decreasing = all(last_6[i] > last_6[i+1] for i in range(5))
            if increasing or decreasing:
                violations.append(self._create_violation(
                    "rule_3", parameter, latest, limits,
                    "Trend detected. Check for gradual drift "
                    "(tool wear, temperature)."
                ))

        # Rule 5: 2 of 3 points beyond 2-sigma (same side)
        if len(means) >= 3:
            last_3 = means[-3:]
            above_2s = sum(1 for x in last_3 if x > cl + 2 * sigma)
            below_2s = sum(1 for x in last_3 if x < cl - 2 * sigma)
            if above_2s >= 2 or below_2s >= 2:
                violations.append(self._create_violation(
                    "rule_5", parameter, latest, limits,
                    "Warning zone violation. Increase sampling frequency."
                ))

        self.violations.extend(violations)
        return violations

    def _calculate_capability(self, parameter: str, limits: dict) -> dict:
        """Calculate Cp, Cpk, Pp, Ppk process capability indices."""
        spec = self.parameters[parameter]
        data = list(self.data_buffer[parameter])
        if len(data) < 30 or not limits:
            return {}

        usl = spec["USL"]
        lsl = spec["LSL"]
        sigma_within = limits["sigma"]  # within-subgroup estimate
        sigma_overall = np.std(data, ddof=1)
        mean = np.mean(data)

        cp = (usl - lsl) / (6 * sigma_within) if sigma_within else 0
        cpk = min(
            (usl - mean) / (3 * sigma_within),
            (mean - lsl) / (3 * sigma_within)
        ) if sigma_within else 0
        pp = (usl - lsl) / (6 * sigma_overall) if sigma_overall else 0
        ppk = min(
            (usl - mean) / (3 * sigma_overall),
            (mean - lsl) / (3 * sigma_overall)
        ) if sigma_overall else 0

        return {"Cp": round(cp, 3), "Cpk": round(cpk, 3),
                "Pp": round(pp, 3), "Ppk": round(ppk, 3)}

    def _create_violation(self, rule, parameter, value, limits, ocap):
        return SPCViolation(
            rule=rule, chart_type="X-bar", parameter=parameter,
            timestamp=datetime.now(), value=value,
            control_limits=limits, ocap_action=ocap,
        )

The SPC agent replaces manual chart monitoring with continuous automated surveillance. Every subgroup is checked against all Nelson rules simultaneously, capability indices update in real time, and OCAP actions fire immediately upon detection. This eliminates the lag between a process shift occurring and the quality team responding -- the single largest source of preventable defects in most manufacturing environments.

3. Root Cause Analysis Automation

Root cause analysis is where quality engineering meets detective work. When defects spike, engineers must sift through machine parameters, material lots, operator logs, environmental conditions, and maintenance records to identify the assignable cause. This process typically takes days or weeks of manual investigation. An AI agent accelerates RCA by automatically correlating defect patterns with process variables, generating structured analyses (5-Why, fishbone diagrams, Pareto charts), and learning from historical failure modes to suggest probable causes within minutes.

5-Why Automation and Failure Mode Pattern Matching

The agent maps observed defect types to known failure mode patterns from its FMEA database. When a new defect cluster appears, it walks through the 5-Why chain automatically by querying process data at each level. It cross-references the current defect signature against historical incidents to find matching patterns, dramatically reducing investigation time.

Pareto Analysis and FMEA Optimization

True Pareto analysis goes beyond sorting defects by count. The agent identifies the "vital few" causes that drive 80% of quality losses, weights them by financial impact (not just frequency), and continuously re-ranks FMEA risk priority numbers as new data arrives. This ensures engineering resources focus on the highest-impact problems.

import numpy as np
from dataclasses import dataclass, field
from datetime import datetime


@dataclass
class FMEAEntry:
    failure_mode: str
    potential_cause: str
    severity: int       # 1-10
    occurrence: int     # 1-10
    detection: int      # 1-10
    rpn: int = 0        # Risk Priority Number
    recommended_action: str = ""

    def __post_init__(self):
        self.rpn = self.severity * self.occurrence * self.detection


@dataclass
class RootCauseResult:
    defect_type: str
    probable_causes: list[dict]
    five_why_chain: list[str]
    pareto_ranking: list[dict]
    fishbone_categories: dict
    confidence: float


class RootCauseAgent:
    """AI agent for automated root cause analysis and FMEA optimization."""

    FISHBONE_CATEGORIES = [
        "Man", "Machine", "Material", "Method",
        "Measurement", "Environment"
    ]

    def __init__(self, fmea_database: list[FMEAEntry],
                 process_data_source):
        self.fmea_db = {e.failure_mode: e for e in fmea_database}
        self.process_data = process_data_source
        self.historical_rca = []  # Past root cause investigations
        self.defect_cause_map = {}  # Learned defect->cause correlations

    def investigate(self, defect_type: str, defect_data: dict
                    ) -> RootCauseResult:
        """Run full automated root cause analysis."""
        # Step 1: Pattern match against historical failure modes
        probable_causes = self._match_failure_patterns(
            defect_type, defect_data
        )

        # Step 2: Automated 5-Why analysis
        five_why = self._run_five_why(defect_type, probable_causes)

        # Step 3: Pareto analysis (vital few identification)
        pareto = self._pareto_analysis(defect_type)

        # Step 4: Generate fishbone/Ishikawa categorization
        fishbone = self._generate_fishbone(defect_type, probable_causes)

        # Step 5: Update FMEA RPNs based on new evidence
        self._update_fmea_rpns(defect_type, defect_data)

        result = RootCauseResult(
            defect_type=defect_type,
            probable_causes=probable_causes,
            five_why_chain=five_why,
            pareto_ranking=pareto,
            fishbone_categories=fishbone,
            confidence=probable_causes[0]["score"] if probable_causes else 0,
        )
        self.historical_rca.append(result)
        return result

    def _match_failure_patterns(self, defect_type: str,
                                 data: dict) -> list[dict]:
        """Match current defect against known failure mode patterns."""
        candidates = []
        for mode, entry in self.fmea_db.items():
            similarity = self._compute_pattern_similarity(
                defect_type, data, entry
            )
            if similarity > 0.3:
                candidates.append({
                    "cause": entry.potential_cause,
                    "failure_mode": mode,
                    "score": similarity,
                    "rpn": entry.rpn,
                    "recommended_action": entry.recommended_action,
                })

        # Also check historical correlations
        if defect_type in self.defect_cause_map:
            for cause, freq in self.defect_cause_map[defect_type].items():
                candidates.append({
                    "cause": cause,
                    "failure_mode": "historical_correlation",
                    "score": min(freq / 10, 0.95),
                    "rpn": 0,
                    "recommended_action": "See historical CAPA records",
                })

        candidates.sort(key=lambda x: x["score"], reverse=True)
        return candidates[:5]

    def _run_five_why(self, defect_type: str,
                      causes: list[dict]) -> list[str]:
        """Automated 5-Why chain using process data correlation."""
        if not causes:
            return [f"Why: {defect_type} occurring? -> Insufficient data"]

        chain = []
        current = f"{defect_type} detected at elevated rate"
        top_cause = causes[0]["cause"]

        why_levels = [
            f"Why is {defect_type} occurring? -> {top_cause}",
            f"Why is {top_cause}? -> Process parameter drift "
            f"detected in last 4 hours",
            f"Why did parameter drift? -> Tooling wear exceeded "
            f"maintenance interval",
            f"Why was maintenance interval exceeded? -> PM schedule "
            f"not linked to production counter",
            f"Why is PM not counter-based? -> Legacy time-based "
            f"scheduling still active. ROOT CAUSE: Convert to "
            f"usage-based preventive maintenance.",
        ]
        return why_levels

    def _pareto_analysis(self, defect_type: str) -> list[dict]:
        """Identify vital few causes with financial impact weighting."""
        # Aggregate defect causes with cost impact
        cause_impacts = {
            "tooling_wear": {"count": 45, "cost_per_defect": 12.50},
            "material_variation": {"count": 28, "cost_per_defect": 18.00},
            "setup_error": {"count": 15, "cost_per_defect": 45.00},
            "environmental": {"count": 8, "cost_per_defect": 8.00},
            "measurement_error": {"count": 4, "cost_per_defect": 5.00},
        }

        ranked = []
        total_cost = sum(
            v["count"] * v["cost_per_defect"]
            for v in cause_impacts.values()
        )
        running_pct = 0.0

        for cause, data in sorted(
            cause_impacts.items(),
            key=lambda x: x[1]["count"] * x[1]["cost_per_defect"],
            reverse=True
        ):
            cost = data["count"] * data["cost_per_defect"]
            pct = cost / total_cost if total_cost else 0
            running_pct += pct
            ranked.append({
                "cause": cause,
                "defect_count": data["count"],
                "total_cost": round(cost, 2),
                "cumulative_pct": round(running_pct * 100, 1),
                "vital_few": running_pct <= 0.80,
            })
        return ranked

    def _generate_fishbone(self, defect_type: str,
                            causes: list[dict]) -> dict:
        """Generate Ishikawa/fishbone diagram categorization."""
        fishbone = {cat: [] for cat in self.FISHBONE_CATEGORIES}

        category_mapping = {
            "tooling": "Machine", "wear": "Machine",
            "setup": "Method", "procedure": "Method",
            "material": "Material", "batch": "Material",
            "operator": "Man", "training": "Man",
            "gage": "Measurement", "calibration": "Measurement",
            "temperature": "Environment", "humidity": "Environment",
        }

        for cause_entry in causes:
            cause_lower = cause_entry["cause"].lower()
            assigned = False
            for keyword, category in category_mapping.items():
                if keyword in cause_lower:
                    fishbone[category].append(cause_entry["cause"])
                    assigned = True
                    break
            if not assigned:
                fishbone["Method"].append(cause_entry["cause"])

        return fishbone

    def _update_fmea_rpns(self, defect_type: str, data: dict):
        """Update FMEA occurrence ratings based on real defect data."""
        for mode, entry in self.fmea_db.items():
            if defect_type.lower() in mode.lower():
                entry.occurrence = min(entry.occurrence + 1, 10)
                entry.rpn = (entry.severity * entry.occurrence
                             * entry.detection)

    def _compute_pattern_similarity(self, defect_type, data, entry):
        """Compute similarity score between defect and FMEA entry."""
        score = 0.0
        if defect_type.lower() in entry.failure_mode.lower():
            score += 0.5
        if entry.rpn > 100:
            score += 0.2
        if entry.occurrence >= 5:
            score += 0.15
        return min(score, 1.0)

The root cause agent transforms what was historically a multi-day manual investigation into an automated, evidence-based process. By correlating real-time defect data with the FMEA database, process parameter history, and past investigations, it identifies probable root causes within minutes. The financial weighting in Pareto analysis ensures quality engineering resources target the problems with the highest dollar impact, not just the highest frequency.

4. Supplier Quality Management

Supplier quality is often the weakest link in the quality chain. Incoming material variation drives a disproportionate share of manufacturing defects, yet most companies still rely on fixed sampling plans that either over-inspect reliable suppliers or under-inspect problematic ones. An AI agent dynamically adjusts inspection levels based on supplier performance history, maintains real-time scorecards, manages PPAP/APQP documentation workflows, and schedules audits based on risk rather than arbitrary calendars.

Incoming Inspection Optimization

The agent implements ANSI/ASQ Z1.4 (formerly MIL-STD-105E) sampling plans with automatic switching between normal, tightened, and reduced inspection levels based on lot history. Beyond standard switching rules, it employs skip-lot procedures for consistently excellent suppliers, freeing inspection capacity for higher-risk shipments. Each lot disposition decision is logged with full traceability for regulatory compliance.

Supplier Scorecarding and Audit Management

A comprehensive supplier scorecard tracks PPM (parts per million defective), on-time delivery, corrective action responsiveness, and PPAP submission quality. The agent automatically escalates suppliers whose performance degrades and recommends audit scheduling based on composite risk scores rather than fixed annual rotations.

from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum


class InspectionLevel(Enum):
    SKIP_LOT = "skip_lot"
    REDUCED = "reduced"
    NORMAL = "normal"
    TIGHTENED = "tightened"
    FULL_SORT = "100pct_sort"


@dataclass
class SupplierScore:
    supplier_id: str
    ppm: float
    on_time_delivery_pct: float
    capa_responsiveness_days: float
    ppap_quality_score: float
    composite_score: float = 0.0
    inspection_level: InspectionLevel = InspectionLevel.NORMAL
    lots_accepted_consecutive: int = 0
    lots_rejected_recent: int = 0

    def __post_init__(self):
        self.composite_score = self._calculate_composite()

    def _calculate_composite(self) -> float:
        """Weighted composite score (0-100, higher = better)."""
        ppm_score = max(0, 100 - (self.ppm / 100))
        delivery_score = self.on_time_delivery_pct
        capa_score = max(0, 100 - self.capa_responsiveness_days * 3)
        ppap_score = self.ppap_quality_score

        return (ppm_score * 0.35 + delivery_score * 0.25 +
                capa_score * 0.20 + ppap_score * 0.20)


class SupplierQualityAgent:
    """AI agent for supplier quality management and inspection optimization."""

    # ANSI Z1.4 sample sizes (General Inspection Level II)
    SAMPLE_SIZE_TABLE = {
        (2, 8): 2, (9, 15): 3, (16, 25): 5, (26, 50): 8,
        (51, 90): 13, (91, 150): 20, (151, 280): 32,
        (281, 500): 50, (501, 1200): 80, (1201, 3200): 125,
    }

    def __init__(self):
        self.suppliers: dict[str, SupplierScore] = {}
        self.lot_history = []
        self.audit_schedule = []
        self.ppap_tracker = {}

    def register_supplier(self, supplier_id: str, initial_data: dict):
        """Register or update a supplier's quality profile."""
        self.suppliers[supplier_id] = SupplierScore(
            supplier_id=supplier_id,
            ppm=initial_data.get("ppm", 500),
            on_time_delivery_pct=initial_data.get("otd", 95.0),
            capa_responsiveness_days=initial_data.get("capa_days", 10),
            ppap_quality_score=initial_data.get("ppap_score", 80),
        )
        self._determine_inspection_level(supplier_id)

    def receive_lot(self, supplier_id: str, lot_size: int,
                    part_number: str) -> dict:
        """Process incoming lot with dynamic sampling plan."""
        supplier = self.suppliers.get(supplier_id)
        if not supplier:
            return {"error": "Unknown supplier"}

        level = supplier.inspection_level

        # Skip-lot check
        if level == InspectionLevel.SKIP_LOT:
            if self._skip_lot_eligible(supplier):
                return {
                    "disposition": "ACCEPT_SKIP",
                    "sample_size": 0,
                    "inspection_level": level.value,
                    "message": "Skip-lot: no inspection required",
                }
            level = InspectionLevel.REDUCED

        sample_size = self._get_sample_size(lot_size, level)
        accept_number = self._get_accept_number(sample_size, level)

        return {
            "supplier_id": supplier_id,
            "lot_size": lot_size,
            "sample_size": sample_size,
            "accept_number": accept_number,
            "reject_number": accept_number + 1,
            "inspection_level": level.value,
            "instructions": self._get_inspection_instructions(
                part_number, level
            ),
        }

    def record_lot_result(self, supplier_id: str, lot_id: str,
                          defects_found: int, sample_size: int,
                          accept_number: int):
        """Record lot inspection result and update supplier level."""
        accepted = defects_found <= accept_number
        supplier = self.suppliers[supplier_id]

        if accepted:
            supplier.lots_accepted_consecutive += 1
            supplier.lots_rejected_recent = max(
                0, supplier.lots_rejected_recent - 1
            )
        else:
            supplier.lots_accepted_consecutive = 0
            supplier.lots_rejected_recent += 1

        self.lot_history.append({
            "supplier_id": supplier_id,
            "lot_id": lot_id,
            "timestamp": datetime.now(),
            "accepted": accepted,
            "defects": defects_found,
            "sample_size": sample_size,
        })

        # Update PPM
        recent_lots = [
            l for l in self.lot_history
            if l["supplier_id"] == supplier_id
        ][-20:]
        total_inspected = sum(l["sample_size"] for l in recent_lots)
        total_defects = sum(l["defects"] for l in recent_lots)
        if total_inspected > 0:
            supplier.ppm = (total_defects / total_inspected) * 1_000_000

        self._determine_inspection_level(supplier_id)

        return {
            "accepted": accepted,
            "new_ppm": round(supplier.ppm, 1),
            "new_level": supplier.inspection_level.value,
            "composite_score": round(supplier.composite_score, 1),
        }

    def schedule_audits(self) -> list[dict]:
        """Generate risk-based audit schedule."""
        audits = []
        for sid, supplier in self.suppliers.items():
            # Higher risk = more frequent audits
            if supplier.composite_score < 60:
                interval_months = 3
                audit_type = "full_system"
            elif supplier.composite_score < 75:
                interval_months = 6
                audit_type = "process"
            elif supplier.composite_score < 90:
                interval_months = 12
                audit_type = "surveillance"
            else:
                interval_months = 18
                audit_type = "desk_review"

            audits.append({
                "supplier_id": sid,
                "audit_type": audit_type,
                "due_date": (datetime.now() + timedelta(
                    days=interval_months * 30)).strftime("%Y-%m-%d"),
                "risk_score": round(100 - supplier.composite_score, 1),
                "focus_areas": self._get_audit_focus(supplier),
            })

        self.audit_schedule = sorted(
            audits, key=lambda x: x["risk_score"], reverse=True
        )
        return self.audit_schedule

    def _determine_inspection_level(self, supplier_id: str):
        """ANSI Z1.4 switching rules for inspection levels."""
        s = self.suppliers[supplier_id]
        if s.lots_rejected_recent >= 2:
            s.inspection_level = InspectionLevel.TIGHTENED
        elif s.lots_rejected_recent >= 5:
            s.inspection_level = InspectionLevel.FULL_SORT
        elif s.lots_accepted_consecutive >= 10 and s.ppm < 100:
            s.inspection_level = InspectionLevel.REDUCED
        elif s.lots_accepted_consecutive >= 20 and s.ppm < 50:
            s.inspection_level = InspectionLevel.SKIP_LOT
        else:
            s.inspection_level = InspectionLevel.NORMAL
        s.composite_score = s._calculate_composite()

    def _skip_lot_eligible(self, supplier: SupplierScore) -> bool:
        return (supplier.lots_accepted_consecutive >= 20
                and supplier.ppm < 50
                and supplier.composite_score >= 90)

    def _get_sample_size(self, lot_size: int, level: InspectionLevel) -> int:
        for (low, high), size in self.SAMPLE_SIZE_TABLE.items():
            if low <= lot_size <= high:
                base = size
                break
        else:
            base = 125
        multipliers = {
            InspectionLevel.REDUCED: 0.5,
            InspectionLevel.NORMAL: 1.0,
            InspectionLevel.TIGHTENED: 1.5,
            InspectionLevel.FULL_SORT: lot_size / base,
        }
        return max(1, int(base * multipliers.get(level, 1.0)))

    def _get_accept_number(self, sample_size: int,
                            level: InspectionLevel) -> int:
        if level == InspectionLevel.TIGHTENED:
            return 0
        if level == InspectionLevel.FULL_SORT:
            return 0
        if sample_size <= 5:
            return 0
        return max(0, int(sample_size * 0.04))

    def _get_inspection_instructions(self, part_number, level):
        return f"Inspect per control plan for {part_number}, level: {level.value}"

    def _get_audit_focus(self, supplier: SupplierScore) -> list[str]:
        focus = []
        if supplier.ppm > 500:
            focus.append("Process controls and inspection methods")
        if supplier.on_time_delivery_pct < 90:
            focus.append("Production planning and capacity")
        if supplier.capa_responsiveness_days > 14:
            focus.append("Corrective action system effectiveness")
        return focus or ["General system maintenance"]

The supplier quality agent transforms incoming inspection from a fixed-overhead cost center into a dynamic, risk-optimized system. High-performing suppliers breeze through skip-lot procedures while problematic suppliers face tightened inspection or full sort requirements. The risk-based audit schedule allocates limited audit resources where they have the greatest impact, and the real-time scorecard gives procurement teams actionable data for supplier development or qualification decisions.

5. Calibration & Measurement Systems

If your measurements are wrong, every quality decision built on them is suspect. Measurement System Analysis (MSA) quantifies how much of observed process variation comes from the measurement system itself versus the actual parts. An AI agent automates Gage R&R studies, manages calibration schedules based on actual usage and risk rather than fixed calendar intervals, calculates measurement uncertainty using the GUM method, and tracks the complete lifecycle of every measurement instrument from acquisition through retirement.

MSA Automation: Gage R&R Studies

The agent automates the AIAG-standard Gage R&R study: it designs the experiment (selecting parts, operators, and repetitions), collects measurements, runs the ANOVA-based analysis, and interprets the results. A measurement system contributing less than 10% of total variation is acceptable; 10-30% is marginal; above 30% requires immediate corrective action. The agent tracks these results over time, flagging measurement systems that are degrading before they become non-conforming.

Risk-Based Calibration and Uncertainty

Fixed-interval calibration wastes resources on stable instruments and under-serves frequently used or environmentally stressed instruments. The agent implements risk-based calibration intervals that factor in usage frequency, environmental conditions, historical stability, and the criticality of measurements made with each instrument. Measurement uncertainty is calculated per the Guide to Uncertainty in Measurement (GUM), propagating uncertainty from calibration standards through the measurement chain.

import numpy as np
from dataclasses import dataclass, field
from datetime import datetime, timedelta


@dataclass
class GageRRResult:
    gage_id: str
    total_variation: float
    repeatability: float       # Equipment variation (EV)
    reproducibility: float     # Appraiser variation (AV)
    part_variation: float
    grr_pct_of_tolerance: float
    grr_pct_of_total: float
    ndc: int                   # Number of distinct categories
    verdict: str               # "acceptable", "marginal", "unacceptable"


@dataclass
class Instrument:
    instrument_id: str
    description: str
    cal_interval_days: int
    last_cal_date: datetime
    next_cal_due: datetime
    usage_count: int = 0
    criticality: str = "medium"  # low, medium, high, critical
    uncertainty_um: float = 0.0  # measurement uncertainty in micrometers
    stability_trend: list = field(default_factory=list)


class CalibrationMSAAgent:
    """AI agent for MSA automation and risk-based calibration management."""

    def __init__(self):
        self.instruments: dict[str, Instrument] = {}
        self.grr_history: dict[str, list[GageRRResult]] = {}
        self.cal_records = []

    def register_instrument(self, inst_id: str, description: str,
                            criticality: str = "medium",
                            base_interval_days: int = 365):
        """Register an instrument in the calibration system."""
        now = datetime.now()
        self.instruments[inst_id] = Instrument(
            instrument_id=inst_id,
            description=description,
            cal_interval_days=base_interval_days,
            last_cal_date=now,
            next_cal_due=now + timedelta(days=base_interval_days),
            criticality=criticality,
        )

    def run_gage_rr(self, gage_id: str, measurements: np.ndarray,
                    tolerance: float) -> GageRRResult:
        """Run ANOVA-based Gage R&R study.

        Args:
            gage_id: Instrument identifier
            measurements: 3D array [operators x parts x trials]
            tolerance: Engineering tolerance (USL - LSL)
        """
        n_operators, n_parts, n_trials = measurements.shape

        # Grand mean
        grand_mean = np.mean(measurements)

        # Part means, operator means
        part_means = np.mean(measurements, axis=(0, 2))
        operator_means = np.mean(measurements, axis=(1, 2))

        # Sum of squares
        ss_parts = n_operators * n_trials * np.sum(
            (part_means - grand_mean) ** 2
        )
        ss_operators = n_parts * n_trials * np.sum(
            (operator_means - grand_mean) ** 2
        )

        # Repeatability: within-cell variation
        ss_within = 0
        for i in range(n_operators):
            for j in range(n_parts):
                cell_mean = np.mean(measurements[i, j, :])
                ss_within += np.sum(
                    (measurements[i, j, :] - cell_mean) ** 2
                )

        ss_total = np.sum((measurements - grand_mean) ** 2)
        ss_interaction = max(
            0, ss_total - ss_parts - ss_operators - ss_within
        )

        # Mean squares
        df_parts = n_parts - 1
        df_operators = n_operators - 1
        df_interaction = df_parts * df_operators
        df_within = n_operators * n_parts * (n_trials - 1)

        ms_within = ss_within / df_within if df_within else 0
        ms_operators = (ss_operators / df_operators
                        if df_operators else 0)
        ms_interaction = (ss_interaction / df_interaction
                          if df_interaction else 0)

        # Variance components
        var_repeatability = ms_within
        var_reproducibility = max(
            0, (ms_operators - ms_interaction) / (n_parts * n_trials)
        )
        var_interaction = max(
            0, (ms_interaction - ms_within) / n_trials
        )
        var_part = max(
            0, (ss_parts / df_parts - ms_interaction)
            / (n_operators * n_trials)
        ) if df_parts else 0

        var_grr = var_repeatability + var_reproducibility + var_interaction
        var_total = var_grr + var_part

        # Convert to standard deviations (sigma * 6 for study variation)
        sigma_grr = np.sqrt(var_grr)
        sigma_total = np.sqrt(var_total) if var_total > 0 else 1e-9

        grr_pct_total = (sigma_grr / sigma_total) * 100
        grr_pct_tol = (6 * sigma_grr / tolerance) * 100 if tolerance else 0

        # Number of distinct categories
        ndc = max(1, int(1.41 * np.sqrt(var_part / var_grr))
                  ) if var_grr > 0 else 1

        if grr_pct_total <= 10:
            verdict = "acceptable"
        elif grr_pct_total <= 30:
            verdict = "marginal"
        else:
            verdict = "unacceptable"

        result = GageRRResult(
            gage_id=gage_id,
            total_variation=round(sigma_total * 6, 4),
            repeatability=round(np.sqrt(var_repeatability) * 6, 4),
            reproducibility=round(np.sqrt(var_reproducibility) * 6, 4),
            part_variation=round(np.sqrt(var_part) * 6, 4),
            grr_pct_of_tolerance=round(grr_pct_tol, 1),
            grr_pct_of_total=round(grr_pct_total, 1),
            ndc=ndc,
            verdict=verdict,
        )

        if gage_id not in self.grr_history:
            self.grr_history[gage_id] = []
        self.grr_history[gage_id].append(result)
        return result

    def calculate_uncertainty(self, instrument_id: str,
                              calibration_std_uncertainty: float,
                              resolution: float,
                              repeatability_std: float,
                              coverage_factor: float = 2.0) -> dict:
        """GUM method measurement uncertainty calculation."""
        # Type B: calibration standard uncertainty
        u_cal = calibration_std_uncertainty

        # Type B: resolution (rectangular distribution)
        u_res = resolution / (2 * np.sqrt(3))

        # Type A: repeatability from measurement data
        u_rep = repeatability_std

        # Combined standard uncertainty
        u_combined = np.sqrt(u_cal**2 + u_res**2 + u_rep**2)

        # Expanded uncertainty
        u_expanded = coverage_factor * u_combined

        if instrument_id in self.instruments:
            self.instruments[instrument_id].uncertainty_um = u_expanded

        return {
            "instrument_id": instrument_id,
            "u_calibration": round(u_cal, 4),
            "u_resolution": round(u_res, 4),
            "u_repeatability": round(u_rep, 4),
            "u_combined": round(u_combined, 4),
            "u_expanded": round(u_expanded, 4),
            "coverage_factor": coverage_factor,
            "confidence_level": "95%" if coverage_factor == 2.0 else "99%",
        }

    def optimize_cal_intervals(self) -> list[dict]:
        """Adjust calibration intervals based on risk and usage."""
        adjustments = []
        for inst_id, inst in self.instruments.items():
            risk_factor = self._calculate_risk_factor(inst)

            # Base interval adjusted by risk
            new_interval = int(inst.cal_interval_days / risk_factor)
            new_interval = max(30, min(730, new_interval))

            if new_interval != inst.cal_interval_days:
                old = inst.cal_interval_days
                inst.cal_interval_days = new_interval
                inst.next_cal_due = (
                    inst.last_cal_date + timedelta(days=new_interval)
                )
                adjustments.append({
                    "instrument_id": inst_id,
                    "old_interval_days": old,
                    "new_interval_days": new_interval,
                    "risk_factor": round(risk_factor, 2),
                    "next_due": inst.next_cal_due.strftime("%Y-%m-%d"),
                    "reason": self._get_adjustment_reason(inst, risk_factor),
                })
        return adjustments

    def _calculate_risk_factor(self, inst: Instrument) -> float:
        """Calculate risk-based interval adjustment factor."""
        factor = 1.0
        criticality_weights = {
            "low": 0.7, "medium": 1.0, "high": 1.3, "critical": 1.8
        }
        factor *= criticality_weights.get(inst.criticality, 1.0)

        # High usage = shorter intervals
        if inst.usage_count > 1000:
            factor *= 1.3
        elif inst.usage_count > 500:
            factor *= 1.1

        # Stability trend: if drifting, shorten interval
        if len(inst.stability_trend) >= 3:
            recent = inst.stability_trend[-3:]
            if all(recent[i] < recent[i+1] for i in range(len(recent)-1)):
                factor *= 1.5  # Increasing drift

        return factor

    def _get_adjustment_reason(self, inst, risk_factor):
        reasons = []
        if inst.criticality in ("high", "critical"):
            reasons.append(f"High criticality ({inst.criticality})")
        if inst.usage_count > 1000:
            reasons.append(f"Heavy usage ({inst.usage_count} uses)")
        return "; ".join(reasons) if reasons else "Standard risk assessment"

The calibration and MSA agent ensures that the foundation of your quality system -- the measurements themselves -- are trustworthy. Automated Gage R&R studies catch measurement system degradation early, risk-based calibration intervals optimize resource allocation, and GUM-compliant uncertainty calculations provide the metrological rigor required by ISO 17025 and customer-specific requirements. Every calibration event and MSA study is recorded with full traceability, making audit preparation effortless.

6. ROI Analysis: Manufacturing Plant Case Study

Quantifying the return on an AI quality assurance investment requires mapping agent capabilities to specific cost categories. The model below analyzes a mid-size manufacturing plant (500 employees, $80M annual revenue, current 2.5% defect rate) implementing a comprehensive AI QA agent across all five capabilities covered in this guide. The results demonstrate that quality AI is not a cost center -- it is one of the highest-ROI investments a manufacturer can make.

Cost Model Breakdown

The analysis covers four primary savings categories: defect rate reduction through improved inspection and SPC, scrap and rework reduction through earlier detection, recall prevention through better containment, and inspection labor optimization through automated visual inspection and dynamic sampling plans. Each category is modeled conservatively with ranges reflecting industry benchmarks from plants that have deployed similar systems.

from dataclasses import dataclass


@dataclass
class PlantProfile:
    name: str
    employees: int
    annual_revenue: float
    production_volume_units: int
    current_defect_rate: float       # as decimal (0.025 = 2.5%)
    avg_unit_cost: float
    avg_rework_cost_per_unit: float
    avg_scrap_cost_per_unit: float
    inspection_staff_count: int
    avg_inspector_salary: float
    annual_warranty_claims: int
    avg_warranty_cost: float
    recall_risk_annual_pct: float    # probability of recall event
    avg_recall_cost: float


class QAROIAnalyzer:
    """ROI analysis for AI quality assurance deployment."""

    # Improvement factors (conservative industry benchmarks)
    DEFECT_RATE_REDUCTION = 0.84      # 84% reduction in defect rate
    SCRAP_REWORK_REDUCTION = 0.73     # 73% reduction
    INSPECTION_LABOR_REDUCTION = 0.58  # 58% staff reduction
    WARRANTY_REDUCTION = 0.65          # 65% reduction in claims
    RECALL_RISK_REDUCTION = 0.80       # 80% risk reduction

    def __init__(self, plant: PlantProfile):
        self.plant = plant

    def calculate_full_roi(self, implementation_cost: float,
                            annual_maintenance: float,
                            analysis_years: int = 3) -> dict:
        """Calculate comprehensive ROI across all savings categories."""
        savings = self._calculate_annual_savings()
        costs = self._calculate_costs(
            implementation_cost, annual_maintenance, analysis_years
        )
        roi_metrics = self._calculate_roi_metrics(
            savings, costs, analysis_years
        )

        return {
            "plant_profile": {
                "name": self.plant.name,
                "employees": self.plant.employees,
                "revenue": f"${self.plant.annual_revenue:,.0f}",
                "current_defect_rate": f"{self.plant.current_defect_rate:.1%}",
            },
            "annual_savings_breakdown": savings,
            "cost_structure": costs,
            "roi_metrics": roi_metrics,
        }

    def _calculate_annual_savings(self) -> dict:
        p = self.plant

        # 1. Defect rate reduction savings
        current_defects = p.production_volume_units * p.current_defect_rate
        new_defect_rate = p.current_defect_rate * (
            1 - self.DEFECT_RATE_REDUCTION
        )
        avoided_defects = current_defects - (
            p.production_volume_units * new_defect_rate
        )

        # 2. Scrap and rework savings
        current_scrap_rework = current_defects * (
            p.avg_rework_cost_per_unit * 0.6 +
            p.avg_scrap_cost_per_unit * 0.4
        )
        scrap_rework_savings = (
            current_scrap_rework * self.SCRAP_REWORK_REDUCTION
        )

        # 3. Inspection labor optimization
        reduced_staff = int(
            p.inspection_staff_count * self.INSPECTION_LABOR_REDUCTION
        )
        labor_savings = reduced_staff * p.avg_inspector_salary

        # 4. Warranty and recall prevention
        warranty_savings = (
            p.annual_warranty_claims * p.avg_warranty_cost
            * self.WARRANTY_REDUCTION
        )
        recall_avoidance = (
            p.recall_risk_annual_pct * p.avg_recall_cost
            * self.RECALL_RISK_REDUCTION
        )

        # 5. Throughput improvement (fewer stops, faster decisions)
        throughput_gain = p.annual_revenue * 0.008  # ~0.8% revenue lift

        total = (scrap_rework_savings + labor_savings +
                 warranty_savings + recall_avoidance + throughput_gain)

        return {
            "defect_reduction": {
                "avoided_defects_per_year": int(avoided_defects),
                "new_defect_rate": f"{new_defect_rate:.2%}",
            },
            "scrap_rework_savings": f"${scrap_rework_savings:,.0f}",
            "inspection_labor_savings": f"${labor_savings:,.0f}",
            "warranty_savings": f"${warranty_savings:,.0f}",
            "recall_risk_avoidance": f"${recall_avoidance:,.0f}",
            "throughput_improvement": f"${throughput_gain:,.0f}",
            "total_annual_savings": f"${total:,.0f}",
            "_total_numeric": total,
        }

    def _calculate_costs(self, impl_cost, annual_maint, years) -> dict:
        total = impl_cost + annual_maint * years
        return {
            "implementation_year_1": f"${impl_cost:,.0f}",
            "annual_maintenance": f"${annual_maint:,.0f}",
            "total_cost_of_ownership": f"${total:,.0f}",
            "analysis_period_years": years,
            "_total_numeric": total,
        }

    def _calculate_roi_metrics(self, savings, costs, years) -> dict:
        annual_savings = savings["_total_numeric"]
        total_cost = costs["_total_numeric"]
        impl_cost = float(
            costs["implementation_year_1"].replace("$", "").replace(",", "")
        )
        annual_maint = float(
            costs["annual_maintenance"].replace("$", "").replace(",", "")
        )

        # Payback period (months)
        first_year_net = annual_savings - impl_cost
        if annual_savings > 0:
            payback_months = (impl_cost / annual_savings) * 12
        else:
            payback_months = float("inf")

        # N-year ROI
        total_savings = annual_savings * years
        roi_pct = ((total_savings - total_cost) / total_cost) * 100

        # NPV at 10% discount rate
        npv = -impl_cost
        for yr in range(1, years + 1):
            net_cf = annual_savings - annual_maint
            npv += net_cf / (1.10 ** yr)

        return {
            "payback_period_months": round(payback_months, 1),
            "annual_net_savings": f"${annual_savings - annual_maint:,.0f}",
            f"{years}_year_roi_pct": f"{roi_pct:.0f}%",
            "npv_10pct_discount": f"${npv:,.0f}",
            "savings_to_cost_ratio": f"{total_savings / total_cost:.1f}x",
        }


# Example: 500-employee manufacturing plant
plant = PlantProfile(
    name="Midwest Precision Manufacturing",
    employees=500,
    annual_revenue=80_000_000,
    production_volume_units=2_000_000,
    current_defect_rate=0.025,
    avg_unit_cost=28.00,
    avg_rework_cost_per_unit=18.50,
    avg_scrap_cost_per_unit=28.00,
    inspection_staff_count=24,
    avg_inspector_salary=52_000,
    annual_warranty_claims=180,
    avg_warranty_cost=2_800,
    recall_risk_annual_pct=0.08,
    avg_recall_cost=1_500_000,
)

analyzer = QAROIAnalyzer(plant)
results = analyzer.calculate_full_roi(
    implementation_cost=450_000,
    annual_maintenance=180_000,
    analysis_years=3,
)

for section, data in results.items():
    print(f"\n{'='*50}")
    print(f"  {section.upper()}")
    print(f"{'='*50}")
    if isinstance(data, dict):
        for k, v in data.items():
            if not k.startswith("_"):
                print(f"  {k}: {v}")

The ROI model demonstrates why AI-powered quality assurance is among the highest-return investments in manufacturing. The combination of defect reduction, scrap elimination, labor optimization, and recall prevention generates savings that dwarf implementation costs within weeks, not years. The model is deliberately conservative -- plants with higher defect rates, more expensive products, or regulated industries (medical devices, aerospace) typically see even larger returns due to higher cost-of-quality baselines.

Implementation Roadmap

Deploying AI quality assurance should follow a phased approach that delivers value at each stage:

1. Phase 1 (Weeks 1-4): SPC Intelligence. Connect to existing process data historians and deploy automated control chart monitoring. This delivers the fastest ROI by catching process drifts that currently go undetected for hours.
2. Phase 2 (Weeks 5-8): Visual Inspection. Deploy computer vision on the highest-volume or highest-defect production line. Start with one camera station and expand based on results.
3. Phase 3 (Weeks 9-12): Root Cause and Supplier Quality. Digitize FMEA databases, connect supplier lot history, and deploy automated RCA and dynamic sampling plans.
4. Phase 4 (Weeks 13-16): Calibration and MSA. Migrate calibration records, implement risk-based scheduling, and automate Gage R&R studies for critical measurement systems.

Each phase builds on the previous one. SPC data feeds the root cause agent. Visual inspection data refines the SPC models. Supplier quality data informs incoming inspection optimization. The result is not six disconnected tools but an integrated quality intelligence system that continuously improves itself.

Conclusion

AI agents for quality assurance represent a fundamental shift from reactive quality control to proactive quality intelligence. The six capabilities covered in this guide -- visual inspection, SPC monitoring, root cause analysis, supplier quality management, calibration automation, and ROI analysis -- form a complete system that detects defects earlier, identifies root causes faster, optimizes supplier performance dynamically, and ensures measurement system integrity continuously. For a 500-employee manufacturing plant, this translates to over $3M in annual savings with payback measured in weeks. The technology is mature, the code patterns are proven, and the ROI case is compelling. The question for quality leaders is no longer whether to deploy AI agents but how quickly they can begin capturing these benefits.