Module 7: Observability Latency And Evals In Production

4. Hands-on: the escalation rate metric

Description

This lesson calculates, with 100% deterministic Python code, really run to write this lesson, this entire guide's only AI SLI that is literal with no exception or caveat: the escalation rate. observability/escalation_rate.py takes a fixed set of 50 test events — never random, never datetime.now() — and reuses, without modifying a single line, two pieces of inherited code: parse_manifest() (Module 1, lesson 3) and SHIPMENT_FIELDS_SCHEMA (Module 4, lesson 6). No Bedrock invocation happens anywhere in this lesson — and none is needed, because the decision to escalate or not gets made before Bedrock ever enters the picture.

Connection to the module

Lesson 2 defined the formula; this lesson executes it. This lesson's result isn't an isolated exercise: GENAI-COST-PROFILE.md (Module 2, lesson 8, section 7) already declared, in writing, that the 10% escalation-rate assumption used in its section 4 is a hypothesis to be revised "the first time a real escalation-rate number exists" — and explicitly named this lesson as the source of that number. What follows is, literally, that promise being fulfilled.


Analogy: a traffic light's count, not a survey

A traffic engineer who wants to know what proportion of cars at an intersection turn left doesn't need to survey drivers about their intentions — they put up a camera, count cars for an hour, and divide. The result is a real, verifiable number, reproducible by anyone else who reviews the same recording. Calculating the escalation rate is exactly that kind of counting, not a survey: we don't ask anyone "how often do you think a manifest needs AI?" — we run parse_manifest(), the same real code process-shipment-manifest already executes in production, over a set of manifests, and count how many fail to produce the five fields Shipments needs. There's no opinion at any step of this calculation.


Step 1 — The exact "escalates" criterion, already established, reused here

This guide's Module 1, lesson 3, Exercise 1 already precisely established the criterion, in prose: a manifest escalates — would publish ManifestParseFailed — if parse_manifest() fails to produce all five SHIPMENT_FIELDS_SCHEMA fields, regardless of whether the resulting dictionary is completely empty or partially filled. This lesson turns that sentence into code:

def would_escalate(parsed: dict) -> bool:
    """True if parse_manifest() did not produce the five required fields --
    the exact condition under which process-shipment-manifest would publish
    ManifestParseFailed instead of writing to Shipments (Module 1, lesson 3)."""
    return bool(set(SHIPMENT_FIELDS_SCHEMA) - set(parsed.keys()))

One single line, a set difference: if any SHIPMENT_FIELDS_SCHEMA field is missing from parsed's keys, that difference isn't empty, bool(...) is True, and the event counts as escalated. This doesn't even need to import anything from post_invoke_checks.py — unlike the guardrail block rate (SLI 3), which does validate type and empty values, this criterion, ManifestParseFailed's, only cares about presence of keys, exactly the criterion process-shipment-manifest already applies before attempting to write a record.


Step 2 — The fixed set of 50 test events

Fifty manifests, in a fixed order, never generated at random: 44 well-formed, cycling among the three already-known shipments (4471 Peru→Chile, 4472 Colombia→Ecuador, 4473 Chile→Peru), and six that escalate, at fixed positions, each with a different reason:

PositionTypeContentMissing fields
7Complete free textThe exact email from Module 1, lesson 3 (shipment 4471)all five
15PartialThe exact example from Module 1, lesson 3, Exercise 1 (shipment 4474)3 of 5
23Complete free textA new email, shipment 4475all five
31Partial4471 without weightKg — the same case post_invoke_checks.py already used as an example (Module 4, lesson 6)1 of 5
39Complete free textAn email with no shipment reference at allall five
47Partial4473 with only shipmentId/originCountry/carrier2 of 5

observability/escalation_rate.py, at the root of andes-cargo-infra/:

#!/usr/bin/env python3
"""escalation_rate.py -- computes the real escalation-rate SLI (Module 7,
lesson 2) from a FIXED, deterministic batch of 50 test manifest events.

No Bedrock invocation happens anywhere in this script. The escalation rate
is defined entirely in terms of two pieces of heritage code, both reused
verbatim, never modified:

  - parse_manifest() -- Module 1, lesson 3 of this guide, the same function
    process-shipment-manifest already runs against every uploaded manifest.
  - SHIPMENT_FIELDS_SCHEMA -- Module 4, lesson 6 of this guide
    (guardrails/post_invoke_checks.py), the five-field contract Shipments
    requires.

An event ESCALATES -- would publish ManifestParseFailed on the real system
-- if parse_manifest() does not produce all five fields in
SHIPMENT_FIELDS_SCHEMA. This is exactly the criterion Module 1, lesson 3,
Exercise 1 already established in prose.

Never uses random or datetime.now(). Same 50 events, same order, every run,
on any machine.
"""

