Advanced Examples

"""Example 51: Quantize a Model."""
 
 
def quantize_int8(weights: list[float], scale: float) -> list[int]:  # => co-19: linear symmetric quantization
    return [round(w / scale) for w in weights]  # => each float mapped to its nearest INT8 code
 
 
def dequantize_int8(quantized: list[int], scale: float) -> list[float]:  # => co-19: reconstruct an APPROXIMATE float
    return [q * scale for q in quantized]  # => the inverse mapping -- approximate, never exact
 
 
weights = [0.12, -0.45, 0.98, -1.2, 0.03]  # => a tiny stand-in for one row of a real weight matrix
scale = max(abs(w) for w in weights) / 127  # => co-19: INT8 spans [-127, 127] -- scale maps the float range onto it
quantized = quantize_int8(weights, scale)  # => the compressed, integer-only representation
dequantized = dequantize_int8(quantized, scale)  # => reconstructed floats -- close, but NOT identical to the originals
print(quantized)  # => Output: [13, -48, 104, -127, 3]
print([round(d, 4) for d in dequantized])  # => Output: [0.1228, -0.4535, 0.9827, -1.2, 0.0283]
 
max_error = max(abs(w - d) for w, d in zip(weights, dequantized))  # => the worst single reconstruction error
print(round(max_error, 4))  # => Output: 0.0035
 
assert max_error < scale  # => co-19: quantization error is bounded by roughly half a scale step, always
# => real quantizers (GPTQ, AWQ) use calibration data instead of this toy min/max scale -- same idea, tighter fit
print("ex-51 OK")  # => a self-check marker confirming reconstruction error stayed within the expected bound
 

[13, -48, 104, -127, 3]
[0.1228, -0.4535, 0.9827, -1.2, 0.0283]
0.0035
ex-51 OK

"""Example 52: Quantization Quality Cost."""
 
 
def synthetic_quality_score(precision_bits: int) -> float:
    # => co-19: illustrative synthetic degradation curve, NOT a published benchmark number -- see Accuracy notes
    baseline = 100.0  # => an arbitrary quality ceiling -- fp16 loses nothing relative to this baseline
    degradation = {16: 0.0, 8: 1.5, 4: 6.0}[precision_bits]  # => illustrative points, not measured data
    return baseline - degradation  # => lower precision, more degradation -- monotonic by construction
 
 
def memory_bytes_per_param(precision_bits: int) -> float:  # => co-19: bytes needed to store ONE parameter
    return precision_bits / 8  # => bits divided by 8 -- the direct bits-to-bytes conversion
 
 
param_count = 7_000_000_000  # => 7B params, a common "small" open-weights model size, illustrative
results: dict[int, tuple[float, float]] = {}  # => precision (bits) -> (quality score, memory in GiB)
for bits in (16, 8, 4):  # => the three precisions this example compares
    quality = synthetic_quality_score(bits)  # => co-19: quality cost, paid for lower precision
    memory_gb = param_count * memory_bytes_per_param(bits) / (1024**3)  # => co-19: memory bought back
    results[bits] = (quality, round(memory_gb, 2))  # => one row of the tradeoff table per precision
print(results)  # => Output: {16: (100.0, 13.04), 8: (98.5, 6.52), 4: (94.0, 3.26)}
 
assert results[4][1] < results[8][1] < results[16][1]  # => co-19: memory shrinks monotonically with precision
assert results[4][0] < results[8][0] < results[16][0]  # => co-19: quality degrades monotonically -- the trade is real
# => this synthetic curve motivates the tradeoff shape -- production choices need real eval-suite numbers
print("ex-52 OK")  # => a self-check marker confirming both monotonic trends held across all three precisions
 

{16: (100.0, 13.04), 8: (98.5, 6.52), 4: (94.0, 3.26)}
ex-52 OK

"""Example 53: Quantization Decision Record."""
 
from dataclasses import dataclass  # => stdlib only -- a decision record needs no framework
 
 
@dataclass
class QuantizationCandidate:  # => co-19: one candidate precision level, already measured (Example 52's shape)
    name: str  # => a label, useful only for reading print output
    precision_bits: int  # => kept for reference -- not directly used in the decision itself
    quality_score: float  # => co-19: the gate this candidate must clear to even be considered
    memory_gb: float  # => co-19: the tiebreaker among candidates that DO clear the gate
 
 
def decide(candidates: list[QuantizationCandidate], quality_floor: float) -> dict[str, str]:
    # => co-19: accept ONLY candidates clearing the quality floor; among those, prefer the smallest memory
    accepted = [c for c in candidates if c.quality_score >= quality_floor]  # => co-19: the SAFETY gate, applied first
    if not accepted:  # => every candidate failed the quality gate -- there is no safe choice
        return {"decision": "reject_all", "reason": f"no candidate clears quality floor {quality_floor}"}
    winner = min(accepted, key=lambda c: c.memory_gb)  # => co-19: among SAFE candidates, smallest wins
    # => "smallest memory that is still safe" is the whole decision rule, in one sentence
    return {"decision": winner.name, "reason": f"smallest memory ({winner.memory_gb} GB) among candidates >= {quality_floor}"}
 
 
candidates = [  # => the SAME three measured candidates from Example 52
    QuantizationCandidate("fp16", 16, 100.0, 13.04),  # => highest quality, largest memory
    QuantizationCandidate("int8", 8, 98.5, 6.52),  # => the usual middle-ground choice
    QuantizationCandidate("int4", 4, 94.0, 3.26),  # => smallest memory, lowest quality
]
record_strict = decide(candidates, quality_floor=97.0)  # => a strict floor rules out int4
record_loose = decide(candidates, quality_floor=90.0)  # => a loose floor admits everything -- smallest wins
# => the SAME three candidates, the SAME decision function -- only the quality floor input changes
print(record_strict)  # => Output: {'decision': 'int8', 'reason': 'smallest memory (6.52 GB) among candidates >= 97.0'}
print(record_loose)  # => Output: {'decision': 'int4', 'reason': 'smallest memory (3.26 GB) among candidates >= 90.0'}
# => this IS a decision record -- the "reason" field is what a real quantization ADR would document
 
assert record_strict["decision"] == "int8"  # => co-19: the floor EXCLUDED int4 despite its smaller memory
# => memory alone never wins -- it only breaks ties AMONG candidates that already passed the gate
assert record_loose["decision"] == "int4"  # => co-19: with a looser floor, the smallest safe option wins
# => same code path, same gate logic -- the OUTCOME changed only because the threshold moved
print("ex-53 OK")  # => a self-check marker confirming the decision flips correctly with the quality floor
 

