মেট্রিক ও এক্সপেরিমেন্ট রিপোর্ট: accuracy, precision, recall, ল্যাটেন্সি ও CSV

# file: prompts/v1.txt
Classify the support message as billing, delivery or other.
### MESSAGE
{text}

# file: prompts/v2.txt
Classify the support message as billing, delivery or other.
A refund request is billing. Reply with the label only.
### MESSAGE
{text}

# file: toys.py
"""toys.py: stand-in providers (NOT real models) so experiments can be repeated exactly."""
import random

BILLING = ["invoice", "charged", "payment"]
DELIVERY = ["parcel", "package", "delivery", "arrived", "tracking", "driver"]

def make_triage(seed=7, noise=0.3):
    """Toy classifier. It reads the prompt: a 'refund' hint adds a billing word, 'label only' stops wordy
    replies on long messages. With temperature above 0 it sometimes answers 'other' (seeded)."""
    rng = random.Random(seed)
    def provider(messages, temperature=0, **settings):
        head, text = messages[-1]["content"].lower().split("### message", 1)
        billing = BILLING + (["refund"] if "refund" in head else [])
        label = ("billing" if any(w in text for w in billing)
                 else "delivery" if any(w in text for w in DELIVERY) else "other")
        if temperature > 0 and rng.random() < noise:
            label = "other"
        if "label only" not in head and len(text.split()) > 9:
            return f"I think it is {label}."
        return label
    return provider

# file: runner.py
"""runner.py (shortened from the previous page): load cases, run one prompt version, log JSONL."""
import hashlib, json
from pathlib import Path
from promptlab import ProviderError, render

def load_cases(path):
    return json.loads(Path(path).read_text(encoding="utf-8"))

def run_experiment(client, version, cases, log, **settings):
    template = Path("prompts", f"{version}.txt").read_text(encoding="utf-8")
    sha = hashlib.sha256(template.encode()).hexdigest()[:8]
    records = []
    Path(log).parent.mkdir(parents=True, exist_ok=True)
    with open(log, "a", encoding="utf-8") as out:
        for case in cases:
            rec = {"case": case["id"], "version": version, "prompt_sha": sha,
                   "model": client.model, "settings": settings}
            try:
                r = client.complete([{"role": "user", "content": render(template, text=case["text"])}], **settings)
                rec.update(status="ok", text=r.text.strip(), input_tokens=r.usage.input_tokens,
                           output_tokens=r.usage.output_tokens, latency_ms=r.latency_ms)
            except ProviderError as exc:
                rec.update(status="error", error=str(exc))
            out.write(json.dumps(rec, sort_keys=True) + "\n")
            out.flush()
            records.append(rec)
    return records

# file: cases.json
[
  {"id": "c1", "text": "I was charged twice for invoice 3318.", "expected": "billing"},
  {"id": "c2", "text": "My parcel has not arrived after two weeks.", "expected": "delivery"},
  {"id": "c3", "text": "Please send me a refund for the broken lamp that I bought last month.", "expected": "billing"},
  {"id": "c4", "text": "How do I change my profile photo?", "expected": "other"},
  {"id": "c5", "text": "Tracking shows nothing for my package.", "expected": "delivery"},
  {"id": "c6", "text": "The driver left the box outside in the rain, and the invoice total looks wrong too.", "expected": "delivery"},
  {"id": "c7", "text": "Why was my card debited two times?", "expected": "billing"},
  {"id": "c8", "text": "The payment page crashed and I think I was charged.", "expected": "billing"},
  {"id": "c9", "text": "Can you ship to a post office box?", "expected": "delivery"},
  {"id": "c10", "text": "Where can I download the mobile app?", "expected": "other"},
  {"id": "c11", "text": "The package arrived damaged and I want a refund.", "expected": "billing"},
  {"id": "c12", "text": "Thanks, the invoice is correct now and everything is fine.", "expected": "other"}
]

# file: metrics.py
"""metrics.py: small scoring functions (standard library only)."""
import csv
from collections import Counter

def norm(value):
    """Lowercase, trim and collapse spaces, so 'Billing.' and 'billing' match."""
    return " ".join(str(value).lower().split()).strip(".!")

def accuracy(pairs):
    """Share of (predicted, expected) pairs that match exactly after norm()."""
    return sum(norm(p) == norm(g) for p, g in pairs) / len(pairs) if pairs else 0.0

def first_label(text, labels):
    """Lenient reading of a wordy answer: the label that appears earliest in text, else 'none'."""
    found = [(text.lower().find(l), l) for l in labels if l in text.lower()]
    return min(found)[1] if found else "none"

def field_accuracy(preds, golds, fields):
    """Accuracy per field, plus 'all' = share of records where every field is right."""
    n = len(golds)
    result = {f: sum(norm(p.get(f)) == norm(g[f]) for p, g in zip(preds, golds)) / n for f in fields}
    result["all"] = sum(all(norm(p.get(f)) == norm(g[f]) for f in fields) for p, g in zip(preds, golds)) / n
    return result

