Lesson 4: Measuring what a model can and cannot do

%pip install -q -U langchain langgraph langchain-groq langchain-google-genai langchain-ollama

import os
from langchain.chat_models import init_chat_model

PROVIDER = "groq"   # "groq" or "gemini" in Colab; "ollama" on your own computer
MODELS = {"groq": "groq:openai/gpt-oss-20b", "gemini": "google_genai:gemini-3.5-flash-lite", "ollama": "ollama:llama3.2:3b"}
KEY = {"groq": "GROQ_API_KEY", "gemini": "GOOGLE_API_KEY", "ollama": None}[PROVIDER]
if KEY and KEY not in os.environ:
    from google.colab import userdata   # keys live in Colab's Secrets panel
    os.environ[KEY] = userdata.get(KEY)
llm = init_chat_model(MODELS[PROVIDER], temperature=0, max_retries=6)

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__}"

from datetime import date

COURSE_DATE = date(2026, 10, 9)    # facts, model names and prices checked on this day
DATA_AS_OF = date(2026, 6, 8)      # the shared data is a snapshot from this morning

def days_since(iso_day, today=DATA_AS_OF):
    """Calendar days from an ISO date string to `today` (the refund rule counts calendar days)."""
    return (today - date.fromisoformat(iso_day)).days

print(DATA_AS_OF.strftime("%A %d %B %Y"))
print(days_since("2026-05-19"), days_since("2026-05-30"))

CATEGORIES = ("billing", "cancellation", "account_access", "bug", "how_to", "feature_request")
PRIORITIES = ("low", "normal", "high", "urgent")

CATEGORY_DEFINITIONS = {
    "billing": "Charges, invoices, prices, plan changes, and refund requests that do not end the plan.",
    "cancellation": "The customer wants to end a paid plan, with or without a refund.",
    "account_access": "Signing in, passwords, two-factor, SSO, and deleting a personal account.",
    "bug": "Something in Quillpath does not work as documented, including slowness and outages.",
    "how_to": "How to do something Quillpath already supports, or where to find a document.",
    "feature_request": "Something Quillpath does not do today that the customer would like.",
}

PRIORITY_DEFINITIONS = {
    "urgent": "A whole team cannot work right now.",
    "high": "One person cannot work, money was taken wrongly, or data is being lost.",
    "normal": "Everything else that needs an answer within a working day.",
    "low": "Ideas and questions with no time pressure, stated or implied.",
}

assert set(CATEGORY_DEFINITIONS) == set(CATEGORIES) and set(PRIORITY_DEFINITIONS) == set(PRIORITIES)
print(len(CATEGORIES), "categories,", len(PRIORITIES), "priorities")