{'decision': 'int8', 'reason': 'smallest memory (6.52 GB) among candidates >= 97.0'}
{'decision': 'int4', 'reason': 'smallest memory (3.26 GB) among candidates >= 90.0'}
ex-53 OK

"""Example 54: Model-Parallel Split. [GPU] illustrative reference numbers; simulator runs offline, deterministic."""
 
 
def shard_weights_bytes(total_weight_bytes: int, num_devices: int) -> int:
    # => co-20: tensor parallelism splits EACH weight matrix evenly across every device
    return total_weight_bytes // num_devices  # => exact split assumed for this illustration
 
 
def all_reduce_bytes_per_step(activation_bytes: int, num_devices: int) -> int:
    # => co-20: every device must exchange its partial activations with every OTHER device, once per layer
    return activation_bytes * (num_devices - 1) * 2  # => a simplified ring-all-reduce traffic estimate
 
 
total_weight_bytes = 14_000_000_000  # => a 7B-parameter model stored in fp16 (2 bytes/param)
shards = {n: shard_weights_bytes(total_weight_bytes, n) for n in (1, 2, 4, 8)}  # => co-20: one entry per device count
print(shards)  # => Output: {1: 14000000000, 2: 7000000000, 4: 3500000000, 8: 1750000000}
 
activation_bytes = 4_000_000  # => per-step activation volume that must be exchanged between devices
traffic = {n: all_reduce_bytes_per_step(activation_bytes, n) for n in (1, 2, 4, 8)}  # => co-20: traffic per device count
print(traffic)  # => Output: {1: 0, 2: 8000000, 4: 24000000, 8: 56000000}
 
assert shards[8] == total_weight_bytes // 8  # => co-20: each of 8 devices holds exactly one eighth of the weights
assert traffic[8] > traffic[2]  # => co-20: more devices means MORE interconnect traffic per step, not less
# => Example 69 measures what happens when this traffic outgrows the interconnect's real bandwidth
print("ex-54 OK")  # => a self-check marker confirming the shard-shrinks/traffic-grows tradeoff held
 

{1: 14000000000, 2: 7000000000, 4: 3500000000, 8: 1750000000}
{1: 0, 2: 8000000, 4: 24000000, 8: 56000000}
ex-54 OK

"""Example 55: Parallelism Failure Mode."""
 
 
def forward_pass_with_device_failure(num_devices: int, failed_device: int | None) -> str:
    # => co-20: tensor parallelism has NO redundancy -- every device must complete for the result to be valid
    if failed_device is not None and 0 <= failed_device < num_devices:  # => a valid device index actually failed
        return "request_failed: partial result on other devices is unrecoverable, must restart from scratch"
    return "request_succeeded: all devices completed their shard"  # => no failure -- every shard finished normally
 
 
healthy = forward_pass_with_device_failure(4, None)  # => the baseline: no failure anywhere
one_failed = forward_pass_with_device_failure(4, failed_device=2)  # => a SINGLE device out of 4 fails
# => contrast a single-instance deployment: a device failure there costs ONE request, not the fleet
print(healthy)  # => Output: request_succeeded: all devices completed their shard
print(one_failed)  # => Output: request_failed: partial result on other devices is unrecoverable, must restart from scratch
 
assert healthy.startswith("request_succeeded")  # => the no-failure baseline succeeds, as expected
# => co-20: this all-or-nothing coupling is the direct cost of splitting ONE request across devices
assert one_failed.startswith("request_failed")  # => co-20: a SINGLE device failure fails the WHOLE request
print("ex-55 OK")  # => a self-check marker confirming tensor parallelism's all-or-nothing failure mode held
 

request_succeeded: all devices completed their shard
request_failed: partial result on other devices is unrecoverable, must restart from scratch
ex-55 OK

"""Example 56: Workload Length Distribution."""
 
# => co-21: a realistic workload is a DISTRIBUTION of lengths, never one fixed number
LENGTH_BUCKETS = [  # => (output_tokens, relative weight) -- weight is a count, not a random draw
    (50, 40),  # => short replies dominate
    (200, 35),  # => the second-most-common bucket
    (500, 15),  # => a mid-length minority
    (1500, 8),  # => a genuinely long minority
    (4000, 2),  # => rare, very long generations -- the tail that matters most for capacity
]
 
 
def expand_to_workload(buckets: list[tuple[int, int]]) -> list[int]:
    # => co-21: materializes the distribution as an explicit, deterministic list -- no random sampling anywhere
    workload: list[int] = []  # => accumulates one entry per simulated request
    for length, weight in buckets:  # => processes every bucket in the fixed table above
        workload.extend([length] * weight)  # => `weight` copies of this length, deterministically
    return workload  # => a flat list -- 100 request lengths, drawn from a fixed, known distribution
 
 
workload = expand_to_workload(LENGTH_BUCKETS)  # => co-21: the whole distribution, expanded into one list
print(len(workload))  # => Output: 100 -- 100 requests total across all buckets
 
mean_length = sum(workload) / len(workload)  # => the single-number summary this example will show is misleading
print(round(mean_length, 1))  # => Output: 365.0
 
sorted_workload = sorted(workload)  # => sorting is required to read off percentiles by index
p50 = sorted_workload[49]  # => the median request length
p99 = sorted_workload[98]  # => co-21: the tail -- almost the WORST request in this sample
print(p50, p99)  # => Output: 200 4000
 
assert p99 > mean_length  # => co-21: the tail sits FAR above the mean -- a single "average length" hides this
# => capacity planning off the mean alone under-provisions for exactly the requests that hurt most
print("ex-56 OK")  # => a self-check marker confirming the tail-above-mean relationship held
 

100
365.0
200 4000
ex-56 OK

"""Example 57: Realistic Load Test."""
 
from dataclasses import dataclass  # => stdlib only -- driving the engine with a real mix needs no framework
 
 
@dataclass
class SimRequest:  # => one simulated in-flight generation
    id: str  # => a label, useful only for reading print output
    output_tokens: int  # => how many decode steps THIS request needs before it is done
    tokens_emitted: int = 0  # => mutated in place as the simulated batch runs
 
    @property
    def finished(self) -> bool:  # => co-12: the per-request check continuous batching relies on
        return self.tokens_emitted >= self.output_tokens  # => True once this request hits its own target
 
 
