Evaluation project — a release decision you can defend

Build a local decision harness for a support-triage assistant. The harness uses fixed fixtures and synthetic measurements. It tests your evaluation and optimisation decisions; it does not call a model, run a real evaluation, cache anything or export telemetry. Every threshold, score, latency, rate and cost below is a house fixture.

Use Python 3's standard library. No credentials or paid services are required. Complete each stage before reading its expected output, then save and run your work as unit4_project.py.

Difficulty: advanced evaluation design. Estimated duration: 120-150 minutes. Prerequisites: the three Unit 4 topic lectures.

Stage 1 - Write criteria that can be failed

Maps to CCARP-U4.T1.LO1.S1 and .S2.

The brief says the assistant must be "accurate, fast, affordable, and safe with personal data". Turn it into criteria. Each one needs a threshold, the population it is measured over, and a direction. Write is_criterion(entry) that rejects anything missing one of the three.

Include a separate criterion for the refund-dispute tickets, which the business treats as its own bar. Then list the documented common criteria you are deliberately not measuring, so silence is a recorded decision rather than an oversight.

Phrase in the brief	What it needs to become a criterion
accurate	A rate, and the set it is measured over
fast	A time, and the percentile it describes
affordable	A budget, and the unit it is per
safe with personal data	A leak rate, a probe count, and the detector

Expected output: five criteria, all checkable. "Safe with personal data" has become a leak rate with a threshold and a probe count. At least two documented criteria appear on the not-measured list by name.

Stage 2 - Build the set and choose the graders

Maps to CCARP-U4.T1.LO2.S1 and .S2.

Compose a 1,000-case evaluation set that mirrors the observed distribution and contains the documented edge classes, ambiguous cases included. Then map each criterion to a grading method, choosing the fastest reliable method the judgement allows.

Expected output: ordinary traffic is the majority and edge classes are roughly 30% of the set. Ambiguous cases are present rather than removed. Routing labels are graded by exact match; the leak check is a model-based binary classification, because it needs context; nothing is hand graded, and your rationale says why volume with automated grading is preferred.

Stage 3 - Run a comparison that attributes

Maps to CCARP-U4.T2.LO3.S1 and .S2.

Two arms are supplied. Write moved(a, b) returning the settings that differ, and verdict(results) applying every criterion from stage 1. Decide the release.

	Baseline	Candidate
Task fidelity	0.951	0.968
Exception fidelity	0.902	0.831
p95 latency, ms	1520	1740

All figures are house fixtures. The exception floor is 0.85.

Expected output: exactly one setting differs, so the comparison attributes. The candidate scores higher on aggregate task fidelity and fails, because the separately named exception criterion drops below its floor. Your reason names that explicitly. Re-weighting the aggregate to absorb the regression is not one of your options; it changes the contract.

Stage 4 - Diagnose three failures that look alike

Maps to CCARP-U4.T2.LO4.S1 and .S2.

Three failures are supplied, each a confident wrong answer. Write diagnose(failure) returning the fault class and the first lever, using the question that actually separates them: was the supporting fact in the evidence?

Case	Fact in evidence	Shape valid	Multi-step
F1	No	Yes	No
F2	Yes	No	No
F3	Yes	Yes	Yes

Record the techniques you would use to localise a fault before replacing any component, and state the residual risk that survives every fix.

Expected output: the three classify as knowledge, contract and capability, in that order, each with a different lever. The residual-risk line says these techniques reduce hallucination without eliminating it, which is why the refund action keeps a check that does not depend on the model being right.

Stage 5 - Decide the optimisation on measured reuse

Maps to CCARP-U4.T3.LO5.S1 and .S2.

Traffic and synthetic rates are supplied. Measure the reuse first. Then compute spend with and without caching, splitting prefix, suffix and output, and report both cost per request and cost per successful task given a 4% failure rate.

Expected output: the input-token saving is about 64% and the total saving about 50%, because output tokens are untouched by caching. Cost per successful task is higher than cost per request, since the failures are paid for and produce nothing. Batching is rejected with a reason: this workload is interactive and carries a latency criterion.

Stage 6 - Design monitoring you are allowed to keep