# (id, language, plan, subject, body, category, priority, gold article or None)
DEV = [
    ("T-1001", "en", "team", "Charged twice this month", "Hi, my card was charged 288 USD twice on 1 June for the Team plan (invoice INV-2026-004512). Please refund the duplicate. Ana Duarte, owner, Ardent Design", "billing", "high", "billing-refunds"),
    ("T-1002", "en", "business", "Switch to annual billing?", "We are on Business monthly with 8 seats. If we move to annual, what do we pay and when does it start?", "billing", "normal", "billing-plans"),
    ("T-1003", "en", "team", "VAT number on invoices", "Our accountant needs our VAT number on every invoice. Where do I add it, and can you reissue May's invoice?", "billing", "normal", "billing-invoices"),
    ("T-1004", "en", "team", "Refund for annual plan", "I bought the annual Team plan 20 days ago and we have decided not to use Quillpath. Can I get my money back? Yuki Sato, Studio Nami", "cancellation", "normal", "billing-refunds"),
    ("T-1005", "en", "business", "Cancel our subscription", "Please cancel our Business plan at the end of this month. We are moving to another tool.", "cancellation", "normal", "billing-refunds"),
    ("T-1006", "en", "free", "Forgot my password", "The reset email never arrived and I checked spam. My login is kofi.osei@example.com, phone +44 7700 900321.", "account_access", "high", "account-login"),
    ("T-1007", "en", "business", "SSO login loop", "Since this morning everyone in our team is sent back to the login page after signing in through our identity provider. Nobody can get in.", "account_access", "urgent", "account-sso"),
    ("T-1008", "en", "team", "Locked out", "I typed my password wrong a few times and now it says my account is locked. I have a client demo in an hour.", "account_access", "high", "account-login"),
    ("T-1009", "en", "team", "Add a teammate", "How do I invite a new person to our workspace, and will it change our bill?", "how_to", "low", "billing-plans"),
    ("T-1010", "en", "team", "Automations not firing", "Our rule 'move card to Done when checklist complete' stopped running yesterday. We use a lot of automations. Nothing changed on our side.", "bug", "normal", "boards-automations"),
    ("T-1011", "en", "business", "Export all our boards", "We need a full export of every board with comments for an audit. What formats do you support?", "how_to", "normal", "exports-data"),
    ("T-1012", "en", "business", "Slack posts duplicated", "Every card update now posts twice in our Slack channel.", "bug", "normal", "integrations-slack"),
    ("T-1013", "en", "team", "App crashes on start", "Since Friday's update the Android app closes as soon as I open it. Android 16. I already reinstalled and cleared the cache.", "bug", "normal", None),
    ("T-1014", "en", "free", "Dark mode please", "Would love a dark mode in the web app. My eyes would thank you.", "feature_request", "low", "feature-requests"),
    ("T-1015", "en", "business", "Is Quillpath down?", "Boards are loading very slowly for the whole team and most pages time out.", "bug", "urgent", "status-incidents"),
    ("T-1016", "en", "enterprise", "Data processing agreement", "Our legal team needs your DPA and the list of subprocessors before we renew.", "how_to", "normal", "data-privacy"),
    ("T-1017", "de", "team", "Doppelte Abbuchung", "Hallo, am 1. Juni wurden uns 60 USD zweimal abgebucht (Rechnung INV-2026-004530). Bitte erstatten Sie die doppelte Zahlung. Jonas Feldmann", "billing", "high", "billing-refunds"),
    ("T-1018", "es", "business", "No puedo entrar", "Desde ayer no puedo iniciar sesión. Dice que mi contraseña es incorrecta, pero la acabo de cambiar.", "account_access", "high", "account-login"),
    ("T-1019", "en", "team", "Delete my account", "Please delete my personal account and my data. I am leaving the company.", "account_access", "normal", "data-privacy"),
    ("T-1020", "en", "team", "Gantt view?", "Is there a timeline or Gantt view? I could not find one.", "feature_request", "low", "feature-requests"),
    ("T-1021", "en", "business", "Recurring cards", "Can I make a card that comes back every Monday?", "how_to", "low", "boards-automations"),
    ("T-1022", "en", "team", "Charged after cancelling", "I cancelled on 2 June but was charged again on 5 June. Why?", "billing", "high", None),
    ("T-1023", "en", "enterprise", "Contract question", "Our procurement team wants to change the liability clause in our Enterprise agreement. Who should they talk to?", "how_to", "normal", None),
    ("T-1024", "ja", "team", "請求書のダウンロード", "請求書をPDFでダウンロードするにはどうすればいいですか？", "how_to", "low", "billing-invoices"),
]
print(len(DEV), "dev tickets")

import re, time
PACE = 4   # seconds between calls: Gemini's free tier allows about 15 requests a minute

SYSTEM = ("You triage support tickets for Quillpath, a project-management app.\nCategories:\n"
          + "\n".join(f"- {c}: {d}" for c, d in CATEGORY_DEFINITIONS.items())
          + "\nReply with the category, then a space, then how sure you are that it is right, as a number from 0 to 100.")

def triage(row):
    reply, err = ask(llm, [("system", SYSTEM), ("human", f"Subject: {row[3]}\n\n{row[4]}")])
    if reply is None:
        print(row[0], "no answer:", err)
        return "none", None
    reply = reply.text.lower()
    category = next((c for c in CATEGORIES if c in reply), "none")
    sure = re.search(r"\d+", reply)
    return category, int(sure.group()) if sure else None

hosted = {}
for row in DEV:
    hosted[row[0]] = triage(row)
    time.sleep(PACE)