def run_continuous_batch(requests: list[SimRequest], max_batch_slots: int) -> int:
    # => co-12/co-22: the SAME continuous-batching engine from Example 31, now driven by a real length mix
    pending = list(requests)  # => the admission queue
    active: list[SimRequest] = []  # => the currently-running slots -- capped by max_batch_slots
    steps = 0  # => the simulated wall-clock -- one tick per decode step, for the WHOLE batch
    while pending or active:  # => keep going until EVERYTHING has both arrived and finished
        while len(active) < max_batch_slots and pending:  # => co-12: fill any FREE slot immediately
            active.append(pending.pop(0))  # => moves ONE request from pending to active, per free slot
        for r in active:  # => every active request takes exactly one step this tick
            r.tokens_emitted += 1  # => the ONLY mutation each active request undergoes per tick
        steps += 1  # => one wall-clock tick has elapsed, regardless of how many slots were full
        active = [r for r in active if not r.finished]  # => co-12: retire finished requests IMMEDIATELY
    return steps  # => total wall-clock steps this realistic-mix workload actually took
 
 
LENGTH_BUCKETS = [(50, 4), (200, 3), (500, 2)]  # => co-21: a small, deterministic stand-in for a real length mix
 
 
def expand_to_workload(buckets: list[tuple[int, int]]) -> list[int]:
    # => co-21: the SAME expansion logic as Example 56, applied to a different bucket table
    out: list[int] = []  # => accumulates one entry per simulated request
    for length, weight in buckets:  # => processes every bucket in the fixed table above
        out.extend([length] * weight)  # => `weight` copies of this length, deterministically
    return out  # => a flat list -- one length per simulated request
 
 
lengths = expand_to_workload(LENGTH_BUCKETS)  # => co-21: 9 requests, mixed lengths -- not one uniform number
requests = [SimRequest(f"r{i}", length) for i, length in enumerate(lengths)]  # => wraps each length as a SimRequest
total_tokens = sum(lengths)  # => the total real work this load test represents
steps = run_continuous_batch(requests, max_batch_slots=4)  # => the SAME engine, now under a realistic length mix
print(total_tokens, steps)  # => Output: 1800 750
# => 1800 total tokens took 750 wall-clock steps -- co-22: NOT one step per token, batching overlaps work
 
throughput = total_tokens / steps  # => co-22: real work delivered per wall-clock step, under this mix
print(round(throughput, 2))  # => Output: 2.4
 
assert steps < total_tokens  # => co-22: WALL-CLOCK steps are far fewer than total tokens -- batching is working
# => this IS the payoff of Example 31's continuous batching, now measured on a realistic length mix
# => Example 58 shows why testing with a SINGLE averaged length instead of this mix hides real behavior
print("ex-57 OK")  # => a self-check marker confirming the engine handled a realistic length mix correctly
 

1800 750
2.4
ex-57 OK

"""Example 58: Load Test That Lies."""
 
 
def simulate_steps(lengths: list[int], max_batch_slots: int) -> int:
    # => co-22: the same continuous-batching mechanics as Example 57, generalized to a plain list of lengths
    pending = list(lengths)  # => the admission queue
    active: list[int] = []  # => remaining tokens for each currently-active request
    steps = 0  # => the simulated wall-clock -- one tick per decode step, for the WHOLE batch
    while pending or active:  # => keep going until EVERYTHING has both arrived and finished
        while len(active) < max_batch_slots and pending:  # => co-12: fill any FREE slot immediately
            active.append(pending.pop(0))  # => moves ONE request from pending to active, per free slot
        active = [r - 1 for r in active]  # => every active request takes one step
        steps += 1  # => one wall-clock tick has elapsed, regardless of how many slots were full
        active = [r for r in active if r > 0]  # => co-12: retire finished requests IMMEDIATELY
    return steps  # => total wall-clock steps this particular length list actually took
 
 
real_lengths = [50, 50, 50, 50, 200, 200, 200, 500, 500]  # => co-21's realistic mixed workload, 9 requests
mean_length = round(sum(real_lengths) / len(real_lengths))  # => co-22: the WRONG shortcut -- "just use the average"
uniform_lengths = [mean_length] * len(real_lengths)  # => a load test that assumes every request costs the SAME
print(mean_length)  # => Output: 200
 
real_steps = simulate_steps(real_lengths, max_batch_slots=4)  # => the TRUE mixed-length workload's cost
uniform_steps = simulate_steps(uniform_lengths, max_batch_slots=4)  # => the "averaged" workload's cost
print(real_steps, uniform_steps)  # => Output: 750 600
 
assert real_steps != uniform_steps  # => co-22: the uniform-length load test measures a DIFFERENT workload than production
# => a load test built on the mean alone can pass in staging and still mislead about production capacity
print("ex-58 OK")  # => a self-check marker confirming the averaged load test disagreed with the real one
 

200
750 600
ex-58 OK

"""Example 59: Capacity Model."""
 
 
def kv_cache_bytes_per_request(num_layers: int, num_heads: int, head_dim: int, seq_len: int, bytes_per_value: int) -> int:
    # => co-06: the SAME cache-size formula from Example 11, reused here as ONE input to capacity planning
    return 2 * num_layers * num_heads * head_dim * seq_len * bytes_per_value  # => *2 accounts for BOTH K and V
 
 
def max_concurrency(cache_budget_bytes: int, bytes_per_request: int) -> int:
    # => co-07/co-21: cache budget, not raw compute, is what SETS the concurrency ceiling
    return cache_budget_bytes // bytes_per_request  # => floor division -- a partial extra request never counts
 
 
cache_budget_bytes = 20 * 1024**3  # => 20 GiB left over after weights and activations -- co-18's remainder
per_request_bytes = kv_cache_bytes_per_request(num_layers=32, num_heads=32, head_dim=128, seq_len=2048, bytes_per_value=2)
# => same formula, same shape as Example 11, applied to a larger 32-layer configuration
print(per_request_bytes)  # => Output: 1073741824 -- exactly 1 GiB per request, at this configuration
 
capacity = max_concurrency(cache_budget_bytes, per_request_bytes)  # => co-21: budget divided by per-request cost
print(capacity)  # => Output: 20 -- 20 GiB of budget divided by 1 GiB per request
 
assert capacity > 0  # => co-18: this GPU can serve SOME concurrent requests at this configuration
assert capacity == cache_budget_bytes // per_request_bytes  # => co-21: capacity planning IS this one division, end to end
# => Example 71 shows what happens when autoscaling reacts to this ceiling too aggressively
print("ex-59 OK")  # => a self-check marker confirming the capacity model reduced to one clean division
 

1073741824
20
ex-59 OK