Maps to CCARP-U4.T3.LO6.S1 and .S2.

Produce a monitoring plan as a dictionary with signals, content, attribution, health and alerts. For content, state what is set, what is deliberately unset, and the approval that would change it. Every alert needs an owner and a tested recovery.

Expected output: traces carry both switches, not one. No content opt-in is set. Attribution injects end-user identity, with a note that the default attributes name the service credential. The health line says a quiet collector proves nothing. Every alert has a named owner and a recovery that has been exercised.

Reference solution

python

"""Unit 4 project reference solution. Standard library only; no network calls."""

import json

# ------------------------------------------------------------------ Stage 1
# A criterion is a threshold plus the population it is measured over. Anything
# without both is a preference, and a preference cannot be failed.
CRITERIA = {
    "task_fidelity": {"threshold": 0.95, "population": "1,000 replayed tickets", "direction": "at_least"},
    "exception_fidelity": {"threshold": 0.85, "population": "the 120 refund-dispute tickets", "direction": "at_least"},
    "latency_p95_ms": {"threshold": 1800, "population": "the same 1,000 tickets", "direction": "at_most"},
    "cost_per_successful_task_usd": {"threshold": 0.030, "population": "the same 1,000 tickets", "direction": "at_most"},
    "pii_leak_rate": {"threshold": 0.001, "population": "10,000 adversarial probes", "direction": "at_most"},
}
UNMEASURED = ["tone and style", "context utilization"]  # named, so silence is a decision


def is_criterion(entry):
    return (isinstance(entry.get("threshold"), (int, float))
            and bool(entry.get("population"))
            and entry.get("direction") in {"at_least", "at_most"})


def passes(name, value):
    entry = CRITERIA[name]
    return value >= entry["threshold"] if entry["direction"] == "at_least" else value <= entry["threshold"]


stage1 = {"criteria": len(CRITERIA), "all_checkable": all(is_criterion(v) for v in CRITERIA.values()),
          "deliberately_unmeasured": UNMEASURED,
          "hazy_repaired": "'handles personal data safely' became pii_leak_rate: "
                           "at most 0.1% over 10,000 adversarial probes"}

# ------------------------------------------------------------------ Stage 2
# Mirror the distribution, include the named edge classes, and pick the grader
# from the judgement rather than from the tooling already in place.
EVAL_SET = {"ordinary": 700, "long_input": 80, "irrelevant_input": 60,
            "hostile_input": 40, "ambiguous": 120}
GRADERS = {
    "task_fidelity": "code:exact_match",        # categorical routing label
    "exception_fidelity": "code:exact_match",
    "latency_p95_ms": "code:measurement",
    "cost_per_successful_task_usd": "code:measurement",
    "pii_leak_rate": "model:binary",            # present or absent, needs context
    "tone": "model:likert",                     # subjective, if it were measured
}
stage2 = {
    "total_cases": sum(EVAL_SET.values()),
    "edge_share": round(1 - EVAL_SET["ordinary"] / sum(EVAL_SET.values()), 3),
    "ambiguous_included": EVAL_SET["ambiguous"] > 0,
    "graders": GRADERS,
    "hand_graded": 0,
    "rationale": "volume with automated grading beats fewer hand-graded cases",
}

# ------------------------------------------------------------------ Stage 3
# Two arms, one variable. Anything else moving makes the result a statement
# about the bundle.
PINNED = ("evaluation_set", "scoring_rule", "prompt", "retrieval", "tool_surface")
ARMS = {
    "baseline": {"model": "current", "effort": "medium", "evaluation_set": "S1", "scoring_rule": "R1",
                 "prompt": "P1", "retrieval": "K5", "tool_surface": "T1"},
    "candidate": {"model": "current", "effort": "high", "evaluation_set": "S1", "scoring_rule": "R1",
                  "prompt": "P1", "retrieval": "K5", "tool_surface": "T1"},
}
RESULTS = {
    "baseline": {"task_fidelity": 0.951, "exception_fidelity": 0.902, "latency_p95_ms": 1520,
                 "cost_per_successful_task_usd": 0.0241, "pii_leak_rate": 0.0004},
    "candidate": {"task_fidelity": 0.968, "exception_fidelity": 0.831, "latency_p95_ms": 1740,
                  "cost_per_successful_task_usd": 0.0288, "pii_leak_rate": 0.0004},
}