def confusion(pairs):
    """Counter of (expected, predicted) -> count."""
    return Counter((g, p) for p, g in pairs)

def precision_recall(pairs, labels):
    """Per label: precision = tp/(tp+fp), recall = tp/(tp+fn), f1 = their harmonic mean (0.0 if undefined)."""
    cm, out = confusion(pairs), {}
    for label in labels:
        tp = cm[(label, label)]
        fp = sum(c for (g, p), c in cm.items() if p == label and g != label)
        fn = sum(c for (g, p), c in cm.items() if g == label and p != label)
        pr = tp / (tp + fp) if tp + fp else 0.0
        rc = tp / (tp + fn) if tp + fn else 0.0
        out[label] = {"precision": round(pr, 2), "recall": round(rc, 2),
                      "f1": round(2 * pr * rc / (pr + rc), 2) if pr + rc else 0.0}
    return out

def show_confusion(pairs, labels):
    cm, cols = confusion(pairs), labels + ["none"]
    print(f"{'expected/predicted':<20}" + "".join(f"{c:>10}" for c in cols))
    for g in labels:
        print(f"{g:<20}" + "".join(f"{cm[(g, c)]:>10}" for c in cols))

def write_csv(path, rows):
    with open(path, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=list(rows[0]))
        writer.writeheader()
        writer.writerows(rows)

from promptlab import Client
from runner import load_cases, run_experiment
from metrics import *
from toys import make_triage

LABELS = ["billing", "delivery", "other"]
cases = load_cases("cases.json")
gold = {c["id"]: c["expected"] for c in cases}

def pairs(records, lenient=False):
    """(predicted, expected) for the cases that returned an answer; error records are left out."""
    return [(first_label(r["text"], LABELS) if lenient else r["text"], gold[r["case"]])
            for r in records if r["status"] == "ok"]

client = Client(make_triage(), model="toy-triage")
runs = {v: run_experiment(client, v, cases, f"logs/{v}.jsonl") for v in ("v1", "v2")}
for v, records in runs.items():
    print(v, "strict", round(accuracy(pairs(records)), 3), "lenient", round(accuracy(pairs(records, True)), 3))

v1 strict 0.417 lenient 0.5
v2 strict 0.667 lenient 0.667

golds = [{"vendor": "Nile Tools", "total": "120.50", "currency": "USD"},
         {"vendor": "Aiko Print", "total": "89.00", "currency": "JPY"},
         {"vendor": "Lena GmbH", "total": "1205.00", "currency": "EUR"},
         {"vendor": "Mateo SAS", "total": "40.00", "currency": "COP"},
         {"vendor": "Priya Foods", "total": "15.75", "currency": "INR"}]
preds = [{"vendor": "Nile Tools", "total": "120.50", "currency": "USD"},
         {"vendor": "Aiko Print", "total": "89.00", "currency": None},
         {"vendor": "Lena GMBH", "total": "1,205.00", "currency": "EUR"},
         {"vendor": "Mateo", "total": "40.00", "currency": "COP"},
         {"vendor": "Priya Foods", "total": "15.75", "currency": "INR"}]
print({k: round(v, 2) for k, v in field_accuracy(preds, golds, ["vendor", "total", "currency"]).items()})

{'vendor': 0.8, 'total': 0.8, 'currency': 0.8, 'all': 0.4}

show_confusion(pairs(runs["v2"], True), LABELS)
for label, scores in precision_recall(pairs(runs["v2"], True), LABELS).items():
    print(label, scores)

expected/predicted     billing  delivery     other      none
billing                      4         0         1         0
delivery                     1         2         1         0
other                        1         0         2         0
billing {'precision': 0.67, 'recall': 0.8, 'f1': 0.73}
delivery {'precision': 1.0, 'recall': 0.5, 'f1': 0.67}
other {'precision': 0.5, 'recall': 0.67, 'f1': 0.57}

import csv, json, random

replies = {"t1": {"v1": "Sorry. Refund started.",
                  "v2": "Hello Kwame, I am sorry your order 48213 arrived damaged. We will start your refund today."},
           "t2": {"v1": "Your parcel is late, we know.",
                  "v2": "Hello Lena, I am sorry order 51877 is late. A colleague will check its location and write to you."},
           "t3": {"v1": "Hello Mateo, thank you for writing. We will send the right item.",
                  "v2": "Hello Mateo, sorry about the wrong item in order 60412. We will send the correct one."}}