gold = {row[0]: row[5] for row in DEV}
right = [tid for tid in gold if hosted[tid][0] == gold[tid]]
print(f"{len(right)} of {len(gold)} right")
print("T-1004:", hosted["T-1004"], "| gold:", gold["T-1004"])
print("wrong:", {tid: hosted[tid] for tid in gold if tid not in right})

from math import sqrt

def wilson(k, n, z=1.96):
    """95% interval for a success rate of k out of n (the Wilson score interval)."""
    p = k / n
    centre = (p + z * z / (2 * n)) / (1 + z * z / n)
    half = z * sqrt(p * (1 - p) / n + z * z / (4 * n * n)) / (1 + z * z / n)
    return centre - half, centre + half

k, n = len(right), len(gold)
low, high = wilson(k, n)
print(f"hosted model: {k / n:.0%}, 95% interval {low:.0%} to {high:.0%}")
most_common = max(CATEGORIES, key=list(gold.values()).count)
k0 = list(gold.values()).count(most_common)
print(f"always {most_common}: {k0 / n:.0%}, 95% interval {wilson(k0, n)[0]:.0%} to {wilson(k0, n)[1]:.0%}")
for size in (24, 100, 400):
    lo, hi = wilson(round(0.8 * size), size)
    print(f"80% right on {size:3} tickets: interval {hi - lo:.0%} wide")

LISTING = "\n".join(f"{t[0]} | {t[2]} | {t[3]}" for t in DEV)
TASKS = [  # (kind, question, the answer code computes)
    ("count", f"Tickets (id | plan | subject):\n{LISTING}\n\nHow many tickets are from the team plan? Reply with a number only.", str(sum(t[2] == "team" for t in DEV))),
    ("count", f"Tickets (id | plan | subject):\n{LISTING}\n\nHow many subjects end with a question mark? Reply with a number only.", str(sum(t[3].endswith("?") for t in DEV))),
    ("count", f"Ticket text: {DEV[0][4]}\n\nHow many words are in this text? Count words separated by spaces. Reply with a number only.", str(len(DEV[0][4].split()))),
    ("date", "How many days are there from 2026-05-19 to 2026-06-08? Reply with a number only.", str(days_since("2026-05-19"))),
    ("date", "How many days are there from 2026-02-17 to 2026-06-08? Reply with a number only.", str(days_since("2026-02-17"))),
    ("date", "What day of the week is 2026-05-19? Reply with the day only.", date(2026, 5, 19).strftime("%A")),
    ("extract", f"Ticket text: {DEV[0][4]}\n\nWhat is the invoice number? Reply with it only.", "INV-2026-004512"),
    ("extract", f"Ticket text: {DEV[16][4]}\n\nWhat amount was charged twice, in USD? Reply with a number only.", "60"),
    ("extract", f"Ticket text: {DEV[3][4]}\n\nWho sent this ticket? Reply with the name only.", "Yuki Sato"),
]
for kind, question, answer in TASKS:
    reply, err = ask(llm, question)
    if reply is None:
        print(f"{kind:8} code: {answer:16} no answer ({err})")
    else:
        reply = reply.text.strip().rstrip(".")
        print(f"{kind:8} code: {answer:16} model: {reply[:20]:20} {'ok' if reply.lower() == answer.lower() else 'WRONG'}")
    time.sleep(PACE)

for c in CATEGORIES:
    ids = [tid for tid in gold if gold[tid] == c]
    k = sum(hosted[tid][0] == c for tid in ids)
    lo, hi = wilson(k, len(ids))
    print(f"{c:16} {k} of {len(ids)}   interval {lo:.0%} to {hi:.0%}")

LONG = "\n\n".join(f"{t[0]}\nSubject: {t[3]}\n{t[4]}" for t in DEV)
FORMAT_A = ("Categories:\n" + "\n".join(f"- {c}: {d}" for c, d in CATEGORY_DEFINITIONS.items())
            + "\nLabel every ticket below. Reply with one line per ticket, in the form T-1001: category.")

def triage_all(instructions):
    system = "You triage support tickets for Quillpath, a project-management app.\n" + instructions
    reply, err = ask(llm, [("system", system), ("human", LONG)])
    if reply is None:
        print("no answer:", err)
    found = dict(re.findall(r"(T-\d{4})\W+([A-Za-z_]+)", reply.text if reply else ""))
    return {tid: found.get(tid, "missing").lower() for tid in gold}

