Module 7: Production Considerations for RAG

Capsule 08: Project - Production Readiness Checklist

Capsule description

You will build an executable Python tool that validates a RAG system's configuration against production criteria. The tool takes a configuration (dict or JSON/YAML file), evaluates each area (scaling, monitoring, backups, security, cost optimization, migration readiness), and produces a readiness report with pass/fail per criterion and a global score. It's a practical tool you can use to audit your real system, prioritize gaps, and plan improvements.

Estimated time: 45-60 minutes


Project goal

  • Implement a validator that evaluates a RAG configuration against 6 production areas.
  • Produce a report with pass/fail status per criterion and a global score (0-100).
  • Prioritize gaps by impact and generate a recommended closure plan.
  • Complete, executable, and reusable code in your project.

Project specifications

Functional requirements

  1. Input: RAG configuration as a dict or JSON/YAML file with fields per area.
  2. Evaluation: Each criterion returns pass, fail, or warning with an explanatory message.
  3. Global score: Weighted average by the criticality of each area.
  4. Output: A readable report in the console and optionally in Markdown/JSON.

Areas to validate

AreaKey criteria
ScalingStrategy defined, limits documented, autoscaling plan
MonitoringDashboards (p95, error rate, throughput), active alerts, runbooks
BackupsAutomatic backups, RTO/RPO defined, restore tested recently
SecurityAuth, rate limiting, input validation
Cost optimizationEmbedding cache, cost-per-query metrics, batch ops
Migration readinessDual-write ready, canary plan, rollback criteria

Success criteria

  • ✅ All 6 areas evaluated with concrete, verifiable criteria.
  • ✅ Pass/fail/warning per criterion with actionable messages.
  • ✅ Global score 0-100 with a per-area breakdown.
  • ✅ Top N gaps prioritized by impact.
  • ✅ Batch mode to validate multiple configs (--validate).

Context before you start

This project synthesizes capsules 02-07 of the module:

  • 02: Scaling strategies (vertical, horizontal, sharding).
  • 03: Monitoring (Prometheus, Grafana, alerts, runbooks).
  • 04: Backup and disaster recovery (RTO, RPO, restore).
  • 05: Security (auth, rate limiting, validation).
  • 06: Cost optimization (cache, batch, metrics).
  • 07: Zero-downtime migration (dual-write, canary, rollback).

The tool does not run your RAG system: it validates the declarative configuration you describe. It's useful for pre-release audits, periodic reviews, and team onboarding.


Step-by-step implementation

Step 1: Define the configuration schema

The tool expects a dictionary with the following structure. Each field is optional; missing ones are considered "not implemented".

# production_readiness_checklist.py
"""
Production Readiness Checklist for RAG systems.
Evaluates configuration against production criteria.
"""

from dataclasses import dataclass
from typing import Literal
import json
from pathlib import Path

# ----- RAG configuration schema -----

DEFAULT_CONFIG = {
    "scaling": {
        "strategy_defined": False,
        "strategy_type": None,  # "vertical" | "horizontal" | "sharding" | None
        "resource_limits_documented": False,
        "autoscaling_plan": False,
    },
    "monitoring": {
        "dashboards_p95": False,
        "dashboards_error_rate": False,
        "dashboards_throughput": False,
        "alerts_active": False,
        "alerts_tested": False,
        "runbooks_available": False,
    },
    "backups": {
        "automatic_backups": False,
        "rto_hours": None,
        "rpo_hours": None,
        "restore_tested_recently": False,
    },
    "security": {
        "auth_enabled": False,
        "rate_limiting_enabled": False,
        "input_validation": False,
    },
    "cost_optimization": {
        "embedding_cache": False,
        "result_cache": False,
        "cost_per_query_metric": False,
        "batch_operations": False,
    },
    "migration_readiness": {
        "dual_write_ready": False,
        "canary_plan": False,
        "rollback_criteria": False,
    },
}

Why this design? The configuration is declarative: you describe what you have today, not how you implemented it. This way you can validate real, simulated, or multi-environment configs.


Step 2: Define criteria and weights per area

Each area has weighted criteria. The area score is the weighted percentage of criteria that pass.

@dataclass
class Criterion:
    key: str
    label: str
    weight: int  # 1-5, higher = more critical
    description: str = ""

