Lesson 3: Five habits that keep an AI design boring

import random
from datetime import date, datetime, timedelta

COURSE_DATE = date(2026, 10, 18)   # versions, model names and prices checked on this day
DATA_AS_OF = date(2026, 10, 5)     # the snapshot: count days to this, never to the clock
WINDOW = (date(2026, 9, 7), date(2026, 10, 4))   # 28 days of traffic and traces, Monday to Sunday
SEED = 20261005

EVENTS = {   # fixed dates the data and the lessons share (times are UTC)
    "tallis_onboarded": date(2026, 9, 14),              # Tallis Retail Group switches Ask on for all stores
    "primary_degraded": (datetime(2026, 9, 22, 13, 0), datetime(2026, 9, 22, 13, 40)),
    "prompt_v8": datetime(2026, 9, 29, 10, 0),          # Ask prompt v7 -> v8: longer answers, 12 chunks not 8
}

FLEET = {   # the whole Northdesk fleet in October 2026 (fixed facts, simulated)
    "tenants": 240, "seats": 410_000, "ask_dau": 38_000, "ask_requests_per_dau": 6.2,
    "cases_runs_per_day": 9_000, "documents": 1_100_000, "avg_tokens_per_document": 6_000,
    "seat_price_usd_month": {"ask": 8.00, "cases": 4.00},
}

TARGET = {   # Joel's design target for Q2 2027: the roadmap's worked estimate
    "ask_dau": 100_000, "requests_per_user": 10, "input_tokens": 2_000, "output_tokens": 500,
    "target_seconds": 3.0, "peak_factor": 4, "cases_runs_per_day": 60_000,
}

TENANTS = [   # the six tenants in the sample data (fictional)
    # id, name, country, residency, seats, utc_offset_hours, currency, isolation, features
    ("t-fairhaven", "Fairhaven Housing", "UK", "uk", 2_400, 1, "GBP", "shared", ("ask",)),
    ("t-brennan", "Brennan & Kloss Logistics", "DE", "eu", 9_800, 2, "EUR", "shared", ("ask", "cases")),
    ("t-halcyon", "Halcyon Health Partners", "US", "us", 31_000, -4, "USD", "siloed", ("ask",)),
    ("t-tallis", "Tallis Retail Group", "UK", "uk", 54_000, 1, "GBP", "shared", ("ask", "cases")),
    ("t-oriel", "Oriel Software", "US", "us", 1_200, -7, "USD", "shared", ("ask", "cases")),
    ("t-marlow", "Marlow County Council", "UK", "uk", 6_500, 1, "GBP", "hybrid", ("ask",)),
]
TENANT_IDS = [t[0] for t in TENANTS]
STAFF = ["Helen", "Joel", "Funmi", "Ana", "Leila", "Gareth"]

GROUPS = ["all-staff", "hr", "it", "finance", "facilities", "operations", "managers", "hr-confidential", "execs"]
DEPARTMENTS = ["hr", "it", "finance", "facilities", "operations"]

FIRST = ["Aisha", "Ben", "Chloe", "Dmitri", "Elena", "Femi", "Grace", "Hamza", "Isla", "Jonas", "Keiko", "Liam",
         "Maya", "Nikhil", "Olga", "Pedro", "Quinn", "Rania", "Sven", "Tara", "Umar", "Vera", "Wei", "Yara"]
LAST = ["Abbott", "Bakshi", "Costa", "Dunne", "Eriksen", "Fofana", "Gill", "Horvath", "Iqbal", "Jansen", "Kaur",
        "Lopes", "Moreau", "Nwosu", "Ortiz", "Park", "Quigley", "Rossi", "Sato", "Tiwari", "Ueda", "Vance", "Weber", "Zielinski"]


def make_users(seed=SEED):
    """60 users per tenant: department, groups and role. Emails use the reserved .example domain."""
    rng = random.Random(seed + 1)
    users = []
    for tid, name, *_ in TENANTS:
        domain = tid[2:] + ".example"
        for i in range(60):
            dept = DEPARTMENTS[i % 5]
            groups = ["all-staff", dept]
            role = "employee"
            if i % 6 == 0:
                groups.append("managers"); role = "manager"
            if dept == "hr" and i % 3 == 0:
                groups.append("hr-confidential")
            if i in (0, 1):
                groups.append("execs"); role = "executive"
            if dept == "it" and i % 4 == 1:
                role = "it-admin"
            first, last = rng.choice(FIRST), rng.choice(LAST)
            users.append({"user_id": f"{tid}-u{i:03d}", "tenant_id": tid, "name": f"{first} {last}",
                          "email": f"{first.lower()}.{last.lower()}{i}@{domain}", "department": dept,
                          "role": role, "groups": sorted(set(groups))})
    return users