from __future__ import annotations

# --- parse_manifest(): Module 1, lesson 3, unmodified ----------------------


def parse_manifest(text: str) -> dict:
    fields = {}
    for line in text.strip().splitlines():
        if "=" in line:
            key, _, value = line.partition("=")
            fields[key.strip()] = value.strip()
    return fields


# --- SHIPMENT_FIELDS_SCHEMA: Module 4, lesson 6, unmodified ----------------

SHIPMENT_FIELDS_SCHEMA = (
    "shipmentId",
    "originCountry",
    "destinationCountry",
    "carrier",
    "weightKg",
)


def would_escalate(parsed: dict) -> bool:
    return bool(set(SHIPMENT_FIELDS_SCHEMA) - set(parsed.keys()))


# --- The three known shipments, cycled for every well-formed event ---------

SHIPMENTS = {
    "4471": {"originCountry": "Peru", "destinationCountry": "Chile", "carrier": "AndesExpress", "weightKg": "120"},
    "4472": {"originCountry": "Colombia", "destinationCountry": "Ecuador", "carrier": "AndesExpress", "weightKg": "85"},
    "4473": {"originCountry": "Chile", "destinationCountry": "Peru", "carrier": "RutaSur", "weightKg": "200"},
}
GOOD_CYCLE = ["4471", "4472", "4473"]


def well_formed_manifest(shipment_id: str) -> str:
    fields = {"shipmentId": shipment_id, **SHIPMENTS[shipment_id]}
    return "\n".join(f"{k}={v}" for k, v in fields.items()) + "\n"


# --- Six escalating events, fixed positions, fixed content ------------

FREE_TEXT_4471 = """Hi team,

Following up on the shipment we discussed on the call. We're sending
120kg of textile goods from our Lima warehouse to the distribution
center in Santiago. AndesExpress is handling the pickup this Thursday.
Shipment reference on our side is AC-4471.

Regards,
Logistics Team
"""

PARTIAL_4474 = """shipmentId=4474
originCountry=Bolivia
Please process this one as priority, the client called twice already.
"""

FREE_TEXT_4475 = """Good morning,

We have a new shipment ready for pickup: 60kg of electronics parts,
origin Quito, destination Lima, carried by RutaSur. Please confirm
once it is scheduled. Our internal reference is EQ-4475.

Best,
Partner Logistics Desk
"""

PARTIAL_4471_NO_WEIGHT = """shipmentId=4471
originCountry=Peru
destinationCountry=Chile
carrier=AndesExpress
"""

FREE_TEXT_NO_REFERENCE = """Hello,

Sending another batch from our Cali warehouse today, similar size to
last week's shipment, same carrier as usual. Will follow up with exact
numbers once the truck is loaded.

Thanks,
Regional Ops
"""

PARTIAL_4473_TWO_MISSING = """shipmentId=4473
originCountry=Chile
carrier=RutaSur
"""

ESCALATING_EVENTS = {
    7: FREE_TEXT_4471,
    15: PARTIAL_4474,
    23: FREE_TEXT_4475,
    31: PARTIAL_4471_NO_WEIGHT,
    39: FREE_TEXT_NO_REFERENCE,
    47: PARTIAL_4473_TWO_MISSING,
}

TOTAL_EVENTS = 50


def build_batch() -> list[tuple[int, str]]:
    """Fixed sequence of 50 manifest texts. Positions 7/15/23/
    31/39/47 are the six escalating events; the rest cycle among the
    three well-formed shipments. Never random, never datetime.now()."""
    batch = []
    good_index = 0
    for i in range(1, TOTAL_EVENTS + 1):
        if i in ESCALATING_EVENTS:
            batch.append((i, ESCALATING_EVENTS[i]))
        else:
            shipment_id = GOOD_CYCLE[good_index % 3]
            good_index += 1
            batch.append((i, well_formed_manifest(shipment_id)))
    return batch