CRITERIA_BY_AREA = {
    "scaling": [
        Criterion("strategy_defined", "Scaling strategy defined", 5, "vertical/horizontal/sharding documented"),
        Criterion("resource_limits_documented", "Resource limits documented", 3, "CPU, RAM, connections defined"),
        Criterion("autoscaling_plan", "Autoscaling plan", 4, "Scale up/down rules defined"),
    ],
    "monitoring": [
        Criterion("dashboards_p95", "Dashboard with p95 latency", 5, "p50/p95/p99 of retrieval visible"),
        Criterion("dashboards_error_rate", "Dashboard with error rate", 5, "Error rate per endpoint"),
        Criterion("dashboards_throughput", "Dashboard with throughput", 3, "Queries/sec, ingestion docs/sec"),
        Criterion("alerts_active", "Active alerts", 5, "Prometheus/Grafana/Cloud alerting"),
        Criterion("alerts_tested", "Alerts tested", 3, "Alert test run executed"),
        Criterion("runbooks_available", "Runbooks available", 4, "Procedure per alert"),
    ],
    "backups": [
        Criterion("automatic_backups", "Automatic backups running", 5, "Cron or scheduled job"),
        Criterion("rto_hours", "RTO defined (hours)", 4, "Recovery Time Objective documented"),
        Criterion("rpo_hours", "RPO defined (hours)", 4, "Recovery Point Objective documented"),
        Criterion("restore_tested_recently", "Restore tested recently", 5, "Last 30 days"),
    ],
    "security": [
        Criterion("auth_enabled", "Authentication active", 5, "API key, JWT, or OAuth"),
        Criterion("rate_limiting_enabled", "Rate limiting active", 4, "Per client/tenant"),
        Criterion("input_validation", "Input validation", 5, "Anti prompt-injection, sanitization"),
    ],
    "cost_optimization": [
        Criterion("embedding_cache", "Embedding cache", 3, "Don't re-embed unchanged docs"),
        Criterion("result_cache", "Result cache", 4, "Redis or similar for repeated queries"),
        Criterion("cost_per_query_metric", "Cost per query metric", 4, "Estimated cost per query"),
        Criterion("batch_operations", "Batch operations in ingestion", 3, "Batch upsert, not one-by-one"),
    ],
    "migration_readiness": [
        Criterion("dual_write_ready", "Dual-write ready", 4, "Code ready to write to 2 DBs"),
        Criterion("canary_plan", "Canary plan documented", 4, "Traffic percentage, duration, metrics"),
        Criterion("rollback_criteria", "Rollback criteria", 5, "When to abort the migration"),
    ],
}

# Weight of each area in the global score (1-5)
AREA_WEIGHTS = {
    "scaling": 3,
    "monitoring": 5,
    "backups": 5,
    "security": 5,
    "cost_optimization": 2,
    "migration_readiness": 2,
}

Step 3: Evaluation engine

Evaluates each criterion against the configuration. Criteria have specific validations (e.g. RTO <= 24 for customer-facing production).

Status = Literal["pass", "fail", "warning"]

@dataclass
class CriterionResult:
    criterion: Criterion
    status: Status
    message: str

@dataclass
class AreaResult:
    area: str
    score: float
    passed: int
    total: int
    results: list[CriterionResult]

def eval_criterion(area: str, criterion: Criterion, config: dict) -> CriterionResult:
    """Evaluate a criterion against the configuration."""
    area_config = config.get(area, {})
    value = area_config.get(criterion.key)

    if value is None:
        return CriterionResult(criterion, "fail", f"Not configured: {criterion.label}")
    if isinstance(value, bool):
        if value:
            return CriterionResult(criterion, "pass", f"✓ {criterion.label}")
        return CriterionResult(criterion, "fail", f"✗ {criterion.label} disabled")

    # Criteria with numeric values
    if criterion.key == "rto_hours":
        if value is None or value <= 0:
            return CriterionResult(criterion, "fail", "RTO not defined")
        if value <= 2:
            return CriterionResult(criterion, "pass", f"RTO ≤ 2h ({value}h) - high production")
        if value <= 8:
            return CriterionResult(criterion, "warning", f"RTO {value}h - acceptable for medium criticality")
        return CriterionResult(criterion, "fail", f"RTO {value}h - high for production")

    if criterion.key == "rpo_hours":
        if value is None or value <= 0:
            return CriterionResult(criterion, "fail", "RPO not defined")
        if value <= 6:
            return CriterionResult(criterion, "pass", f"RPO ≤ 6h ({value}h)")
        if value <= 24:
            return CriterionResult(criterion, "warning", f"RPO {value}h - more acceptable data loss")
        return CriterionResult(criterion, "fail", f"RPO {value}h - high risk")

    return CriterionResult(criterion, "fail", f"Unexpected value: {value}")