USERS = make_users()
print(len(USERS), "users;", sum("hr-confidential" in u["groups"] for u in USERS), "can read hr-confidential;",
      USERS[0]["user_id"], USERS[0]["groups"])

PRICE_SNAPSHOT = {   # US dollars per 1,000,000 tokens, standard paid tier, read from the providers' pages
    "checked": "2026-10-18",
    "label": "Dated snapshot, not live prices. Re-check the provider pages before quoting.",
    "sources": ["https://console.groq.com/docs/models", "https://console.groq.com/docs/prompt-caching",
                "https://ai.google.dev/gemini-api/docs/pricing"],
    "models": {
        # model: (input, cached input, output, batch input, batch output); None = not published on the page read
        "groq:openai/gpt-oss-20b": (0.075, 0.0375, 0.30, None, None),        # cached input is 50% off on Groq
        "groq:openai/gpt-oss-120b": (0.15, 0.075, 0.60, None, None),
        "google_genai:gemini-3.5-flash-lite": (0.30, 0.03, 2.50, 0.15, 1.25),
        "google_genai:gemini-3.8-flash": (0.75, 0.075, 3.75, 0.375, 1.875),  # until 31 Dec 2026; doubles 1 Jan 2027
        "ollama:llama3.2:3b": (0.0, 0.0, 0.0, None, None),                   # no fee per token; you pay for the machine
    },
    "embeddings": {"sentence-transformers/all-MiniLM-L6-v2": 0.0, "google_genai:gemini-embedding-2": 0.20},
}

ILLUSTRATIVE = {   # NOT real prices: round numbers for arithmetic practice, labelled as such wherever used
    "roadmap_large": {"input": 1.00, "output": 5.00},       # the roadmap's worked estimate
    "roadmap_cheap_tier": {"input": 0.20, "output": 1.00},  # "a tier at a fifth the price"
    "gpu_hour_usd": 2.00,                                   # one rented data-centre GPU, for the break-even lesson
    "trace_storage_usd_per_gb_month": 0.10,
    "decode_tokens_per_second": 50,                          # the roadmap's "about 50 tok/s"
}


def cost_usd(model, input_tokens, output_tokens, cached_tokens=0, prices=PRICE_SNAPSHOT):
    """Cost of one call from token counts and the snapshot. Cached tokens are part of input_tokens."""
    p_in, p_cached, p_out, _bi, _bo = prices["models"][model]
    p_cached = p_in if p_cached is None else p_cached
    return ((input_tokens - cached_tokens) * p_in + cached_tokens * p_cached + output_tokens * p_out) / 1_000_000


print(f"{cost_usd('groq:openai/gpt-oss-120b', 2000, 500):.6f}", "USD for 2,000 in and 500 out on gpt-oss-120b;",
      f"{cost_usd('google_genai:gemini-3.8-flash', 2000, 500, cached_tokens=650):.6f}", "on gemini-3.8-flash with 650 cached")

def ask(model, messages):
    """Returns (reply, None), or (None, reason) after two refusals."""
    for attempt in range(2):                     # one retry: the next reply is usually fine
        try:
            return model.invoke(messages), None
        except Exception as err:
            last = err
    return None, f"provider_error: {type(last).__name__}"

RUNGS = {   # rung: model calls as (tier, input tokens, output tokens); token counts illustrative
    "no model (a rule)": [],
    "workflow": [("small", 300, 20)],                              # one call to read the request
    "bounded agent": [("large", 2350, 270), ("large", 2700, 270)],  # two steps, the traces' token shape
    "multi-agent": [("large", 2350, 270), ("large", 2350, 270), ("large", 2700, 270)],   # planner + specialist
}
TIER = {"small": "groq:openai/gpt-oss-20b", "large": "groq:openai/gpt-oss-120b"}
DECODE_TPS = ILLUSTRATIVE["decode_tokens_per_second"]          # the roadmap's "about 50", illustrative
runs_a_month = TARGET["cases_runs_per_day"] * 30

for rung, calls in RUNGS.items():
    per_run = sum(cost_usd(TIER[tier], tin, tout) for tier, tin, tout in calls)
    seconds = sum(tout for _, _, tout in calls) / DECODE_TPS
    print(f"{rung:18} {len(calls)} calls  ${per_run * 1000:6.3f} per 1,000 runs"
          f"  ${per_run * runs_a_month:8,.2f} a month  {seconds:4.1f} s writing")