batch = triage_all(FORMAT_A)
print(len(LONG.split()), "words in one request")
print(sum(batch[t] == gold[t] for t in gold), "of 24 right; missing:", sum(v == "missing" for v in batch.values()))
print("changed from one at a time:", {t: (hosted[t][0], batch[t]) for t in gold if batch[t] != hosted[t][0]})

%pip install -q transformers==5.19.0 torch

import torch
from transformers import AutoTokenizer, AutoModelForCausalLM

SMALL, SMALL_REV = "HuggingFaceTB/SmolLM2-135M-Instruct", "12fd25f77366fa6b3b4b768ec3050bf629380bac"   # pinned revision (course-guide/02)
tok = AutoTokenizer.from_pretrained(SMALL, revision=SMALL_REV)
small = AutoModelForCausalLM.from_pretrained(SMALL, revision=SMALL_REV).eval()
LABEL_IDS = [tok(c, add_special_tokens=False).input_ids for c in CATEGORIES]

def label_probs(system, row):
    """Probability of each category name as the reply, renormalised over the six."""
    chat = [{"role": "system", "content": system}, {"role": "user", "content": f"Subject: {row[3]}\n\n{row[4]}"}]
    p = tok(tok.apply_chat_template(chat, tokenize=False, add_generation_prompt=True), add_special_tokens=False).input_ids
    longest = max(map(len, LABEL_IDS))
    batch_ids = torch.tensor([p + l + [0] * (longest - len(l)) for l in LABEL_IDS])   # pad after each label
    with torch.no_grad():
        logp = torch.log_softmax(small(batch_ids).logits, dim=-1)
    scores = torch.stack([sum(logp[i, len(p) - 1 + j, t] for j, t in enumerate(l)) for i, l in enumerate(LABEL_IDS)])
    return dict(zip(CATEGORIES, torch.softmax(scores, 0).tolist()))

SMALL_A = SYSTEM.split("\nReply")[0] + "\nReply with the category only."
small_probs = {row[0]: label_probs(SMALL_A, row) for row in DEV}          # about a minute on a CPU
small_pick = {t: max(p, key=p.get) for t, p in small_probs.items()}
ks = sum(small_pick[t] == gold[t] for t in gold)
print(f"SmolLM2-135M: {ks} of 24 = {ks / 24:.0%}, interval {wilson(ks, 24)[0]:.0%} to {wilson(ks, 24)[1]:.0%}")
print("T-1004:", small_pick["T-1004"], round(small_probs["T-1004"][small_pick["T-1004"]], 2))
only_hosted = [t for t in gold if hosted[t][0] == gold[t] != small_pick[t]]
only_small = [t for t in gold if small_pick[t] == gold[t] != hosted[t][0]]
print("hosted right, small wrong:", len(only_hosted), "| small right, hosted wrong:", len(only_small), only_small)

FORMAT_B = ("CATEGORIES\n" + "\n".join(f"{c.upper()} = {d}" for c, d in CATEGORY_DEFINITIONS.items())
            + "\nLABEL EVERY TICKET BELOW. ONE LINE PER TICKET, IN THE FORM T-1001 -> CATEGORY.")
FORMAT_C = ("Categories:\n" + "\n".join(f"{i}. {c} ({d})" for i, (c, d) in enumerate(reversed(CATEGORY_DEFINITIONS.items()), 1))
            + "\nLabel every ticket below. Reply with one line per ticket, in the form T-1001: category.")
runs = {"A, run 1": batch}
for name, fmt in [("A, run 2", FORMAT_A), ("A, run 3", FORMAT_A), ("B", FORMAT_B), ("C", FORMAT_C)]:
    time.sleep(PACE)
    runs[name] = triage_all(fmt)
for name, labels in runs.items():
    flips = [t for t in gold if labels[t] != batch[t]]
    print(f"{name:9} {sum(labels[t] == gold[t] for t in gold)} of 24 right, {len(flips)} changed from run 1 {flips}")