def eval_area(area: str, config: dict) -> AreaResult:
    """Evaluate a full area."""
    criteria = CRITERIA_BY_AREA.get(area, [])
    results = [eval_criterion(area, c, config) for c in criteria]
    total_weight = sum(c.weight for c in criteria)
    passed_weight = sum(
        c.weight for r in results
        for c in [r.criterion]
        if r.status == "pass"
    )
    score = (passed_weight / total_weight * 100) if total_weight else 0.0
    passed_count = sum(1 for r in results if r.status == "pass")
    return AreaResult(area, round(score, 1), passed_count, len(results), results)


def eval_all(config: dict) -> dict:
    """Evaluate all areas and compute the global score."""
    area_results = {area: eval_area(area, config) for area in CRITERIA_BY_AREA}
    total_weight = sum(AREA_WEIGHTS.get(a, 1) for a in area_results)
    global_score = sum(
        AREA_WEIGHTS.get(area, 1) * r.score
        for area, r in area_results.items()
    ) / total_weight if total_weight else 0.0
    return {
        "global_score": round(global_score, 1),
        "areas": area_results,
    }

Step 4: Gap prioritization

Orders the gaps (fail or warning criteria) by impact: criterion weight × area weight.

@dataclass
class Gap:
    area: str
    criterion: str
    label: str
    status: Status
    message: str
    impact_score: float


def get_prioritized_gaps(eval_result: dict, top_n: int = 5) -> list[Gap]:
    """Extract gaps and order them by impact."""
    gaps = []
    for area, area_result in eval_result["areas"].items():
        area_weight = AREA_WEIGHTS.get(area, 1)
        for r in area_result.results:
            if r.status in ("fail", "warning"):
                criterion_weight = next(
                    (c.weight for c in CRITERIA_BY_AREA[area] if c.key == r.criterion.key),
                    1,
                )
                impact = area_weight * criterion_weight
                gaps.append(Gap(
                    area=area,
                    criterion=r.criterion.key,
                    label=r.criterion.label,
                    status=r.status,
                    message=r.message,
                    impact_score=impact,
                ))
    gaps.sort(key=lambda g: g.impact_score, reverse=True)
    return gaps[:top_n]

Step 5: Report generator

Produces a readable report in the console.

def generate_report(eval_result: dict, config_name: str = "RAG Config") -> str:
    """Generate a readiness report."""
    lines = []
    lines.append("=" * 70)
    lines.append(f"  PRODUCTION READINESS REPORT — {config_name}")
    lines.append("=" * 70)

    # Global score
    score = eval_result["global_score"]
    status_emoji = "✅" if score >= 80 else ("⚠️" if score >= 60 else "❌")
    lines.append(f"\n  Global score: {score:.1f}/100  {status_emoji}")
    lines.append("")

    # Per area
    lines.append("  STATUS BY AREA")
    lines.append("  " + "-" * 50)
    for area, r in eval_result["areas"].items():
        area_label = area.replace("_", " ").title()
        emoji = "🟢" if r.score >= 80 else ("🟡" if r.score >= 50 else "🔴")
        lines.append(f"  {emoji} {area_label}: {r.score:.1f}% ({r.passed}/{r.total})")
    lines.append("")

    # Detail per criterion
    lines.append("  DETAIL BY CRITERION")
    lines.append("  " + "-" * 50)
    for area, r in eval_result["areas"].items():
        area_label = area.replace("_", " ").title()
        lines.append(f"\n  [{area_label}]")
        for res in r.results:
            sym = "✓" if res.status == "pass" else ("!" if res.status == "warning" else "✗")
            lines.append(f"    {sym} {res.message}")

    # Top gaps
    gaps = get_prioritized_gaps(eval_result, top_n=5)
    if gaps:
        lines.append("\n  TOP 5 PRIORITIZED GAPS")
        lines.append("  " + "-" * 50)
        for i, g in enumerate(gaps, 1):
            lines.append(f"  {i}. [{g.area}] {g.label}{g.message}")

    # Suggested plan
    lines.append("\n  SUGGESTED PLAN (2 WEEKS)")
    lines.append("  " + "-" * 50)
    critical_areas = [
        (a, r) for a, r in eval_result["areas"].items()
        if r.score < 60
    ]
    critical_areas.sort(key=lambda x: x[1].score)
    if critical_areas:
        week1 = [a for a, _ in critical_areas[:2]]
        week2 = [a for a, _ in critical_areas[2:4]]
        lines.append(f"  Week 1: {', '.join(week1)}")
        lines.append(f"  Week 2: {', '.join(week2) if week2 else 'Refinement and documentation'}")
    else:
        lines.append("  Week 1–2: Refine yellow areas and document runbooks")

    lines.append("\n" + "=" * 70)
    return "\n".join(lines)