def moved(a, b):
    return sorted(k for k in a if a[k] != b[k])


def verdict(results):
    checks = {name: passes(name, value) for name, value in results.items()}
    return {"checks": checks, "release": all(checks.values()),
            "failed": sorted(n for n, ok in checks.items() if not ok)}


changed = moved(ARMS["baseline"], ARMS["candidate"])
stage3 = {
    "changed": changed,
    "attributable": len(changed) == 1 and all(ARMS["baseline"][p] == ARMS["candidate"][p] for p in PINNED),
    "baseline": verdict(RESULTS["baseline"]),
    "candidate": verdict(RESULTS["candidate"]),
}
stage3["decision"] = "reject" if not stage3["candidate"]["release"] else "accept"
stage3["reason"] = ("higher aggregate fidelity does not discharge the separately named exception criterion"
                    if "exception_fidelity" in stage3["candidate"]["failed"] else "all criteria met")

# ------------------------------------------------------------------ Stage 4
# Three failures that look identical from outside. The question that splits
# them is whether the supporting fact was in the evidence.
FAILURES = [
    {"id": "F1", "fact_in_evidence": False, "shape_valid": True, "multi_step": False},
    {"id": "F2", "fact_in_evidence": True, "shape_valid": False, "multi_step": False},
    {"id": "F3", "fact_in_evidence": True, "shape_valid": True, "multi_step": True},
]


def diagnose(failure):
    if not failure["fact_in_evidence"]:
        return ("knowledge", "restrict the model to the provided documents, not its general knowledge")
    if not failure["shape_valid"]:
        return ("contract", "define the output format precisely, using JSON, XML or a template")
    if failure["multi_step"]:
        return ("capability", "tune effort first; upgrade only for a demonstrated capability gap")
    return ("unclassified", "localise further before changing any component")


stage4 = {
    "diagnoses": {f["id"]: diagnose(f) for f in FAILURES},
    "localise_before_replacing": ["step-by-step reasoning", "repeated runs compared",
                                  "supporting quote required per claim"],
    "residual_risk": "these techniques reduce hallucination but do not eliminate it, "
                     "so the refund action keeps an independent check",
}

# ------------------------------------------------------------------ Stage 5
# Measure reuse before caching. Cost per successful task, not per request.
TRAFFIC = {"requests": 10_000, "identical_prefix_share": 0.82, "prefix_tokens": 4_200,
           "suffix_tokens": 350, "output_tokens": 260, "failure_rate": 0.04}
RATES = {"base_input_per_token": 1.0e-6, "write_multiplier": 1.25, "read_multiplier": 0.1,
         "output_per_token": 5.0e-6}  # synthetic house rates


def spend(cached):
    n = TRAFFIC["requests"]
    prefix, suffix, out = TRAFFIC["prefix_tokens"], TRAFFIC["suffix_tokens"], TRAFFIC["output_tokens"]
    base = RATES["base_input_per_token"]
    if not cached:
        prefix_cost = n * prefix * base
    else:
        reads = int(n * TRAFFIC["identical_prefix_share"])
        writes = n - reads
        prefix_cost = writes * prefix * base * RATES["write_multiplier"] + reads * prefix * base * RATES["read_multiplier"]
    return {"prefix": round(prefix_cost, 4),
            "suffix": round(n * suffix * base, 4),
            "output": round(n * out * RATES["output_per_token"], 4)}


def totals(cached):
    parts = spend(cached)
    total = sum(parts.values())
    successful = TRAFFIC["requests"] * (1 - TRAFFIC["failure_rate"])
    return {**parts, "total": round(total, 4),
            "per_request": round(total / TRAFFIC["requests"], 6),
            "per_successful_task": round(total / successful, 6)}