items = [(case, v, text) for case, d in replies.items() for v, text in d.items()]
random.Random(3).shuffle(items)
key = {f"r{i}": (case, v) for i, (case, v, _) in enumerate(items, 1)}
with open("review_sheet.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.writer(f)
    writer.writerow(["item", "reply", "score_1_to_5", "comment"])
    writer.writerows([f"r{i}", text, "", ""] for i, (_, _, text) in enumerate(items, 1))
json.dump(key, open("review_key.json", "w"))
print(open("review_sheet.csv", encoding="utf-8").read().splitlines()[0])
print(len(key), "items; first two:", list(key.items())[:2])

item,reply,score_1_to_5,comment
6 items; first two: [('r1', ('t1', 'v1')), ('r2', ('t2', 'v1'))]

invented = {("t1", "v1"): 2, ("t1", "v2"): 5, ("t2", "v1"): 1, ("t2", "v2"): 4, ("t3", "v1"): 3, ("t3", "v2"): 5}
by_version = {"v1": [], "v2": []}
for item, (case, v) in key.items():
    by_version[v].append(invented[(case, v)])
print({v: round(sum(s) / len(s), 2) for v, s in by_version.items()}, "n per version:", len(by_version["v1"]))

{'v1': 2.0, 'v2': 4.67} n per version: 3

fake = [r["latency_ms"] for r in runs["v2"]]
print("fake latency, v2: mean", round(sum(fake) / len(fake)), "ms, max", max(fake), "ms")
real = Client(make_triage(), real_timing=True).complete([{"role": "user", "content": "### MESSAGE\nhello"}])
print("real timing is an int >= 0:", isinstance(real.latency_ms, int) and real.latency_ms >= 0)

fake latency, v2: mean 58 ms, max 70 ms
real timing is an int >= 0: True

PRICES = {"input_per_million": 1.00, "output_per_million": 4.00}   # EXAMPLE numbers, not real prices

def cost(records, prices=PRICES):
    ok = [r for r in records if r["status"] == "ok"]
    tokens_in, tokens_out = sum(r["input_tokens"] for r in ok), sum(r["output_tokens"] for r in ok)
    return (tokens_in * prices["input_per_million"] + tokens_out * prices["output_per_million"]) / 1_000_000

for v, records in runs.items():
    per_call = cost(records) / len(records)
    print(v, "per call", f"{per_call:.7f}", "| per 10,000 calls", round(per_call * 10_000, 3), "(example currency units)")

v1 per call 0.0000418 | per 10,000 calls 0.418 (example currency units)
v2 per call 0.0000507 | per 10,000 calls 0.507 (example currency units)

noisy = Client(make_triage(seed=11), model="toy-triage")
hot = [round(accuracy(pairs(run_experiment(noisy, "v2", cases, f"logs/hot{i}.jsonl", temperature=0.7), True)), 2) for i in range(5)]
cold = [round(accuracy(pairs(run_experiment(noisy, "v2", cases, f"logs/cold{i}.jsonl", temperature=0), True)), 2) for i in range(3)]
print("temperature 0.7, five runs:", hot, "min", min(hot), "max", max(hot))
print("temperature 0, three runs :", cold)

temperature 0.7, five runs: [0.58, 0.67, 0.5, 0.58, 0.67] min 0.5 max 0.67
temperature 0, three runs : [0.67, 0.67, 0.67]

def summarise(records):
    ok = [r for r in records if r["status"] == "ok"]
    return {"version": records[0]["version"], "prompt_sha": records[0]["prompt_sha"], "cases": len(records), "ok": len(ok),
            "strict_accuracy": round(accuracy(pairs(records)), 3), "lenient_accuracy": round(accuracy(pairs(records, True)), 3),
            "mean_latency_ms_fake": round(sum(r["latency_ms"] for r in ok) / len(ok)),
            "cost_per_10k_example": round(cost(records) / len(records) * 10_000, 3)}

write_csv("report.csv", [summarise(r) for r in runs.values()])
print(open("report.csv", encoding="utf-8").read().strip())

version,prompt_sha,cases,ok,strict_accuracy,lenient_accuracy,mean_latency_ms_fake,cost_per_10k_example
v1,55f857e1,12,12,0.417,0.5,65,0.418
v2,fbece8b2,12,12,0.667,0.667,58,0.507

for label, s in precision_recall(pairs(runs["v1"], True), LABELS).items():
    print(label, "recall", s["recall"])

billing recall 0.4
delivery recall 0.5
other recall 0.67

golds2 = golds + [{"vendor": "Omar Traders", "total": "75.00", "currency": "EGP"}]
preds2 = preds + [{"vendor": "Omar Traders", "total": "57.00", "currency": "EGP"}]
print({k: round(v, 2) for k, v in field_accuracy(preds2, golds2, ["vendor", "total", "currency"]).items()})

{'vendor': 0.83, 'total': 0.67, 'currency': 0.83, 'all': 0.33}

# %% setup
versions = ["v1", "v2"]
client = Client(make_triage(), model="toy-triage")

# %% run
results = {v: run_experiment(client, v, cases, f"logs/nb_{v}.jsonl") for v in versions}

# %% score
rows = [summarise(results[v]) for v in versions]
for row in rows:
    print(row["version"], row["strict_accuracy"], row["lenient_accuracy"])

# %% verdict and report
gap = round((rows[1]["strict_accuracy"] - rows[0]["strict_accuracy"]) * 100)
one_case = round(100 / len(cases))
print(f"v2 is ahead by {gap} points" if gap >= one_case else "no clear difference")
write_csv("nb_report.csv", rows)

v1 0.417 0.5
v2 0.667 0.667
v2 is ahead by 25 points