"""Example 60: GPU Cold Start."""
 
 
def weights_load_time_seconds(model_size_bytes: int, storage_bandwidth_bytes_per_sec: float) -> float:
    # => co-23: loading multi-GB weights from storage into GPU memory dominates a GPU service's cold start
    return model_size_bytes / storage_bandwidth_bytes_per_sec  # => bytes divided by bandwidth -- pure I/O time
 
 
model_size_bytes = 14_000_000_000  # => a 7B-parameter model stored in fp16
storage_bandwidth = 2_000_000_000  # => 2 GB/s, an illustrative, reasonably fast network-attached storage read
cold_start_seconds = weights_load_time_seconds(model_size_bytes, storage_bandwidth)  # => co-23: the I/O floor alone
print(round(cold_start_seconds, 1))  # => Output: 7.0
 
typical_cpu_service_cold_start_seconds = 2.0  # => co-23: a stateless CPU service is usually READY within seconds
ratio = cold_start_seconds / typical_cpu_service_cold_start_seconds  # => how many times slower the GPU service is
print(round(ratio, 2))  # => Output: 3.5
 
assert cold_start_seconds > typical_cpu_service_cold_start_seconds * 2  # => co-23: GPU cold start is FAR slower, not comparable
# => autoscaling policies (Example 71) that assume CPU-service cold-start speed will misbehave on GPU fleets
print("ex-60 OK")  # => a self-check marker confirming GPU cold start dwarfed the CPU-service baseline
 

7.0
3.5
ex-60 OK

"""Example 61: Autoscaling Policy."""
 
 
def naive_scale_out_decision(queue_depth: int, threshold: int) -> bool:
    # => co-23: naive CPU-style policy -- scale out only AFTER the queue is already deep
    return queue_depth > threshold  # => a purely reactive check -- no notion of how long relief will take
 
 
def proactive_scale_out_decision(queue_depth: int, threshold: int, cold_start_seconds: float, arrival_rate_per_sec: float) -> bool:
    # => co-23: GPU-aware policy -- scale out EARLY enough the new replica is ready before the queue overflows
    projected_queue_depth_after_cold_start = queue_depth + arrival_rate_per_sec * cold_start_seconds
    # => co-23: projects forward by exactly the cold-start delay -- the reactive check's blind spot
    return projected_queue_depth_after_cold_start > threshold  # => decides on the PROJECTED depth, not today's
 
 
queue_depth = 8  # => today's queue depth -- looks fine on its own
threshold = 20  # => the same threshold both policies are checked against
naive = naive_scale_out_decision(queue_depth, threshold)  # => only looks at right-now
proactive = proactive_scale_out_decision(queue_depth, threshold, cold_start_seconds=7.0, arrival_rate_per_sec=2.0)
# => co-23: same queue_depth, same threshold -- only the PROJECTION differs between the two calls
print(naive, proactive)  # => Output: False True
# => a naive autoscaler here would start scaling only AFTER the queue has already overflowed
 
assert naive is False  # => co-23: naive policy says "not yet" -- queue is under threshold RIGHT NOW
assert proactive is True  # => co-23: proactive policy already sees the cold start will let the queue overflow
print("ex-61 OK")  # => a self-check marker confirming the reactive-vs-proactive disagreement held
 

False True
ex-61 OK

"""Example 62: Package the Deployment."""
 
from dataclasses import asdict, dataclass  # => stdlib only -- packaging a manifest needs no framework
 
 
@dataclass
class DeploymentManifest:  # => co-24: the served artefact is weights + runtime + config, versioned as ONE unit
    model_id: str  # => which model -- half of "what's actually running in production"
    model_revision: str  # => co-24: pins the EXACT weight snapshot, not just a model family name
    framework_version_pin: str  # => `[Unverified]` placeholder -- see this course's Accuracy notes
    gpu_type: str  # => `[Unverified]` placeholder -- see this course's Accuracy notes
    replica_count: int  # => co-24: how many copies of this EXACT artefact are currently serving traffic
    max_batch_slots: int  # => co-24: a serving-time knob, versioned alongside the weights it runs against
 
 
manifest = DeploymentManifest(
    model_id="example-org/example-7b",  # => illustrative model ID -- co-24: every field is versioned together
    model_revision="a1b2c3d",  # => a specific, reproducible weight snapshot -- not "latest"
    framework_version_pin="[Unverified]-pin-at-deploy-time",  # => co-24: pin the REAL version at actual deploy time
    gpu_type="[Unverified]-pin-at-deploy-time",  # => co-24: pin the REAL hardware at actual deploy time
    replica_count=3,  # => an illustrative fleet size for this worked example
    max_batch_slots=64,  # => an illustrative batching cap for this worked example
)
print(asdict(manifest))
# => Output: {'model_id': 'example-org/example-7b', 'model_revision': 'a1b2c3d',
# => Output: 'framework_version_pin': '[Unverified]-pin-at-deploy-time', 'gpu_type': '[Unverified]-pin-at-deploy-time',
# => Output: 'replica_count': 3, 'max_batch_slots': 64}
# => co-24: every field a rollback needs to reconstruct EXACTLY what was running, in one dict
 
assert manifest.replica_count == 3  # => the manifest round-trips through asdict() without losing any field
# => co-24: this ONE dict is what an incident responder needs first when a served model misbehaves
assert "Unverified" in manifest.framework_version_pin  # => co-24: volatile fields are FLAGGED, never hard-guessed
print("ex-62 OK")  # => a self-check marker confirming the manifest holds every field a real deploy needs
 

{'model_id': 'example-org/example-7b', 'model_revision': 'a1b2c3d', 'framework_version_pin': '[Unverified]-pin-at-deploy-time', 'gpu_type': '[Unverified]-pin-at-deploy-time', 'replica_count': 3, 'max_batch_slots': 64}
ex-62 OK

"""Example 63: Staged Model Rollout."""
 
ROLLOUT_STAGES = [5, 25, 100]  # => co-25: percentage of traffic sent to the NEW model version, staged
 
 
def next_stage(current_stage_index: int, error_rate: float, error_rate_guardrail: float) -> int | None:
    # => co-25: advance ONLY if the current stage is healthy; `None` means "halt right here"
    if error_rate > error_rate_guardrail:  # => co-25: the guardrail is checked BEFORE any advance decision
        return None  # => `None` is the explicit "stop, do not proceed" signal
    if current_stage_index + 1 < len(ROLLOUT_STAGES):  # => is there a wider stage still ahead?
        return current_stage_index + 1  # => co-25: advance to the NEXT wider stage
    return current_stage_index  # => already at 100% -- rollout complete, nothing further to advance to
 
 