Step 6: Load configuration from a file

Supports JSON and YAML.

def load_config(path: str) -> dict:
    """Load configuration from JSON or YAML."""
    p = Path(path)
    if not p.exists():
        raise FileNotFoundError(f"Does not exist: {path}")

    raw = p.read_text(encoding="utf-8")
    suffix = p.suffix.lower()

    if suffix == ".json":
        config = json.loads(raw)
    elif suffix in (".yaml", ".yml"):
        try:
            import yaml
            config = yaml.safe_load(raw)
        except ImportError:
            raise ImportError("PyYAML required for YAML files: pip install pyyaml")
    else:
        raise ValueError("Unsupported format. Use .json or .yaml")

    # Merge with defaults for missing fields
    return merge_config(DEFAULT_CONFIG, config)


def merge_config(base: dict, override: dict) -> dict:
    """Recursive merge of configs. Override wins."""
    result = base.copy()
    for k, v in override.items():
        if k in result and isinstance(result[k], dict) and isinstance(v, dict):
            result[k] = merge_config(result[k], v)
        else:
            result[k] = v
    return result

Step 7: Main script

Two modes: validate a config or run test scenarios.

import sys

def main():
    print("=" * 70)
    print("  PRODUCTION READINESS CHECKLIST — RAG Systems")
    print("  Module 7 — Production Considerations")
    print("=" * 70)

    if len(sys.argv) > 1 and sys.argv[1] == "--validate":
        run_validation_mode()
    elif len(sys.argv) > 1:
        run_config_file_mode(sys.argv[1])
    else:
        run_demo_mode()


def run_demo_mode():
    """Uses an example configuration and generates a report."""
    print("\n  Demo mode: example config\n")
    config = {
        "scaling": {"strategy_defined": True, "strategy_type": "horizontal", "resource_limits_documented": True, "autoscaling_plan": False},
        "monitoring": {"dashboards_p95": True, "dashboards_error_rate": True, "dashboards_throughput": True, "alerts_active": True, "alerts_tested": False, "runbooks_available": True},
        "backups": {"automatic_backups": True, "rto_hours": 2, "rpo_hours": 6, "restore_tested_recently": False},
        "security": {"auth_enabled": True, "rate_limiting_enabled": True, "input_validation": True},
        "cost_optimization": {"embedding_cache": False, "result_cache": True, "cost_per_query_metric": False, "batch_operations": True},
        "migration_readiness": {"dual_write_ready": False, "canary_plan": False, "rollback_criteria": False},
    }
    result = eval_all(config)
    print(generate_report(result, "Demo RAG System"))


def run_config_file_mode(path: str):
    """Validates a configuration from a file."""
    print(f"\n  Loading config from: {path}\n")
    config = load_config(path)
    result = eval_all(config)
    print(generate_report(result, Path(path).stem))