def compute_escalation_rate(batch: list[tuple[int, str]]) -> tuple[int, int, float]:
    escalated = 0
    for _, text in batch:
        parsed = parse_manifest(text)
        if would_escalate(parsed):
            escalated += 1
    total = len(batch)
    rate = round((escalated / total) * 100, 1)
    return escalated, total, rate


def main() -> int:
    batch = build_batch()
    escalated_positions = []
    for index, text in batch:
        parsed = parse_manifest(text)
        if would_escalate(parsed):
            missing = sorted(set(SHIPMENT_FIELDS_SCHEMA) - set(parsed.keys()))
            escalated_positions.append((index, missing))

    escalated, total, rate = compute_escalation_rate(batch)

    print(f"Total test events            {total}")
    print(f"Escalated (ManifestParseFailed) {escalated}")
    print(f"Escalation rate               {rate}%")
    print()
    print("Escalated positions, with missing fields:")
    for index, missing in escalated_positions:
        print(f"  [{index:02d}] missing: {', '.join(missing)}")

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Step 3 — Running the calculation, for real

python3 observability/escalation_rate.py

What to expect (literal — really run to write this lesson):

Total test events            50
Escalated (ManifestParseFailed) 6
Escalation rate               12.0%

Escalated positions, with missing fields:
  [07] missing: carrier, destinationCountry, originCountry, shipmentId, weightKg
  [15] missing: carrier, destinationCountry, weightKg
  [23] missing: carrier, destinationCountry, originCountry, shipmentId, weightKg
  [31] missing: weightKg
  [39] missing: carrier, destinationCountry, originCountry, shipmentId, weightKg
  [47] missing: destinationCountry, weightKg

Six out of fifty, 12.0%. Run the script a second time, on any machine: the output is identical, byte for byte — build_batch() reads no source of randomness or time, so there's no way for two runs to differ.

Notice a detail at positions 31 and 47: the missing-fields list is in alphabetical order (sorted()), not the order in which they appear in SHIPMENT_FIELDS_SCHEMA — it's a presentation decision, not a change to would_escalate()'s criterion, which stays exactly "is anything missing, whatever it is?"


Step 4 — The pytest suite, seven fixed cases

observability/test_escalation_rate.py:

"""pytest suite for escalation_rate.py -- fixed, deterministic manifest
texts only. No random, no datetime.now(). Run with:
    pytest test_escalation_rate.py -v
"""

from escalation_rate import (
    SHIPMENT_FIELDS_SCHEMA,
    build_batch,
    compute_escalation_rate,
    parse_manifest,
    would_escalate,
)


def test_well_formed_manifest_does_not_escalate():
    text = "shipmentId=4471\noriginCountry=Peru\ndestinationCountry=Chile\ncarrier=AndesExpress\nweightKg=120\n"
    parsed = parse_manifest(text)
    assert would_escalate(parsed) is False


def test_empty_free_text_escalates():
    text = "Hi team, following up on the shipment we discussed.\n"
    parsed = parse_manifest(text)
    assert parsed == {}
    assert would_escalate(parsed) is True


def test_partial_manifest_missing_one_field_escalates():
    text = "shipmentId=4471\noriginCountry=Peru\ndestinationCountry=Chile\ncarrier=AndesExpress\n"
    parsed = parse_manifest(text)
    assert would_escalate(parsed) is True


def test_batch_has_fixed_length():
    batch = build_batch()
    assert len(batch) == 50


def test_batch_is_deterministic_across_calls():
    batch_a = build_batch()
    batch_b = build_batch()
    assert batch_a == batch_b


def test_escalation_rate_is_six_of_fifty():
    batch = build_batch()
    escalated, total, rate = compute_escalation_rate(batch)
    assert total == 50
    assert escalated == 6
    assert rate == 12.0


def test_schema_has_five_fields():
    assert len(SHIPMENT_FIELDS_SCHEMA) == 5
cd observability/
pytest test_escalation_rate.py -v -p no:randomly

What to expect (literal — really run):

============================= test session starts ==============================
platform darwin -- Python 3.14.0, pytest-9.1.1, pluggy-1.6.0
rootdir: observability/
collected 7 items

test_escalation_rate.py::test_well_formed_manifest_does_not_escalate PASSED [ 14%]
test_escalation_rate.py::test_empty_free_text_escalates PASSED           [ 28%]
test_escalation_rate.py::test_partial_manifest_missing_one_field_escalates PASSED [ 42%]
test_escalation_rate.py::test_batch_has_fixed_length PASSED              [ 57%]
test_escalation_rate.py::test_batch_is_deterministic_across_calls PASSED [ 71%]
test_escalation_rate.py::test_escalation_rate_is_six_of_fifty PASSED     [ 85%]
test_escalation_rate.py::test_schema_has_five_fields PASSED              [100%]