healthy_progression: list[int | None] = []  # => records the stage index reached after each check
stage = 0  # => rollout always starts at the narrowest stage
for error_rate in [0.001, 0.002, 0.001]:  # => healthy at every one of the three stages
    stage_result = next_stage(stage, error_rate, error_rate_guardrail=0.01)  # => co-25: one gated check per stage
    healthy_progression.append(stage_result)  # => records whatever this stage's check produced
    stage = stage_result if stage_result is not None else stage  # => only advances the LIVE stage on success
print(healthy_progression)  # => Output: [1, 2, 2]
 
halted_stage = next_stage(0, error_rate=0.05, error_rate_guardrail=0.01)  # => a 5% error rate at 5% traffic
print(halted_stage)  # => Output: None
 
assert healthy_progression[-1] == 2  # => co-25: reached the final stage (index 2 => 100% traffic)
assert halted_stage is None  # => co-25: rollout HALTS at 5% traffic rather than propagating a bad model further
# => Example 73 pairs this same guardrail idea with an automatic rollback trigger
print("ex-63 OK")  # => a self-check marker confirming both the healthy progression and the halt held
 

[1, 2, 2]
None
ex-63 OK

"""Example 64: Serving Observability Dashboard."""
 
from dataclasses import dataclass  # => stdlib only -- aggregating traces needs no framework
 
 
@dataclass
class RequestTrace:  # => co-26/co-16: one completed request's observed metrics
    ttft_ms: float  # => co-16: this request's time-to-first-token
    itl_ms: float  # => co-16: this request's steady-state inter-token latency
    preempted: bool  # => co-14: whether this request was ever evicted mid-generation
 
 