def run_validation_mode():
    """Runs predefined validation scenarios."""
    print("\n  Validation mode: 3 scenarios\n")
    scenarios = [
        {"name": "MVP / PoC", "config": {"scaling": {"strategy_defined": False}, "monitoring": {"alerts_active": False}, "backups": {"automatic_backups": False}, "security": {"auth_enabled": False}, "cost_optimization": {}, "migration_readiness": {}}},
        {"name": "Early production", "config": {"scaling": {"strategy_defined": True, "strategy_type": "horizontal"}, "monitoring": {"dashboards_p95": True, "alerts_active": True}, "backups": {"automatic_backups": True, "rto_hours": 4, "rpo_hours": 12, "restore_tested_recently": False}, "security": {"auth_enabled": True, "rate_limiting_enabled": True}, "cost_optimization": {"result_cache": True}, "migration_readiness": {"rollback_criteria": True}}},
        {"name": "Mature production", "config": {"scaling": {"strategy_defined": True, "resource_limits_documented": True, "autoscaling_plan": True}, "monitoring": {"dashboards_p95": True, "dashboards_error_rate": True, "dashboards_throughput": True, "alerts_active": True, "alerts_tested": True, "runbooks_available": True}, "backups": {"automatic_backups": True, "rto_hours": 1, "rpo_hours": 4, "restore_tested_recently": True}, "security": {"auth_enabled": True, "rate_limiting_enabled": True, "input_validation": True}, "cost_optimization": {"embedding_cache": True, "result_cache": True, "cost_per_query_metric": True, "batch_operations": True}, "migration_readiness": {"dual_write_ready": True, "canary_plan": True, "rollback_criteria": True}}},
    ]
    for s in scenarios:
        result = eval_all(merge_config(DEFAULT_CONFIG, s["config"]))
        print(f"  Scenario: {s['name']}")
        print(f"  Score: {result['global_score']:.1f}/100")
        print()
    print("  ✓ Validation completed\n")


if __name__ == "__main__":
    main()

Example configuration file

Save this as rag_config_example.json to test:

{
  "scaling": {
    "strategy_defined": true,
    "strategy_type": "horizontal",
    "resource_limits_documented": true,
    "autoscaling_plan": true
  },
  "monitoring": {
    "dashboards_p95": true,
    "dashboards_error_rate": true,
    "dashboards_throughput": true,
    "alerts_active": true,
    "alerts_tested": true,
    "runbooks_available": true
  },
  "backups": {
    "automatic_backups": true,
    "rto_hours": 2,
    "rpo_hours": 6,
    "restore_tested_recently": true
  },
  "security": {
    "auth_enabled": true,
    "rate_limiting_enabled": true,
    "input_validation": true
  },
  "cost_optimization": {
    "embedding_cache": true,
    "result_cache": true,
    "cost_per_query_metric": true,
    "batch_operations": true
  },
  "migration_readiness": {
    "dual_write_ready": true,
    "canary_plan": true,
    "rollback_criteria": true
  }
}

Expected output

Running python production_readiness_checklist.py:

======================================================================
  PRODUCTION READINESS REPORT — Demo RAG System
======================================================================

  Global score: 72.8/100  ⚠️

  STATUS BY AREA
  --------------------------------------------------
  🟡 Scaling: 66.7% (2/3)
  🟢 Monitoring: 88.0% (5/6)
  🟡 Backups: 72.2% (3/4)
  🟢 Security: 100.0% (3/3)
  🟡 Cost Optimization: 50.0% (2/4)
  🔴 Migration Readiness: 0.0% (0/3)

  DETAIL BY CRITERION
  --------------------------------------------------
  [Scaling]
    ✓ Scaling strategy defined
    ✓ Resource limits documented
    ✗ Autoscaling plan disabled
  ...

  TOP 5 PRIORITIZED GAPS
  --------------------------------------------------
  1. [backups] Restore tested recently — ✗ disabled
  2. [monitoring] Alerts tested — ✗ disabled
  3. [scaling] Autoscaling plan — ✗ disabled
  4. [migration_readiness] Rollback criteria — ✗ disabled
  5. [cost_optimization] Cost per query metric — ✗ disabled

  SUGGESTED PLAN (2 WEEKS)
  --------------------------------------------------
  Week 1: migration_readiness, cost_optimization
  Week 2: Refinement and documentation

======================================================================

Project deliverables

When you finish you should have:

  1. Executable script production_readiness_checklist.py with demo, file, and validation modes.
  2. Example config in JSON or YAML representing your real system or a typical scenario.
  3. Generated report with score, per-area detail, and top 5 gaps.
  4. 2-week plan to close critical gaps, based on the tool's output.