============================== 7 passed in 0.01s ===============================

Seven out of seven, including test_batch_is_deterministic_across_calls — the test that exists, specifically, to prove in executable code the same guarantee Step 3 already demonstrated by hand: two calls to build_batch(), with nothing different between them, produce identical lists.


Step 5 — Reconciling this number with GENAI-COST-PROFILE.md

Here's the moment this guide's Module 2, lesson 8, section 7 already previewed, with the same explicit-reconciliation discipline that same lesson already applied when comparing the calculator's volume against COST-PROFILE.md:

   GENAI-COST-PROFILE.md, section 4                THIS LESSON, M7.4

   "10% (40 of 400/month)" -- a declared             "12.0% (6 of 50)" -- a real
   HYPOTHESIS, pending a real number                 MEASUREMENT, over a fixed
   (its own section 7 says so)                       TEST batch, not a
                                                      full month of traffic

This lesson's 12.0% doesn't replace GENAI-COST-PROFILE.md's 10%, for the exact same reason Module 2, lesson 8 already explained for that module's lesson 7's 5,000-invocation scenario: fifty test events, hand-built to cover six distinct types of parsing failure, aren't a representative sample of a full month of Andes Cargo's real traffic — the exact same warning sre-and-incident-response-guide, Module 3, lesson 3 already made about its own twenty-manifest batch. What this number is: the first real measurement, over executed code, that the escalation criterion produces a number close to the assumed 10% — not identical, but the same order of magnitude — exactly the kind of first signal GENAI-COST-PROFILE.md, section 7, promised to use to revise, not rewrite from scratch, its own assumption. This module's Module 7, lesson 8 revisits this number when building the complete observability dashboard.


Common mistakes

Treating this lesson's 12.0% as "Andes Cargo's real escalation rate" in any future document, without the "over a fixed test batch" caveat (losing context when citing the number outside this lesson). What happens: someone, in an interview or in M7.8, says "we measured that 12% of Andes Cargo's manifests escalate" without mentioning these are fifty hand-built events, not real production traffic. How to spot it: if your citation of the 12.0% doesn't include, somewhere nearby, the word "test" or "fixed batch." How to fix it: this lesson's Step 5 is explicit about this distinction — fifty test events, deliberately chosen to cover six types of failure, produce a useful signal (the criterion works, the order of magnitude is reasonable), but they are not, nor do they claim to be, a measurement of real traffic. The exact honest statement: "the calculation harness produces 12.0% over a fixed 50-event test batch" — never "Andes Cargo escalates 12% of its manifests."

Modifying SHIPMENT_FIELDS_SCHEMA inside escalation_rate.py, instead of importing it from post_invoke_checks.py (unknowingly duplicating a source of truth). What happens: someone, writing their own version of this script, manually copies the five-field tuple instead of reusing the already-defined constant. How to spot it: if your escalation_rate.py has a second SHIPMENT_FIELDS_SCHEMA definition that doesn't import Module 4's. How to fix it: even though this lesson, for didactic simplicity, locally redefines the tuple (with the exact same value), the correct practice in a real project is to import it directly from guardrails.post_invoke_checks — exactly the same "single source of truth" principle that same module, lesson 6, Exercise 3 already explained: if Andes Cargo ever added a sixth field, that import would guarantee the escalation criterion and the schema validator change together, instead of silently diverging.

Confusing the 12.0% in the output with a percentage of Bedrock invocations that failed (mixing up two different questions). What happens: someone interprets this lesson's result as "12% of the time we invoke the model, the extraction goes wrong." How to spot it: if your explanation of this number mentions Bedrock anywhere. How to fix it: this number has absolutely nothing to do with any extraction's quality — this lesson never invokes Bedrock, not even representatively — it measures, exclusively, what proportion of manifests fail to have the format the deterministic parser expects, a decision made entirely before any model enters the picture. The quality of what Bedrock would do with those escalated manifests is a completely different question, one this same module's M7.5/M7.6 addresses — and one this guide, honestly, can never answer without invoking the real model.


Exercises