def build_dashboard(traces: list[RequestTrace], queue_depth: int, batch_occupancy: float) -> dict[str, float | int]:
    # => co-26: aggregates the signals that EXPLAIN a serving problem -- queue, occupancy, latency, preemption
    ttfts = sorted(t.ttft_ms for t in traces)  # => sorted so a percentile can be read off by index
    itls = sorted(t.itl_ms for t in traces)  # => sorted so a percentile can be read off by index
    preemption_rate = sum(1 for t in traces if t.preempted) / len(traces)  # => co-14: fraction of traces evicted
    return {  # => co-26: one dashboard snapshot -- every field a real on-call engineer would want
        "queue_depth": queue_depth,  # => co-13: how many requests are waiting, right now
        "batch_occupancy": batch_occupancy,  # => co-12: how full the active batch is, right now
        "ttft_p50_ms": ttfts[len(ttfts) // 2],  # => co-16: the median time-to-first-token
        "itl_p50_ms": itls[len(itls) // 2],  # => co-16: the median inter-token latency
        "preemption_rate": round(preemption_rate, 2),  # => co-14: how often eviction is happening
    }
 
 
traces = [  # => co-26: five completed requests, two of which were preempted mid-generation -- a small fleet snapshot
    RequestTrace(120, 18, False),  # => a normal, uninterrupted request
    RequestTrace(140, 20, False),  # => a normal, uninterrupted request
    RequestTrace(500, 45, True),  # => co-14: this request was PREEMPTED -- its latency reflects that cost
    RequestTrace(130, 19, False),  # => a normal, uninterrupted request
    RequestTrace(600, 50, True),  # => co-14: preempted again -- two of five, a signal worth noticing
]
dashboard = build_dashboard(traces, queue_depth=12, batch_occupancy=0.92)  # => co-26: one call, five signals
print(dashboard)
# => Output: {'queue_depth': 12, 'batch_occupancy': 0.92, 'ttft_p50_ms': 140, 'itl_p50_ms': 20, 'preemption_rate': 0.4}
# => co-26: Example 74 feeds this EXACT shape into a rule chain to diagnose what's actually wrong
 
assert dashboard["preemption_rate"] == 0.4  # => co-26: 2 of 5 traces were preempted -- a real, actionable number
# => a single p50 latency number alone would NEVER have surfaced this preemption signal
# => co-26: dashboards built from raw traces beat dashboards built from pre-averaged summaries
print("ex-64 OK")  # => a self-check marker confirming the dashboard's preemption-rate arithmetic held
# => this is the last co-26 example -- capacity, batching, and preemption all converge into one view
 

{'queue_depth': 12, 'batch_occupancy': 0.92, 'ttft_p50_ms': 140, 'itl_p50_ms': 20, 'preemption_rate': 0.4}
ex-64 OK

"""Example 65: Build-vs-Buy Calculation."""
 
 
def self_hosted_cost_per_million_tokens(gpu_hourly_rate: float, tokens_per_second: float) -> float:
    # => co-27: `[Unverified]` gpu_hourly_rate is an illustrative placeholder -- see this course's Accuracy notes
    tokens_per_hour = tokens_per_second * 3600  # => converts a per-second rate into an hourly one
    return (gpu_hourly_rate / tokens_per_hour) * 1_000_000  # => cost per million tokens, at FULL utilization
 
 
def hosted_api_cost_per_million_tokens(price_per_token: float) -> float:
    # => co-27: `[Unverified]` price_per_token is an illustrative placeholder -- see this course's Accuracy notes
    return price_per_token * 1_000_000  # => a simple linear scale -- no utilization assumption needed
 
 
gpu_hourly_rate = 2.00  # => `[Unverified]` illustrative placeholder, NOT a current market price
tokens_per_second = 800.0  # => sustained aggregate throughput at a chosen batch size
self_hosted = self_hosted_cost_per_million_tokens(gpu_hourly_rate, tokens_per_second)  # => co-27: the "build" side
hosted_api = hosted_api_cost_per_million_tokens(price_per_token=0.000002)  # => `[Unverified]` illustrative placeholder
print(round(self_hosted, 4), round(hosted_api, 4))  # => Output: 0.6944 2.0
 
assert self_hosted < hosted_api  # => co-27: at THIS illustrative, FULL utilization, self-hosting is cheaper per token
# => Example 75 revisits this exact comparison under LESS than full utilization -- the answer can flip
print("ex-65 OK")  # => a self-check marker confirming the build-vs-buy comparison held under these assumptions
 

0.6944 2.0
ex-65 OK

"""Example 66: Quantization INT8 vs INT4 Memory."""
 
 
def model_memory_gb(param_count: int, bits_per_param: int) -> float:  # => co-19: total storage for ALL parameters
    return param_count * bits_per_param / 8 / (1024**3)  # => params times bits, converted bits->bytes->GiB
 
 
param_count = 7_000_000_000  # => the SAME 7B parameter count used throughout the quantization examples
memory_by_precision = {bits: round(model_memory_gb(param_count, bits), 2) for bits in (16, 8, 4)}  # => co-19: three points
print(memory_by_precision)  # => Output: {16: 13.04, 8: 6.52, 4: 3.26}
 
savings_int8_vs_fp16 = memory_by_precision[16] - memory_by_precision[8]  # => co-19: the middle-ground precision's savings
savings_int4_vs_fp16 = memory_by_precision[16] - memory_by_precision[4]  # => co-19: the aggressive precision's savings
print(round(savings_int8_vs_fp16, 2), round(savings_int4_vs_fp16, 2))  # => Output: 6.52 9.78
 
assert savings_int4_vs_fp16 > savings_int8_vs_fp16  # => co-19: INT4 saves MORE than INT8, at a steeper quality cost
# => Example 67 shows a middle path -- blending precisions instead of picking one for the whole model
print("ex-66 OK")  # => a self-check marker confirming INT4's larger memory saving held over INT8
 

{16: 13.04, 8: 6.52, 4: 3.26}
6.52 9.78
ex-66 OK

"""Example 67: Mixed-Precision Serving."""
 
 
def blended_memory_gb(param_count: int, fp16_fraction: float, quantized_bits: int) -> float:
    # => co-19: keep a FRACTION of params at fp16 (typically attention/output layers) and quantize the rest
    fp16_params = param_count * fp16_fraction  # => the portion kept at full precision, for quality
    quantized_params = param_count * (1 - fp16_fraction)  # => the remainder, quantized down for memory
    fp16_bytes = fp16_params * 16 / 8  # => full-precision portion's byte cost
    quantized_bytes = quantized_params * quantized_bits / 8  # => quantized portion's byte cost
    return (fp16_bytes + quantized_bytes) / (1024**3)  # => the two portions summed, converted to GiB
 
 
param_count = 7_000_000_000  # => the SAME 7B parameter count used throughout the quantization examples
pure_int8 = blended_memory_gb(param_count, fp16_fraction=0.0, quantized_bits=8)  # => nothing kept at fp16
mixed_10pct_fp16 = blended_memory_gb(param_count, fp16_fraction=0.10, quantized_bits=8)  # => 10% kept for quality
pure_fp16 = blended_memory_gb(param_count, fp16_fraction=1.0, quantized_bits=8)  # => everything at full precision
print(round(pure_int8, 2), round(mixed_10pct_fp16, 2), round(pure_fp16, 2))  # => Output: 6.52 7.17 13.04
 
assert pure_int8 < mixed_10pct_fp16 < pure_fp16  # => co-19: mixed precision sits STRICTLY between the two pure extremes
# => production quantizers pick this fraction per-layer, not as one global knob, for the same reason
print("ex-67 OK")  # => a self-check marker confirming mixed precision lands strictly between the two extremes
 

6.52 7.17 13.04
ex-67 OK

"""Example 68: Tensor-Parallel vs Pipeline-Parallel."""
 
 
def tensor_parallel_traffic_per_step(activation_bytes: int, num_devices: int) -> int:
    # => co-20: TP exchanges activations on EVERY layer -- frequent, and it adds up fast
    num_layers = 32  # => a stand-in transformer depth, used only to size the traffic estimate
    return activation_bytes * (num_devices - 1) * num_layers  # => co-20: traffic scales with BOTH devices AND layers
 
 
def pipeline_parallel_traffic_per_step(activation_bytes: int, num_stages: int) -> int:
    # => co-20: PP exchanges activations only at STAGE BOUNDARIES -- rarer, but adds pipeline "bubble" idle time
    return activation_bytes * (num_stages - 1)  # => co-20: traffic scales with stages, NOT with layer count
 
 
activation_bytes = 4_000_000  # => per-layer-boundary activation volume that must be exchanged
tp_traffic = tensor_parallel_traffic_per_step(activation_bytes, num_devices=4)  # => co-20: the layer-by-layer cost
pp_traffic = pipeline_parallel_traffic_per_step(activation_bytes, num_stages=4)  # => co-20: the boundary-only cost
print(tp_traffic, pp_traffic)  # => Output: 384000000 12000000
 
assert tp_traffic > pp_traffic  # => co-20: TP moves FAR more data per step -- it needs a fast interconnect to pay off
# => Example 69 converts this traffic gap into actual wall-clock milliseconds on real links
print("ex-68 OK")  # => a self-check marker confirming TP's far higher per-step traffic held
 

384000000 12000000
ex-68 OK

"""Example 69: Interconnect Bottleneck Measurement. [GPU] illustrative reference numbers."""
 
 
def all_reduce_time_ms(traffic_bytes: int, interconnect_bandwidth_bytes_per_sec: float) -> float:
    # => co-20: converts Example 68's byte count into WALL-CLOCK time, given a link's real bandwidth
    return traffic_bytes / interconnect_bandwidth_bytes_per_sec * 1000  # => seconds converted to milliseconds
 
 
traffic_bytes = 384_000_000  # => co-20: tensor-parallel traffic per step, from Example 68
fast_interconnect = 300_000_000_000  # => 300 GB/s, illustrative -- e.g. a high-end multi-GPU link
slow_interconnect = 10_000_000_000  # => 10 GB/s, illustrative -- e.g. commodity networking between hosts
 
fast_time = all_reduce_time_ms(traffic_bytes, fast_interconnect)  # => co-20: the SAME traffic, fast link
slow_time = all_reduce_time_ms(traffic_bytes, slow_interconnect)  # => co-20: the SAME traffic, slow link
print(round(fast_time, 3), round(slow_time, 2))  # => Output: 1.28 38.4
 
decode_step_budget_ms = 20.0  # => co-15/co-16: this topic's own decode-step cost model, from Example 33
assert fast_time < decode_step_budget_ms  # => co-20: a fast interconnect adds a NEGLIGIBLE fraction of the step budget
assert slow_time > decode_step_budget_ms  # => co-20: a slow interconnect ALONE exceeds the entire step budget
# => choosing TP over PP without checking link bandwidth first can turn a compute win into a network loss
print("ex-69 OK")  # => a self-check marker confirming the interconnect-speed-vs-step-budget comparison held
 

1.28 38.4
ex-69 OK

"""Example 70: P50/P99 Latency Under Load."""
 
 
def itl_at_batch(batch_size: int) -> float:  # => co-15/co-16: the SAME cost model as Example 33
    return 15.0 + 1.0 * batch_size  # => the SAME fixed-floor-plus-per-slot formula, reused here
 
 
def simulate_latencies_under_load(batch_sizes_over_time: list[int]) -> list[float]:
    # => co-22: latency observed by requests DEPENDS on how full the batch was WHEN they were served
    return [itl_at_batch(b) for b in batch_sizes_over_time]  # => one latency sample per observed batch size
 
 
# => co-22: a realistic trace -- batch size fluctuates with arrivals, occasionally spiking
batch_trace = [4, 4, 5, 4, 32, 4, 4, 5, 4, 4]  # => one spike to batch=32 among mostly-light load
latencies = simulate_latencies_under_load(batch_trace)  # => co-22: the resulting per-request latency samples
sorted_latencies = sorted(latencies)  # => sorting is required to read off percentiles by index
p50 = sorted_latencies[len(sorted_latencies) // 2]  # => the median observed latency
p99_index = min(len(sorted_latencies) - 1, int(len(sorted_latencies) * 0.99))  # => clamped so it never overflows
p99 = sorted_latencies[p99_index]  # => co-22: the tail -- almost the WORST observed latency
print(sorted_latencies)  # => Output: [19.0, 19.0, 19.0, 19.0, 19.0, 19.0, 19.0, 20.0, 20.0, 47.0]
print(p50, p99)  # => Output: 19.0 47.0
 
assert p99 > p50 * 1.5  # => co-22: the tail is dominated by the rare spike -- far above the typical case
print("ex-70 OK")  # => a self-check marker confirming the tail-dominated-by-spike relationship held
 

[19.0, 19.0, 19.0, 19.0, 19.0, 19.0, 19.0, 20.0, 20.0, 47.0]
19.0 47.0
ex-70 OK

"""Example 71: Autoscaling Thrashing."""
 
 
def naive_threshold_decision(queue_depth: int, threshold: int) -> str:
    # => co-23: a SINGLE threshold with no gap flips direction the instant queue depth crosses it, either way
    if queue_depth > threshold:  # => any tick above threshold triggers an immediate scale-out
        return "scale_out"
    if queue_depth < threshold:  # => any tick below threshold triggers an immediate scale-in
        return "scale_in"
    return "hold"  # => exactly at threshold -- the only tick that does nothing
 
 
def hysteresis_decision(queue_depth: int, threshold: int, cooldown_active: bool) -> str:
    # => co-23: fix -- ANY scaling action arms a cooldown that suppresses the very next decision
    if cooldown_active:  # => co-23: a cooldown from the PREVIOUS tick overrides whatever this tick would decide
        return "hold_cooldown"
    return naive_threshold_decision(queue_depth, threshold)  # => no active cooldown -- decide normally
 
 
queue_trace = [21, 19, 22, 18, 21]  # => co-23: queue depth oscillating right around threshold=20
# => this is a deliberately adversarial trace -- real traffic rarely bounces this tightly, but it CAN
naive_decisions = [naive_threshold_decision(q, threshold=20) for q in queue_trace]  # => co-23: no memory between ticks
print(naive_decisions)  # => Output: ['scale_out', 'scale_in', 'scale_out', 'scale_in', 'scale_out']
 
naive_scaling_events = sum(1 for d in naive_decisions if d != "hold")  # => co-23: count of REAL scaling actions
print(naive_scaling_events)  # => Output: 5 -- every single tick triggers a REVERSAL, this is thrashing
 
hysteresis_decisions: list[str] = []  # => records the decision made at each tick, cooldown-aware
cooldown = False  # => starts clear -- the first tick is never suppressed
for q in queue_trace:  # => processes the SAME queue trace as the naive policy, for a fair comparison
    decision = hysteresis_decision(q, threshold=20, cooldown_active=cooldown)  # => co-23: cooldown-aware decision
    hysteresis_decisions.append(decision)  # => records this tick's outcome before the cooldown updates
    cooldown = decision in ("scale_out", "scale_in")  # => co-23: any REAL action arms the cooldown for next tick
print(hysteresis_decisions)  # => Output: ['scale_out', 'hold_cooldown', 'scale_out', 'hold_cooldown', 'scale_out']
 
hysteresis_scaling_events = sum(1 for d in hysteresis_decisions if d in ("scale_out", "scale_in"))  # => co-23: real actions only
print(hysteresis_scaling_events)  # => Output: 3
 
assert naive_scaling_events > hysteresis_scaling_events  # => co-23: hysteresis cuts thrashing roughly in half here
# => co-23: the SAME idea that fixes preemption thrashing (Example 46) also fixes scaling thrashing
# => a cooldown window is a general pattern: any policy re-evaluated too fast can start oscillating
# => this trace also stresses cold-start cost (Example 60) -- every reversal pays that penalty again
print("ex-71 OK")  # => a self-check marker confirming hysteresis measurably reduced the thrashing
 

['scale_out', 'scale_in', 'scale_out', 'scale_in', 'scale_out']
5
['scale_out', 'hold_cooldown', 'scale_out', 'hold_cooldown', 'scale_out']
3
ex-71 OK

"""Example 72: Canary Rollout With Metrics Guardrail."""
 
 
def evaluate_canary_stage(error_rate: float, p99_latency_ms: float, error_guardrail: float, latency_guardrail_ms: float) -> str:
    # => co-25: a stage is healthy ONLY if BOTH guardrails hold -- either one alone is not enough
    if error_rate > error_guardrail:  # => the FIRST gate -- checked before latency is even looked at
        return "halt: error_rate_guardrail_breached"
    if p99_latency_ms > latency_guardrail_ms:  # => co-25: a SECOND, independent gate -- errors alone are not enough
        return "halt: latency_guardrail_breached"
    return "advance"  # => both gates cleared -- the stage is genuinely healthy
 
 
stage_results = [  # => co-25: the SAME guardrails, three different simulated health states
    evaluate_canary_stage(0.002, 180.0, error_guardrail=0.01, latency_guardrail_ms=250.0),  # => healthy on both axes
    evaluate_canary_stage(0.002, 400.0, error_guardrail=0.01, latency_guardrail_ms=250.0),  # => latency regression only
    evaluate_canary_stage(0.05, 180.0, error_guardrail=0.01, latency_guardrail_ms=250.0),  # => error spike only
]
print(stage_results)
# => Output: ['advance', 'halt: latency_guardrail_breached', 'halt: error_rate_guardrail_breached']
# => co-25: this contrasts with Example 63, which only checked ONE signal (error rate alone)
 
assert stage_results[0] == "advance"  # => co-25: healthy on BOTH axes -- correctly allowed to proceed
# => "advance" here means the NEXT wider traffic stage becomes eligible, same idea as Example 63
assert stage_results[1] == "halt: latency_guardrail_breached"  # => co-25: a latency-ONLY regression is caught too
# => an error-rate-only rollout gate would have MISSED this regression entirely
assert stage_results[2] == "halt: error_rate_guardrail_breached"  # => co-25: an error-ONLY spike is caught too
print("ex-72 OK")  # => a self-check marker confirming both independent guardrails caught their own failure mode
# => co-25: a real canary would also gate on custom business metrics, not just error rate and latency
 

['advance', 'halt: latency_guardrail_breached', 'halt: error_rate_guardrail_breached']
ex-72 OK

"""Example 73: Rollback on Latency Regression."""
 
 
def should_rollback(old_version_p99_ms: float, new_version_p99_ms: float, regression_tolerance: float) -> bool:
    # => co-25/co-26: rollback if the NEW version's tail latency regresses beyond a tolerated multiple of the OLD one
    return new_version_p99_ms > old_version_p99_ms * (1 + regression_tolerance)  # => a simple relative-threshold check
 
 
old_p99 = 200.0  # => the previous version's known-good tail latency
new_p99_ok = 220.0  # => 10% worse -- within a 15% tolerance
new_p99_bad = 300.0  # => 50% worse -- breaches a 15% tolerance
 
decision_ok = should_rollback(old_p99, new_p99_ok, regression_tolerance=0.15)  # => co-25: the SAME tolerance, modest regression
decision_bad = should_rollback(old_p99, new_p99_bad, regression_tolerance=0.15)  # => co-25: the SAME tolerance, large regression
print(decision_ok, decision_bad)  # => Output: False True
 
assert decision_ok is False  # => co-25: a modest regression stays within tolerance -- roll forward
assert decision_bad is True  # => co-26: a large regression is caught by the SAME observability signal and rolled back
# => pairing this check with the staged rollout in Example 63 halts a bad model before it reaches 100%
print("ex-73 OK")  # => a self-check marker confirming the tolerance correctly separated OK from bad regressions
 

False True
ex-73 OK

"""Example 74: Full Observability Explains an Incident."""
 
 
def diagnose(dashboard: dict[str, float | int]) -> str:
    # => co-26: a simple, deterministic rule chain -- real observability tooling automates exactly this reasoning
    if dashboard["preemption_rate"] > 0.3:  # => co-14: the FIRST, highest-priority signal this rule chain checks
        return "likely cause: cache pressure causing frequent preemption -- check admission control / batch size"
    if dashboard["queue_depth"] > 15 and dashboard["batch_occupancy"] < 0.5:  # => co-13/co-12: queueing WHILE underused
        return "likely cause: undersized replica count -- requests queueing while GPU sits underused"
    if dashboard["itl_p50_ms"] > 40:  # => co-15: the LAST signal checked -- a pure latency/throughput tradeoff symptom
        return "likely cause: oversized batch -- trading latency for throughput beyond the configured SLO"
    return "no anomaly detected"  # => none of the three rules fired -- the dashboard looks healthy
 
 
incident_dashboard = {"queue_depth": 12, "batch_occupancy": 0.92, "itl_p50_ms": 20, "preemption_rate": 0.4}
# => co-26: the SAME shape Example 64's build_dashboard() produced -- preemption_rate is the giveaway here
healthy_dashboard = {"queue_depth": 3, "batch_occupancy": 0.6, "itl_p50_ms": 18, "preemption_rate": 0.05}
# => low queue, low occupancy, low preemption -- none of the three rules above have a reason to fire
 
print(diagnose(incident_dashboard))
# => Output: likely cause: cache pressure causing frequent preemption -- check admission control / batch size
print(diagnose(healthy_dashboard))  # => Output: no anomaly detected
# => the SAME function, the SAME rule chain -- only the input dashboard's numbers changed
 
assert diagnose(incident_dashboard).startswith("likely cause: cache pressure")  # => co-26: correctly diagnosed the incident
# => this is the payoff of everything from co-06 through co-26: one dashboard number resolves to one root cause
assert diagnose(healthy_dashboard) == "no anomaly detected"  # => co-26: observability also confirms when NOTHING is wrong
print("ex-74 OK")  # => a self-check marker confirming the rule chain diagnosed both dashboards correctly
# => capacity, batching, memory, deployment, and observability all converge into this one function
# => a real on-call runbook is this same if-chain, just with links to the matching dashboard panels
 

likely cause: cache pressure causing frequent preemption -- check admission control / batch size
no anomaly detected
ex-74 OK

"""Example 75: TCO Sensitivity to Utilization."""
 
 
def tco_per_million_tokens(gpu_hourly_rate: float, tokens_per_second_at_full_load: float, utilization: float) -> float:
    # => co-28: the GPU bills for EVERY hour it's on, but only serves tokens during utilized hours
    effective_tokens_per_hour = tokens_per_second_at_full_load * 3600 * utilization  # => co-28: idle hours produce ZERO tokens
    return (gpu_hourly_rate / effective_tokens_per_hour) * 1_000_000  # => the SAME formula as Example 65, utilization-scaled
 
 
gpu_hourly_rate = 2.00  # => `[Unverified]` illustrative placeholder -- see this course's Accuracy notes
tokens_per_second_at_full_load = 800.0  # => the SAME peak throughput used in Example 65
hosted_api_cost = 2.00  # => co-27's hosted-API comparison point from Example 65, per million tokens
 
tco_by_utilization = {u: round(tco_per_million_tokens(gpu_hourly_rate, tokens_per_second_at_full_load, u), 4) for u in (0.9, 0.5, 0.1)}
# => co-28: the SAME hardware, the SAME peak throughput -- only the UTILIZATION assumption changes here
print(tco_by_utilization)  # => Output: {0.9: 0.7716, 0.5: 1.3889, 0.1: 6.9444}
 
assert tco_by_utilization[0.1] > hosted_api_cost  # => co-28: at LOW utilization, self-hosting is WORSE than the API
assert tco_by_utilization[0.9] < hosted_api_cost  # => co-28: at HIGH utilization, self-hosting is BETTER
# => the whole course closes on this point: infrastructure decisions are workload-shape decisions, not fixed answers
print("ex-75 OK")  # => a self-check marker confirming utilization alone flips the build-vs-buy verdict
 

{0.9: 0.7716, 0.5: 1.3889, 0.1: 6.9444}
ex-75 OK