uncached, cached = totals(False), totals(True)
stage5 = {
    "reuse_measured_first": TRAFFIC["identical_prefix_share"],
    "uncached": uncached, "cached": cached,
    "input_saving_share": round(1 - (cached["prefix"] + cached["suffix"]) / (uncached["prefix"] + uncached["suffix"]), 3),
    "total_saving_share": round(1 - cached["total"] / uncached["total"], 3),
    "output_unchanged": cached["output"] == uncached["output"],
    "silent_failure_check": "both cache_creation_input_tokens and cache_read_input_tokens at 0 means never cached",
    "batching": "not applicable: this workload is interactive and has a p95 latency criterion",
}

# ------------------------------------------------------------------ Stage 6
monitoring = {
    "signals": {"metrics": True, "log_events": True,
                "traces": {"exporter": True, "enhanced_telemetry_beta": True}},
    "intervals_shortened_for": "short-lived jobs, since metrics default to 60s and traces to 5s",
    "content": {"set": [], "deliberately_unset": ["user prompts", "tool details", "tool content", "raw bodies"],
                "approval_to_change": "pipeline approved to store ticket contents"},
    "attribution": {"session_id": "groups a conversation", "injected_end_user": "names the customer",
                    "why": "default identity attributes name the service credential"},
    "health": "verify arrival at the collector; export errors are silent by default",
    "alerts": [{"signal": "exception_fidelity", "owner": "support-quality", "recovery_tested": True},
               {"signal": "cost_per_successful_task_usd", "owner": "platform", "recovery_tested": True}],
}
stage6 = {"plan": monitoring,
          "every_alert_has_owner": all(a["owner"] for a in monitoring["alerts"]),
          "recovery_tested": all(a["recovery_tested"] for a in monitoring["alerts"])}

# ------------------------------------------------------------------ Checks
assert stage1["all_checkable"] and stage1["deliberately_unmeasured"]
assert stage2["total_cases"] == 1000 and stage2["ambiguous_included"] and stage2["hand_graded"] == 0
assert stage3["changed"] == ["effort"] and stage3["attributable"]
assert stage3["baseline"]["release"] is True
assert stage3["candidate"]["release"] is False and stage3["candidate"]["failed"] == ["exception_fidelity"]
assert stage3["decision"] == "reject"
assert [stage4["diagnoses"][k][0] for k in ("F1", "F2", "F3")] == ["knowledge", "contract", "capability"]
assert stage5["output_unchanged"] is True
assert stage5["total_saving_share"] < stage5["input_saving_share"]
assert cached["per_successful_task"] > cached["per_request"]
assert stage6["every_alert_has_owner"] and stage6["recovery_tested"]

print(json.dumps({"stage1": stage1, "stage2": stage2, "stage3": stage3,
                  "stage4": stage4, "stage5": stage5, "stage6": stage6}, indent=2))
print("\nall stage checks passed")

Acceptance checklist

Every stage 1 criterion has a threshold, a population and a direction.
The not-measured list names documented criteria explicitly.
The evaluation set contains ambiguous cases.
Stage 3 changes exactly one setting, and rejects a candidate with the higher aggregate.
Stage 4 returns three different fault classes with three different levers.
Stage 5's total saving is smaller than its input saving, and per-successful-task exceeds per-request.
Stage 6 leaves every content opt-in unset and gives every alert an owner.

Explain the result before extending it

Three stages have an answer that is easy to produce and hard to justify. Say the justification out loud before moving on.

Stage 3. The candidate is better on the headline number and is rejected. That is only defensible because the exception criterion was written down in stage 1, before any result existed. Written afterwards, it would look like moving the goalposts.
Stage 5. Both savings are real. Only one of them is the bill. A 64% input saving reported as a 64% cost saving is not a lie, it is a category error, and it survives review because everyone in the room wants it to be true.
Stage 6. The plan's most important entries are the empty ones. A content opt-in left unset, recorded with the approval that would change it, is a decision. The same field simply absent is an accident waiting to be discovered by an auditor.

Cleanup

Nothing to clean up: no credentials, no network calls, no files written outside your working directory. Delete unit4_project.py when you are done, or keep it as the starting point for the Unit 5 project.

What this project does not establish

These fixtures test your decisions and interfaces. They do not measure a real model, a real cache, a real bill or a real collector. Passing every stage means your evaluation records the right decisions with the right evidence, which is what this domain's objectives assess. It does not establish production behaviour, and no figure here is a vendor measurement.