SMALL_B = ("You triage support tickets for Quillpath, a project-management app.\nCATEGORIES\n"
           + "\n".join(f"{c} = {d}" for c, d in CATEGORY_DEFINITIONS.items()) + "\nREPLY WITH THE CATEGORY ONLY.")
SMALL_C = ("You triage support tickets for Quillpath, a project-management app.\nCategories:\n"
           + "\n".join(f"{i}. {c} ({d})" for i, (c, d) in enumerate(reversed(CATEGORY_DEFINITIONS.items()), 1))
           + "\nReply with the category only.")
def top_label(system, row):
    probs = label_probs(system, row)
    return max(probs, key=probs.get)

for name, system in [("B", SMALL_B), ("C", SMALL_C)]:
    pick = {row[0]: top_label(system, row) for row in DEV}   # about a minute each
    print(f"format {name}: {sum(pick[t] == gold[t] for t in gold)} of 24 right, "
          f"{sum(pick[t] != small_pick[t] for t in gold)} changed, {list(pick.values()).count('billing')} called billing")

def calibration(rows, cuts):
    """rows: (confidence, right?) pairs. Prints the accuracy inside each confidence band."""
    for lo, hi in zip(cuts, cuts[1:]):
        band = [ok for conf, ok in rows if lo <= conf < hi]
        if band:
            print(f"   said {lo:g} to {hi:g}: {len(band):2} answers, {sum(band) / len(band):.0%} right")

print("hosted model, stated confidence 0 to 100:")
calibration([(s or 0, c == gold[t]) for t, (c, s) in hosted.items()], [0, 90, 100, 101])
print("small model, probability of its top label:")
calibration([(max(p.values()), small_pick[t] == gold[t]) for t, p in small_probs.items()], [0, 0.5, 0.7, 1.01])
print("said 100 and wrong:", [t for t, (c, s) in hosted.items() if s == 100 and c != gold[t]])

PROBES = {
    "today": "What is today's date? Reply with the date only, as YYYY-MM-DD.",
    "newest Gemini": "What is the newest Gemini model you know of? Reply with its name only.",
    "refund, no date": "Quillpath refunds annual plans cancelled within 14 days of renewal. A customer's annual plan renewed on 2026-05-30 and they cancel today. Is a refund due? Reply yes or no, then the date you took as today.",
}
PROBES["refund, date given"] = f"Today is {DATA_AS_OF}. " + PROBES["refund, no date"]
for name, question in PROBES.items():
    reply, err = ask(llm, question)
    print(f"{name:19} {reply.text.strip()[:60] if reply else f'no answer ({err})'}")
    time.sleep(PACE)
print("rule in code:", "yes" if days_since("2026-05-30") <= 14 else "no", f"({days_since('2026-05-30')} days to {DATA_AS_OF})")

DOCUMENTED = {"groq": "June 2024", "gemini": "March 2026", "ollama": "December 2023"}   # model cards, checked October 2026
said, err1 = ask(llm, "What is the knowledge cutoff of your training data? Reply with the month and year only.")
time.sleep(PACE)
guess, err2 = ask(llm, "You will label 24 customer support tickets for a project-management app into six categories "
                   "(billing, cancellation, account_access, bug, how_to, feature_request), one category each. "
                   "What percentage do you expect to get right? Reply with a number only.")
print("says its cut-off is:", said.text.strip() if said else f"no answer ({err1})", "| model card:", DOCUMENTED[PROVIDER])
print("predicted accuracy:", guess.text.strip() if guess else f"no answer ({err2})", "| measured:", f"{len(right) / len(gold):.0%}")

def tickets_needed(p, width):
    """Smallest set size whose 95% Wilson interval at accuracy p is at most `width` wide."""
    n = 10
    while True:
        lo, hi = wilson(round(p * n), n)
        if hi - lo <= width:
            return n
        n += 10

n_needed = tickets_needed(0.80, 0.10)
lo, hi = wilson(round(0.80 * n_needed), n_needed)
assert hi - lo <= 0.10, "the interval is still wider than 10 points"
print(f"{n_needed} tickets: interval {lo:.0%} to {hi:.0%}; that is {n_needed / 24:.0f} times the dev set")