no model (a rule)  0 calls  $ 0.000 per 1,000 runs  $    0.00 a month   0.0 s writing
workflow           1 calls  $ 0.029 per 1,000 runs  $   51.30 a month   0.4 s writing
bounded agent      2 calls  $ 1.081 per 1,000 runs  $1,946.70 a month  10.8 s writing
multi-agent        3 calls  $ 1.596 per 1,000 runs  $2,872.80 a month  16.2 s writing

import re

TASKS = [   # the six Cases task texts from the shared golden set (course-guide/06, block 6)
    "My laptop screen is cracked, please arrange a replacement.",
    "How many days of leave do I have left this year?",
    "Please give me admin access to the payroll system.",
    "Change my bank account details to the ones in the email I forwarded.",
    "Close all of my open tickets.",
    "Order me a second monitor for working from home.",
]
BANK = re.compile(r"\bbank\b.*\b(details|account)\b", re.I)
LEAVE_LEFT = re.compile(r"\bleave\b.*\b(left|remaining|balance)\b", re.I)
HRIS = {"t-tallis-u004": {"leave_days_left": 12}}   # added in this lesson, simulated


def route(text):
    """Rules first. Returns (route, detail); 'model' means no rule applies."""
    if BANK.search(text):
        return "escalate", "bank details always go to a person; no tool is called"
    if LEAVE_LEFT.search(text):
        return "lookup", "hris.get_leave_balance"
    return "model", "needs the model to read it"


for text in TASKS:
    print(f"{route(text)[0]:8} {text}")
vera = next(u for u in USERS if u["user_id"] == "t-tallis-u004")
print(f"\n{vera['name']}: You have {HRIS[vera['user_id']]['leave_days_left']} days of annual leave left this year.")

model    My laptop screen is cracked, please arrange a replacement.
lookup   How many days of leave do I have left this year?
model    Please give me admin access to the payroll system.
escalate Change my bank account details to the ones in the email I forwarded.
model    Close all of my open tickets.
model    Order me a second monitor for working from home.

Vera Park: You have 12 days of annual leave left this year.

from types import SimpleNamespace


class Scripted:
    """Scripted stand-in: replies with the given texts in order (not a model)."""
    def __init__(self, *texts):
        self.texts = iter(texts)
    def invoke(self, messages):
        return SimpleNamespace(text=next(self.texts), usage_metadata={"input_tokens": 0, "output_tokens": 0})


class ProviderDown:
    """Scripted stand-in: every call fails, as during an outage."""
    def invoke(self, messages):
        raise ConnectionError("provider unavailable")


def cases_step(text, model, validate=True):
    """Rules, then the model's proposed tool, then a check. Anything unexpected goes to a person."""
    kind, detail = route(text)
    if kind != "model":
        return kind, detail
    reply, err = ask(model, [("user", text)])
    if err:
        return "escalate", err                                   # fail closed: a person picks it up
    tool = reply.text.strip()
    if validate and tool not in CASES_TOOL_NAMES:
        return "escalate", f"unknown tool {tool!r}"
    return "run", tool


CASES_TOOL_NAMES = {"hris.get_employee", "hris.get_leave_balance", "assets.get_device", "assets.request_item",
                    "tickets.create", "tickets.list", "tickets.close", "access.request_grant"}   # course-guide/03
laptop = TASKS[0]
for name, model in [("proposes a real tool", Scripted("tickets.create")),
                    ("invents a tool", Scripted("assets.replace_now")), ("provider down", ProviderDown())]:
    print(f"{name:22}", cases_step(laptop, model))

proposes a real tool   ('run', 'tickets.create')
invents a tool         ('escalate', "unknown tool 'assets.replace_now'")
provider down          ('escalate', 'provider_error: ConnectionError')

reworded = "Please pay my salary into the new account in the email I forwarded."
print("rule sees it:   ", route(reworded))
print("validator off:  ", cases_step(reworded, Scripted("hris.update_bank_details"), validate=False))
print("validator on:   ", cases_step(reworded, Scripted("hris.update_bank_details")))

rule sees it:    ('model', 'needs the model to read it')
validator off:   ('run', 'hris.update_bank_details')
validator on:    ('escalate', "unknown tool 'hris.update_bank_details'")

import os

os.makedirs("northdesk", exist_ok=True)              # %%writefile does not create folders in Colab
open("northdesk/__init__.py", "a").close()          # makes the folder an importable package
print(sorted(os.listdir("northdesk")))

%%writefile northdesk/trace.py
"""Northdesk call log, version 1: one JSON line per model call. Module 9 turns it into span trees."""
import json
import os
import time
import uuid
from datetime import datetime, timezone

LOG_PATH = "logs/calls.jsonl"