Recommended delivery format

Include a table like this in your documentation:

AreaCurrent statusScoreMain gapPriorityDue date
ScalingGreen83%Missing autoscaling planMediumWk 1
ObservabilityGreen89%Untested alertsLowWk 2
DRYellow75%Restore untestedHighWk 1
SecurityGreen100%
CostYellow50%Missing embedding cacheMediumWk 2
MigrationRed0%No canary planHighWk 1

Suggested approval criterion

Define a minimum to reach "acceptable readiness":

  • 0 critical gaps without an assigned owner.
  • Basic alerts active (p95, error rate).
  • Backup + restore tested at least once.
  • Incident plan (runbook) available for critical alerts.
  • Global score ≥ 60 for staging, ≥ 80 for customer-facing production.

Technical defense questions

When presenting the checklist, you should be able to answer:

  1. What is your biggest operational risk today according to the report?
  2. Which action reduces the most risk in the least time?
  3. What metric confirms you improved after closing gaps?
  4. How did you prioritize between areas of equal impact?
  5. Which criterion would you remove or add for your context?

Project troubleshooting

"Everything is red/yellow and we don't know where to start"

Prioritize by user impact + probability of failure. Use the Top 5 gaps from the report: the order is already computed by impact_score. Start with the first 2. If they are from different areas (e.g. backups and security), split the work: one person on backups, another on security.


"The checklist is too big, the team is overloaded"

Split it into 2-week phases with concrete objectives. Week 1: only backups + security (blocking). Week 2: monitoring + one more area. Don't try to close everything at once.


"There is no agreement on priority between teams"

Use a common criterion: downtime risk (what happens if it fails?), SLA impact (does it affect users?), cost (how much does not doing it cost?). The tool already orders by impact. If the disagreement persists, document both priorities and the explicit trade-off.


"The config doesn't reflect our system's reality"

The tool validates declarative configuration, not runtime behavior. If your config says auth_enabled: true but there is an endpoint without auth, the tool doesn't detect it. Complement it with: (1) integration tests that verify auth on all endpoints, (2) periodic manual audits. The config must be updated when you change the system.


"We want stricter or different criteria"

The criteria are in CRITERIA_BY_AREA and the weights in AREA_WEIGHTS. Modify the code to add criteria, change weights, or adjust the logic in eval_criterion. If you add a criterion with numeric validation (like RTO), extend eval_criterion with a new if criterion.key == "your_new_criterion".


Post-project exercises

Exercise 1: Export the report to Markdown

Extend the tool with a function that generates a Markdown file with the report, including tables for the per-criterion detail and the top gaps.

Hints:

  • Create export_to_markdown(eval_result, output_path: str).
  • Use Markdown tables (| col1 | col2 |).
  • Include the generation date and the config name.
See solution
from datetime import date

def export_to_markdown(eval_result: dict, output_path: str, config_name: str = "RAG Config"):
    """Export the report to a Markdown file."""
    lines = []
    lines.append(f"# Production Readiness Report — {config_name}")
    lines.append(f"\n**Date:** {date.today().isoformat()}\n")
    lines.append(f"**Global score:** {eval_result['global_score']:.1f}/100\n")
    lines.append("## Status by area\n")
    lines.append("| Area | Score | Passed | Total |")
    lines.append("|------|------:|-------:|-----:|")
    for area, r in eval_result["areas"].items():
        area_label = area.replace("_", " ").title()
        lines.append(f"| {area_label} | {r.score:.1f}% | {r.passed} | {r.total} |")
    lines.append("\n## Detail by criterion\n")
    for area, r in eval_result["areas"].items():
        area_label = area.replace("_", " ").title()
        lines.append(f"### {area_label}\n")
        lines.append("| Criterion | Status |")
        lines.append("|----------|--------|")
        for res in r.results:
            status = "✓" if res.status == "pass" else ("!" if res.status == "warning" else "✗")
            lines.append(f"| {res.criterion.label} | {status} {res.message} |")
    lines.append("\n## Top 5 gaps\n")
    gaps = get_prioritized_gaps(eval_result, 5)
    for i, g in enumerate(gaps, 1):
        lines.append(f"{i}. **[{g.area}]** {g.label}: {g.message}")
    Path(output_path).write_text("\n".join(lines), encoding="utf-8")
    print(f"Report exported to: {output_path}")

