diff --git a/CHANGELOG.md b/CHANGELOG.md index db8d971..8090123 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -109,6 +109,19 @@ different event from one that moved because it was wrong. ### Fixed +- The local arm, run for real for the first time against a pinned + Qwen2.5-1.5B-Instruct on a GPU, turned up four problems the fakes could not + (#3). The run's eight workers each loaded their own copy of the checkpoint on + the first case; it now loads once, and forward passes run one at a time, with + latency timed after any wait. There was no way to put it on a GPU from the + command line; `--device` now does, and the artifact records it. The first + pass on a device paid for kernel setup, which put a 10 second call in a + latency tail whose p50 was 66 ms; the readout now makes that pass when it + loads, and p99 fell to 487 ms. The report called every refused or failed case + a cache hit in its latency line, on runs with no cache at all, and blamed a + missing confidence on yes/no answers for any arm that reports none; both now + say what happened. + - The site printed some figures one step off from the report, such as a measured ECE of 0.00125 as `0.0012` where the report prints `0.0013` (#48). Its formatter meant to round half to even only on an exact tie, but its tie diff --git a/README.md b/README.md index c824694..0b1796d 100644 --- a/README.md +++ b/README.md @@ -191,6 +191,21 @@ uv sync --extra local Without it every case fails with a message telling you this, so if a local run reports no figures at all, that is the first thing to check. +Name the checkpoint, pin it to a commit, and say where it runs: + +``` +uv run plumbline run datasets/public/jevbench-hard.jsonl --format jevbench \ + --adapter local_logits --model Qwen/Qwen2.5-1.5B-Instruct \ + --revision 989aa7980e4cf806f80c7fef2b1adb7bc71aa306 --device cuda +``` + +`--device` defaults to `cpu`, which works anywhere and is slow. On Windows the +torch that PyPI serves is CPU only; for `--device cuda`, install a CUDA build +of torch from the PyTorch index into the same environment. The arm scores only +options that are a single token for the checkpoint and refuses the rest by +name, so on the public fixture, whose choice options are mostly multi-word +identifiers, most choice rows are refused and the yes/no rows carry the run. + ### Running a hosted vendor This one spends money. Set a key, name an adapter, and cap the run. @@ -298,7 +313,7 @@ and self-hosted servers speak it, which is why one adapter covers all of them. | Adapter | Transport | Semantics | Run for real | Adding one | |---|---|---|---|---| | `typesafe_wire` | Jev wire format over HTTP | `calibrated_claim` | Yes, 40 rows against a hosted vendor | `--base-url`. Anything serving the same wire format is one flag, recorded in the artifact, including self-hosted endpoints and open models behind a compatible server. | -| `local_logits` | Option-token logits from a local checkpoint | `restricted_softmax` | **No, tests only** | A HuggingFace model id and a pinned revision. Needs the optional `local` extra. | +| `local_logits` | Option-token logits from a local checkpoint | `restricted_softmax` | Yes, 105 rows against a pinned Qwen2.5-1.5B-Instruct on a GPU; 39 scored, 66 refused as multi-token | A HuggingFace model id and a pinned revision. Needs the optional `local` extra. | | `generative` | Chat completion, parsed | `none` | **No, tests only** | A model string. | | `mock` | None, seeded | configurable | Yes, it is the example report | Built in. A deterministic stand-in, not a system under test. | @@ -350,9 +365,9 @@ Specific, and none of them are going to surprise you later. the bound is and where it bites. - **Verified on Ubuntu, Windows, and macOS, Python 3.12 through 3.14**, and on the oldest release of each dependency that pyproject allows. -- **Two of the three transports have never run outside the test suite.** See - the adapters table above and - [issue #3](https://github.com/TMHSDigital/plumbline/issues/3). +- **The generative transport has never run outside the test suite.** The + local arm has, against a pinned open checkpoint; see the adapters table + above and [issue #3](https://github.com/TMHSDigital/plumbline/issues/3). ## Related work diff --git a/src/plumbline/adapters/local_logits.py b/src/plumbline/adapters/local_logits.py index d5351ee..d877286 100644 --- a/src/plumbline/adapters/local_logits.py +++ b/src/plumbline/adapters/local_logits.py @@ -31,6 +31,7 @@ import math import re +import threading import time from collections.abc import Mapping, Sequence from typing import Protocol @@ -145,16 +146,24 @@ def __init__( self.allow_unpinned_revision = allow_unpinned_revision self.device = device self._readout = readout + # The run's workers reach the first case together. Without the load + # lock each would load its own copy of the checkpoint, gigabytes apiece + # onto one device. The forward lock runs one pass at a time: passes on + # one device contend rather than overlap, and the latency of a pass + # that waited on another would be the wait, not the model. + self._load_lock = threading.Lock() + self._forward_lock = threading.Lock() @property def readout(self) -> LogitReadout: - """The loaded checkpoint, built on first use so import stays cheap.""" - if self._readout is None: - self._readout = TransformersReadout( - model_id=self.model_requested, - revision=self.pinned_revision, - device=self.device, - ) + """The loaded checkpoint, built once on first use so import stays cheap.""" + with self._load_lock: + if self._readout is None: + self._readout = TransformersReadout( + model_id=self.model_requested, + revision=self.pinned_revision, + device=self.device, + ) return self._readout @property @@ -199,9 +208,10 @@ def classify( instructions=self.instructions, text=text, options=", ".join(labels) ) - started = time.perf_counter() - logits = list(readout.option_logits(prompt, token_ids)) - latency_ms = (time.perf_counter() - started) * 1000.0 + with self._forward_lock: + started = time.perf_counter() + logits = list(readout.option_logits(prompt, token_ids)) + latency_ms = (time.perf_counter() - started) * 1000.0 if len(logits) != len(labels): raise PlumblineError( @@ -326,6 +336,11 @@ def __init__(self, *, model_id: str, revision: str, device: str = "cpu") -> None self._model.to(device) self._model.eval() self._device = device + # The first pass on a device initializes its kernels, which took about a + # second on a GPU against a tenth of that for every pass after. That is + # setup, not the model, so it happens here rather than inside the first + # case's latency. + self.option_logits("warm up", [0]) # transformers records the commit it resolved to. When it does not, the # requested revision is the only thing known, and the adapter's own # check is what catches a mismatch it can see. diff --git a/src/plumbline/cli.py b/src/plumbline/cli.py index 9e7eb1e..467ad25 100644 --- a/src/plumbline/cli.py +++ b/src/plumbline/cli.py @@ -90,6 +90,14 @@ def run( "--timeout", help="Seconds one request may take, for an adapter that takes it." ), ] = None, + device: Annotated[ + str | None, + typer.Option( + "--device", + help="Where a local checkpoint runs, such as cpu or cuda, for an adapter that " + "takes it. Recorded in the artifact.", + ), + ] = None, semantics: Annotated[ str | None, typer.Option( @@ -150,6 +158,7 @@ def run( revision=revision, base_url=base_url, timeout=timeout, + device=device, semantics=semantics, seed=seed, accuracy=accuracy, @@ -307,6 +316,7 @@ def _build( revision: str | None, base_url: str | None = None, timeout: float | None = None, + device: str | None = None, semantics: str | None, seed: int, accuracy: float, @@ -321,6 +331,8 @@ def _build( config["base_url"] = base_url if timeout is not None: config["timeout"] = timeout + if device is not None: + config["device"] = device if semantics is not None: config["probability_semantics"] = _semantics(semantics) @@ -361,6 +373,7 @@ def _build( "revision": "--revision", "base_url": "--base-url", "timeout": "--timeout", + "device": "--device", "probability_semantics": "--semantics", } @@ -371,6 +384,7 @@ def _plan_text( """What a dry run prints: what would be sent, to whom, and what it would cost.""" endpoint = getattr(adapter, "base_url", None) timeout = getattr(adapter, "timeout", None) + device = getattr(adapter, "device", None) semantics = adapter.probability_semantics + (", set by --semantics" if semantics_set else "") if planned.estimated_cost_usd is not None and planned.pricing is not None: cost = ( @@ -398,6 +412,7 @@ def _plan_text( f"- Timeout: {timeout:g} s per request." if timeout else "- Timeout: the adapter's default.", + *([f"- Device: {device}."] if device else []), f"- Probability semantics: {semantics}.", f"- Cost: {cost}", f"- Guard: {cost_limit}, {case_limit}; the run would start.", diff --git a/src/plumbline/report/markdown.py b/src/plumbline/report/markdown.py index 80e6ba6..0a4032e 100644 --- a/src/plumbline/report/markdown.py +++ b/src/plumbline/report/markdown.py @@ -661,10 +661,17 @@ def _confidence_lines( ) -> list[str]: values = tuple(record.prediction.confidence for record in successes if record.prediction) if all(value is None for value in values): + # A yes/no asked as one has a single probability and nothing to + # summarize beside it. Any other arm without one simply returns none, + # such as a local checkpoint, which has no vendor to report it. + why = ( + "a yes/no answer has no distribution to summarize" + if successes and all(record.asked_as == "noul" for record in successes) + else "the adapter returns none beside its probabilities" + ) return [ - "- **Confidence**: not reported. This arm reports no confidence statistic: a " - "yes/no answer has no distribution to summarize, so the number does not exist " - "rather than being missing." + f"- **Confidence**: not reported. This arm reports no confidence statistic: {why}, " + "so the number does not exist rather than being missing." ] if any(value is None for value in values): missing = sum(1 for value in values if value is None) @@ -731,7 +738,11 @@ def _latency_lines(result: RunResult) -> list[str]: "- **Latency**: not reported. No call went out, so every latency here would " "be a measurement of disk." ] - summary = latency.summarize(live, excluded_cache_hits=len(result.records) - len(live)) + # Only answers served from cache are cache hits. A refused or failed case + # made no timed call either, and counting it here named every local arm's + # untokenizable cases as cache hits on a run with no cache at all. + hits = sum(1 for record in result.records if record.ok and record.from_cache) + summary = latency.summarize(live, excluded_cache_hits=hits) return [f"- **Latency**: {summary}"] diff --git a/src/plumbline/runner/execute.py b/src/plumbline/runner/execute.py index 9d2d147..e7cb1be 100644 --- a/src/plumbline/runner/execute.py +++ b/src/plumbline/runner/execute.py @@ -631,6 +631,9 @@ def handle(index: int) -> None: # A tight timeout turns slow answers into failures, which is part of # what the run measured. "timeout_seconds": getattr(adapter, "timeout", None), + # Where a local checkpoint ran. It does not change the question, but a + # latency figure means nothing without it. + "device": getattr(adapter, "device", None), # Whether rows carried option descriptions and whether they were sent, # so the report can say when the dataset's descriptions went nowhere. "label_descriptions": { diff --git a/tests/test_cli_run_options.py b/tests/test_cli_run_options.py index 2ed695f..b3121bd 100644 --- a/tests/test_cli_run_options.py +++ b/tests/test_cli_run_options.py @@ -6,6 +6,7 @@ from __future__ import annotations +import dataclasses import json from pathlib import Path @@ -228,3 +229,56 @@ class Patient(MockAdapter): result = execute.run(Patient(gold_by_text={"t": "a"}), cases) assert result.config["timeout_seconds"] == 42.0 + + +def test_a_device_goes_to_the_local_arm_and_is_refused_elsewhere(tmp_path: Path) -> None: + dataset = a_dataset(tmp_path / "d.jsonl") + + refused = invoke("run", str(dataset), "--device", "cuda", "--dry-run") + planned = invoke( + "run", + str(dataset), + "--adapter", + "local_logits", + "--model", + "some/checkpoint", + "--revision", + "c" * 40, + "--device", + "cuda", + "--dry-run", + ) + + assert refused.exit_code == 1 and "does not take --device" in refused.stderr + assert planned.exit_code == 0, planned.stderr + assert "Device: cuda" in planned.stdout + + +class RefusesSome(MockAdapter): + """Refuses every case whose text ends in an odd digit, as a local arm refuses + options it cannot tokenize, and reports no confidence.""" + + def classify(self, text: str, labels: list[str], **kwargs: object): # type: ignore[no-untyped-def] + from plumbline.types import CaseRefusedError + + if int(text[-1]) % 2: + raise CaseRefusedError("cannot ask this case cleanly") + prediction = super().classify(text, labels, **kwargs) # type: ignore[arg-type] + return dataclasses.replace(prediction, confidence=None) + + +def test_refused_cases_are_not_called_cache_hits_and_no_confidence_is_not_blamed_on_yes_no() -> ( + None +): + from plumbline.report import markdown + + cases = [Case(id=f"c{i}", text=f"t{i}", labels=("a", "b"), gold_label="a") for i in range(10)] + result = execute.run(RefusesSome(gold_by_text={case.text: "a" for case in cases}), cases) + + document = markdown.render([result], options=markdown.ReportOptions(n_boot=50)) + latency_line = next(line for line in document.splitlines() if "**Latency**" in line) + confidence_line = next(line for line in document.splitlines() if "**Confidence**" in line) + + assert "over 5 live calls" in latency_line + assert "cache hits" not in latency_line + assert "yes/no" not in confidence_line diff --git a/tests/test_local_logits.py b/tests/test_local_logits.py index 4856302..eb7b9a8 100644 --- a/tests/test_local_logits.py +++ b/tests/test_local_logits.py @@ -253,3 +253,58 @@ def test_the_artifact_marks_the_arm_as_a_restricted_softmax(tmp_path: Path) -> N assert stored["probability_semantics"] == "restricted_softmax" assert stored["revision"] == PINNED assert {record["cost_basis"] for record in stored["records"]} == {"adapter_reports_no_tokens"} + + +# Running for real + + +def test_concurrent_first_cases_load_the_checkpoint_once(monkeypatch: pytest.MonkeyPatch) -> None: + """The run's workers all reach the first case together. Each loading its own + copy of the checkpoint is several gigabytes per worker, onto one GPU.""" + import threading + import time + + from plumbline.adapters import local_logits + + loads: list[int] = [] + + class SlowToLoad(FakeReadout): + def __init__(self, **_kwargs: object) -> None: + loads.append(1) + time.sleep(0.05) # long enough for every worker to arrive meanwhile + super().__init__() + + monkeypatch.setattr(local_logits, "TransformersReadout", SlowToLoad) + adapter = LocalLogitsAdapter(model_requested="tiny-model", revision=PINNED) + start = threading.Barrier(8) + + def first_case() -> None: + start.wait() + adapter.classify("text", LABELS) + + workers = [threading.Thread(target=first_case) for _ in range(8)] + for worker in workers: + worker.start() + for worker in workers: + worker.join() + + assert len(loads) == 1 + + +def test_the_device_reaches_the_readout_and_the_artifact(monkeypatch: pytest.MonkeyPatch) -> None: + from plumbline.adapters import local_logits + + seen: dict[str, object] = {} + + class Recorded(FakeReadout): + def __init__(self, **kwargs: object) -> None: + seen.update(kwargs) + super().__init__() + + monkeypatch.setattr(local_logits, "TransformersReadout", Recorded) + adapter = registry.create("local_logits", model_requested="m", revision=PINNED, device="cuda") + + result = execute.run(adapter, make_cases(3, labels=LABELS), workers=1) + + assert seen["device"] == "cuda" + assert result.config["device"] == "cuda"