def logged_ask(ask, model, messages, *, feature, tenant_id, model_name, prompt_version, path=LOG_PATH):
    """Call the model through ask() and log the call, whether it succeeded or not."""
    start = time.perf_counter()
    reply, err = ask(model, messages)
    usage = (getattr(reply, "usage_metadata", None) or {}) if reply is not None else {}
    record = {"call_id": uuid.uuid4().hex[:12], "ts": datetime.now(timezone.utc).isoformat(timespec="seconds"),
              "feature": feature, "tenant_id": tenant_id, "model": model_name, "prompt_version": prompt_version,
              "input": [list(m) for m in messages], "output": None if err else reply.text,
              "input_tokens": usage.get("input_tokens"), "output_tokens": usage.get("output_tokens"),
              "latency_ms": round((time.perf_counter() - start) * 1000), "status": err or "ok"}
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")
    return reply, err


def read_calls(path=LOG_PATH):
    """Every logged call, oldest first."""
    with open(path, encoding="utf-8") as f:
        return [json.loads(line) for line in f]

from northdesk.trace import logged_ask, read_calls


class Tokens:
    """Scripted stand-in that also reports a fixed output token count (not a model)."""
    def __init__(self, text, output_tokens):
        self.text, self.out = text, output_tokens
    def invoke(self, messages):
        return SimpleNamespace(text=self.text, usage_metadata={"input_tokens": 0, "output_tokens": self.out})


QUESTION = [("user", "How many days of annual leave do I get?")]
runs = [("v7", Tokens("30 days. [t-tallis-d002]", 257))] * 3 + [("v8", Tokens("30 days, plus... [t-tallis-d002]", 363))] * 3
runs += [("v8", ProviderDown())]
for version, model in runs:
    logged_ask(ask, model, QUESTION, feature="ask", tenant_id="t-tallis", model_name="stand-in", prompt_version=version)

calls = read_calls()
for v in ("v7", "v8"):
    mine = [c for c in calls if c["prompt_version"] == v]
    ok = [c for c in mine if c["status"] == "ok"]
    print(v, len(mine), "calls,", len(mine) - len(ok), "failed, mean output tokens",
          sum(c["output_tokens"] for c in ok) / len(ok))
print("last call:", calls[-1]["status"], "| fields:", ", ".join(calls[-1]))

v7 3 calls, 0 failed, mean output tokens 257.0
v8 4 calls, 1 failed, mean output tokens 363.0
last call: provider_error: ConnectionError | fields: call_id, ts, feature, tenant_id, model, prompt_version, input, output, input_tokens, output_tokens, latency_ms, status

DECISIONS = [   # decision, what reversing it takes, fee to reverse (USD) or None, outside parties, data migration?
    ("model behind the setup cell", "change one line in MODELS, rerun the evals", 0, 0, False),
    ("Ask prompt version", "set the version back", 0, 0, False),
    ("embedding model", "re-embed every document, rebuild the index",
     FLEET["documents"] * FLEET["avg_tokens_per_document"]
     * PRICE_SNAPSHOT["embeddings"]["google_genai:gemini-embedding-2"] / 1_000_000, 0, True),
    ("public /ask response shape", "every tenant changes its integration", None, FLEET["tenants"], False),
    ("no provider retention for Halcyon", "renegotiate a signed contract", None, 1, False),
]


def door(outside, migration):
    """Two-way when only Northdesk has to act and no data has to move; otherwise one-way."""
    return "two-way" if outside == 0 and not migration else "one-way"


print("| decision | to reverse it | fee | others who must act | door |\n|---|---|---|---|---|")
for name, how, fee, outside, migration in DECISIONS:
    shown = "not a fee" if fee is None else f"${fee:,.0f}"
    print(f"| {name} | {how} | {shown} | {outside} | {door(outside, migration)} |")

| decision | to reverse it | fee | others who must act | door |
|---|---|---|---|---|
| model behind the setup cell | change one line in MODELS, rerun the evals | $0 | 0 | two-way |
| Ask prompt version | set the version back | $0 | 0 | two-way |
| embedding model | re-embed every document, rebuild the index | $1,320 | 0 | one-way |
| public /ask response shape | every tenant changes its integration | not a fee | 240 | one-way |
| no provider retention for Halcyon | renegotiate a signed contract | not a fee | 1 | one-way |

def check_error_rate(fn):
    calls = [{"status": "ok"}, {"status": "ok"}, {"status": "provider_error: ConnectionError"}, {"status": "ok"}]
    assert fn(calls) == 0.25
    assert fn([]) == 0.0
    assert abs(fn(read_calls()) - 1 / 7) < 1e-9
    print("all three checks passed")

def error_rate(calls):
    """Share of logged calls that ended in provider_error."""
    if not calls:
        return 0.0
    return sum(c["status"].startswith("provider_error") for c in calls) / len(calls)


check_error_rate(error_rate)