Exercise 1 — Modify, yourself, position 39 of the batch (FREE_TEXT_NO_REFERENCE) by adding a shipmentId=4477 line at the beginning of the text, without touching the rest. Before running the script, predict whether the escalation rate changes, and to what new value.

See solution

The escalation rate doesn't change — it stays at 12.0% (6 of 50). Adding shipmentId=4477 makes parse_manifest() produce a dictionary with one key (shipmentId) instead of zero, but would_escalate() still evaluates True, because four of the five required fields are still missing (originCountry, destinationCountry, carrier, weightKg). The criterion is binary — escalates or doesn't — not proportional to how many fields are missing; a manifest missing one field and one missing all five count exactly the same in this formula's numerator. This exercise confirms, with a concrete case, something Step 1 already explained in prose: would_escalate() only cares about presence, not quantity.

Exercise 2 — Explain why compute_escalation_rate() uses round(..., 1) (one decimal) instead of round(..., 2) (two decimals, like Module 2's bedrock_cost_estimate.py does use for dollars). Why is the appropriate precision different in each case?

See solution

One decimal (12.0%, not 12.00%) is enough precision for a rate calculated over 50 events — the difference between 12.0% and 12.04% doesn't change any real operational decision, and adding a second decimal would suggest a precision the sample size doesn't support. bedrock_cost_estimate.py, on the other hand, calculates dollars and cents: two decimals is the natural minimum unit of a currency with cents, not an arbitrary choice of statistical precision. The general rule, applied in both cases: the number of decimals a calculation reports should reflect how meaningful that level of detail is to the decision that number informs, never just "however many decimals Python produces by default."

Exercise 3 — Predict what would happen to this lesson's escalation rate if Andes Cargo, in the real world, started receiving manifests from a new logistics partner who always sends the correct key=value format, but with field names in uppercase (SHIPMENTID=4471 instead of shipmentId=4471). Would that manifest escalate, according to this lesson's exact code?

See solution

Yes, it would escalate — and it's a real case, not an invented edge case. parse_manifest() takes the literal key before the =, with no case normalization (key.strip(), never key.strip().lower() or any variant), so SHIPMENTID and shipmentId are, to parse_manifest(), two completely different keys. The resulting dictionary would have a SHIPMENTID key that doesn't match any of SHIPMENT_FIELDS_SCHEMA's five, so would_escalate() would evaluate True for all five "missing" keys (none of the expected ones are present, even though the information is, under a different case). This is exactly the kind of real case that would explain a genuine jump in the escalation rate — not a system bug, but a new logistics partner with a different case convention — and the kind of investigation a sustained increase in this SLI should trigger, per this module's lesson 2.


Summary and next step

This lesson calculated, with observability/escalation_rate.py really run, the escalation rate over a fixed, deterministic batch of 50 test events: 12.0% (6 of 50), reusing parse_manifest() and SHIPMENT_FIELDS_SCHEMA without modifying a single line of either. You verified the calculation with seven pytest cases, including an explicit determinism-across-runs test. You reconciled this number with the 10% assumption GENAI-COST-PROFILE.md (Module 2, lesson 8, section 7) already left declared as a hypothesis pending revision — the first real data point to measure it against, with the honest caveat that a 50-event test batch isn't production traffic.

Before moving on you should be able to: explain would_escalate()'s exact criterion without help; recite the literal result (12.0%, 6 of 50) and why it's reproducible on any machine; and explain why this number doesn't replace, but complements, GENAI-COST-PROFILE.md's 10%.

Lesson 5 leaves code behind for a moment and answers a purely conceptual question: what a production eval is, and why building the harness that would run one — M7.6's topic — isn't the same as evaluating the quality of what that harness would compare.

Resources

  1. This same course, Module 1, lesson 3 (03-andes-cargos-ai-workload-when-the-deterministic-parser-is-not-enough.md) — the origin of parse_manifest() and the exact escalation criterion, cited verbatim in this lesson's Step 1.
  2. This same course, Module 4, lesson 6 (06-hands-on-the-output-schema-validator.md) — the origin of SHIPMENT_FIELDS_SCHEMA.
  3. This same course, Module 2, lesson 8 (08-project-andes-cargos-genai-cost-profile.md), section 7 — the exact promise this lesson fulfills: the first real escalation-rate number.
  4. sre-and-incident-response-guide, Module 3, lesson 3 — the precedent for "a fixed test batch isn't a production sample," reapplied in this lesson's Step 5.