# Use in main():
# export_to_markdown(result, "readiness_report.md", "My RAG")

Exercise 2: Add a "documented incidents" criterion

Add a criterion to the monitoring area: that there is a record of incidents from the last 90 days (even if it's just a boolean field incidents_logged).

Hints:

  • Add the criterion to CRITERIA_BY_AREA["monitoring"].
  • Add the field to DEFAULT_CONFIG["monitoring"].
  • In eval_criterion, if criterion.key == "incidents_logged", validate the boolean.
See solution
# 1. In CRITERIA_BY_AREA["monitoring"], add:
Criterion("incidents_logged", "Documented incidents (90 days)", 3, "Incident log and post-mortems"),

# 2. In DEFAULT_CONFIG["monitoring"], add:
"incidents_logged": False,

# 3. In eval_criterion, the bool case already covers this field.
# If value is True -> pass, if False -> fail.

Exercise 3: Comparison mode between two configs

Create a function that takes two config paths and shows which criteria pass in one and fail in the other, useful for comparing staging vs production.

Hints:

  • compare_configs(path_a: str, path_b: str) -> str
  • Load both configs, evaluate each one, compare results per criterion.
  • Generate a table: Criterion | Config A | Config B | Improved?
See solution
def compare_configs(path_a: str, path_b: str) -> str:
    """Compare two configs and show differences."""
    config_a = load_config(path_a)
    config_b = load_config(path_b)
    eval_a = eval_all(config_a)
    eval_b = eval_all(config_b)
    lines = []
    lines.append("=" * 60)
    lines.append("  READINESS COMPARISON")
    lines.append(f"  A: {path_a} ({eval_a['global_score']:.1f})")
    lines.append(f"  B: {path_b} ({eval_b['global_score']:.1f})")
    lines.append("=" * 60)
    lines.append("\n| Area | Criterion | A | B | Change |")
    lines.append("|------|----------|---|---|--------|")
    for area in eval_a["areas"]:
        ra = eval_a["areas"][area].results
        rb = eval_b["areas"][area].results
        for i, res_a in enumerate(ra):
            res_b = rb[i]
            a_sym = "✓" if res_a.status == "pass" else "✗"
            b_sym = "✓" if res_b.status == "pass" else "✗"
            change = "→" if a_sym == b_sym else ("↑" if b_sym == "✓" else "↓")
            lines.append(f"| {area} | {res_a.criterion.label[:25]} | {a_sym} | {b_sym} | {change} |")
    return "\n".join(lines)

Completeness checklist

Code structure:

  • DEFAULT_CONFIG covers the 6 areas with fields per criterion.
  • CRITERIA_BY_AREA has criteria with weight and description.
  • AREA_WEIGHTS defines each area's weight in the global score.

Evaluation engine:

  • eval_criterion() handles bool, RTO, RPO, and unexpected values.
  • eval_area() computes the weighted score by criterion weight.
  • eval_all() returns the global score and per-area results.

Prioritization and report:

  • get_prioritized_gaps() orders gaps by impact.
  • generate_report() produces readable output with a suggested plan.

Input and modes:

  • load_config() supports JSON and YAML.
  • Demo mode, file mode, and --validate mode work.

Validation:

  • At least 3 scenarios (MVP, early production, mature) in --validate.
  • The scores reflect the expected maturity level.

Output:

  • Global score 0-100 visible.
  • Status per area (green/yellow/red).
  • Top 5 gaps and a 2-week plan.

Summary

  • You built an executable Python tool that validates a RAG configuration against 6 production areas.
  • Each area has concrete criteria with pass/fail/warning and actionable messages.
  • The global score is a weighted average; the monitoring, backups, and security areas weigh more.
  • The top 5 gaps are ordered by impact (area × criterion) to prioritize work.
  • The suggested plan assigns weeks based on critical areas (score < 60).
  • The configuration is declarative (dict/JSON/YAML): you describe what you have, not how.
  • You can extend criteria by modifying CRITERIA_BY_AREA and eval_criterion.
  • This project closes Module 7 and connects with Module 8 (integrative RAG project).

Additional resources


Estimated time: 45-60 minutes
Next module: ../../module-08-final-project/en/01-module-introduction-8.md