lpgcast.bench.run

Run the corpus through a model, or the oracle, and score every episode.

Attributes

ClientFactory = Callable[[BenchCase, LPGCastSettings], LLMClient] module-attribute

PRICES_ENV = 'LPGCAST_BENCH_PRICES' module-attribute

RECAST_CACHE_SUBDIR = 'recast' module-attribute

Classes

BenchReport

Bases: BaseModel

A benchmark run.

Source code in lpgcast/bench/run.py

class BenchReport(BaseModel):
    """A benchmark run."""

    corpus_version: str = CORPUS_VERSION
    corpus_digest: str
    corpus_modified: bool = Field(
        description="The cases differ from the pinned digest: not publishable."
    )
    split: str
    lpgcast_version: str = __version__
    graflo_version: str = Field(default_factory=lambda: version("graflo"))
    started: str
    prices_source: str | None = None
    rows: list[Row]

    @property
    def publishable(self) -> bool:
        return self.split == "test" and not self.corpus_modified

Attributes

corpus_digest instance-attribute

corpus_modified = Field(description='The cases differ from the pinned digest: not publishable.') class-attribute instance-attribute

corpus_version = CORPUS_VERSION class-attribute instance-attribute

graflo_version = Field(default_factory=lambda: version('graflo')) class-attribute instance-attribute

lpgcast_version = __version__ class-attribute instance-attribute

prices_source = None class-attribute instance-attribute

publishable property

rows instance-attribute

split instance-attribute

started instance-attribute

CaseRecord

Bases: BaseModel

One case's episodes and their sum.

Source code in lpgcast/bench/run.py

class CaseRecord(BaseModel):
    """One case's episodes and their sum."""

    name: str
    family: str
    instance: str
    split: str
    digest: str | None = Field(
        default=None,
        description="The case's content digest; absent before corpus 1.1.0.",
    )
    counts: Counts
    episodes: list[EpisodeRecord]

Attributes

counts instance-attribute

digest = Field(default=None, description="The case's content digest; absent before corpus 1.1.0.") class-attribute instance-attribute

episodes instance-attribute

family instance-attribute

instance instance-attribute

name instance-attribute

split instance-attribute

EpisodeRecord

Bases: BaseModel

One episode's outcome: enough to pair two runs episode by episode.

Source code in lpgcast/bench/run.py

class EpisodeRecord(BaseModel):
    """One episode's outcome: enough to pair two runs episode by episode."""

    source_id: str
    status: str
    failure: str | None
    counts: Counts

Attributes

counts instance-attribute

failure instance-attribute

source_id instance-attribute

status instance-attribute

Row

Bases: BaseModel

Every selected case cast by one model.

Source code in lpgcast/bench/run.py

class Row(BaseModel):
    """Every selected case cast by one model."""

    label: str
    model: dict[str, Any] = Field(description="Model settings that reach the provider.")
    usd: float | None = Field(
        default=None,
        description="List price of the tokens used, billed or replayed; None "
        "without a price for this model.",
    )
    cases: list[CaseRecord]

    @property
    def counts(self) -> Counts:
        return total([case.counts for case in self.cases])

Attributes

cases instance-attribute

counts property

label instance-attribute

model = Field(description='Model settings that reach the provider.') class-attribute instance-attribute

usd = Field(default=None, description='List price of the tokens used, billed or replayed; None without a price for this model.') class-attribute instance-attribute

Functions:

case_settings(case, base)

base with the settings case pins.

Source code in lpgcast/bench/run.py

def case_settings(case: BenchCase, base: LPGCastSettings) -> LPGCastSettings:
    """*base* with the settings *case* pins."""
    pinned = case.settings.model_dump(include=case.settings.model_fields_set)
    return base.model_copy(update=pinned)

load_prices(path=None)

The price table and its _source note; empty when none is configured.

Source code in lpgcast/bench/run.py

def load_prices(
    path: str | None = None,
) -> tuple[dict[str, dict[str, float]], str | None]:
    """The price table and its ``_source`` note; empty when none is configured."""
    source = path or os.environ.get(PRICES_ENV)
    if not source:
        return {}, None
    raw = json.loads(Path(source).read_text(encoding="utf-8"))
    note = raw.get("_source")
    return {k: v for k, v in raw.items() if not k.startswith("_")}, note

model_clients(settings)

Clients for the configured model, one cache per settings.

Source code in lpgcast/bench/run.py

def model_clients(settings: LPGCastSettings) -> ClientFactory:
    """Clients for the configured model, one cache per settings."""

    def build(_case: BenchCase, case_settings_: LPGCastSettings) -> LLMClient:
        return LLMClient(case_settings_.llm, LLMCache(case_settings_.cache))

    return build

new_report(split, started, rows, prices_source)

A report over rows, stamped with the corpus it ran against.

Source code in lpgcast/bench/run.py

def new_report(
    split: str, started: str, rows: list[Row], prices_source: str | None
) -> BenchReport:
    """A report over *rows*, stamped with the corpus it ran against."""
    digest = corpus_digest()
    return BenchReport(
        corpus_digest=digest,
        corpus_modified=digest != CORPUS_DIGEST,
        split=split,
        started=started,
        prices_source=prices_source,
        rows=rows,
    )

oracle_clients()

Oracle clients, scripted per case.

Source code in lpgcast/bench/run.py

def oracle_clients() -> ClientFactory:
    """Oracle clients, scripted per case."""

    def build(case: BenchCase, _settings: LPGCastSettings) -> LLMClient:
        return oracle_client(case.expected)

    return build

price(counts, entry)

USD for counts' tokens at entry's rates.

Source code in lpgcast/bench/run.py

def price(counts: Counts, entry: dict[str, float] | None) -> float | None:
    """USD for *counts*' tokens at *entry*'s rates."""
    if not entry:
        return None
    return (
        counts.input_tokens * float(entry.get("input_per_1m", 0.0))
        + counts.output_tokens * float(entry.get("output_per_1m", 0.0))
    ) / 1_000_000

recast_settings(base)

base with its own cache directory, so the second cast makes a fresh call.

Source code in lpgcast/bench/run.py

def recast_settings(base: LPGCastSettings) -> LPGCastSettings:
    """*base* with its own cache directory, so the second cast makes a fresh call."""
    root = base.cache.dir or default_cache_dir()
    cache = base.cache.model_copy(update={"dir": root / RECAST_CACHE_SUBDIR})
    return base.model_copy(update={"cache": cache})

run_case(case, settings, clients, *, concurrency=8) async

Cast and score every episode of case.

Raises:

Type	Description
LLMConfigurationError	the deployment is misconfigured.
CacheMissError	a miss while the cache requires hits.

Source code in lpgcast/bench/run.py

async def run_case(
    case: BenchCase,
    settings: LPGCastSettings,
    clients: ClientFactory,
    *,
    concurrency: int = 8,
) -> CaseRecord:
    """Cast and score every episode of *case*.

    Raises:
        LLMConfigurationError: the deployment is misconfigured.
        CacheMissError: a miss while the cache requires hits.
    """
    pinned = case_settings(case, settings)
    caster = LPGCaster(case.schema_, settings=pinned, client=clients(case, pinned))
    recaster = None
    if case.recast:
        again = recast_settings(pinned)
        recaster = LPGCaster(case.schema_, settings=again, client=clients(case, again))
    gate = asyncio.Semaphore(max(1, concurrency))
    try:
        async with asyncio.TaskGroup() as group:
            tasks = [
                group.create_task(
                    _episode(
                        caster,
                        recaster,
                        episode,
                        case.expected[episode.source_id],
                        case,
                        gate,
                    )
                )
                for episode in case.episodes
            ]
    except ExceptionGroup as errors:
        raise errors.exceptions[0] from errors
    episodes = [task.result() for task in tasks]
    return CaseRecord(
        name=case.name,
        family=case.family,
        instance=case.instance,
        split=case.split,
        digest=case.digest,
        counts=total([e.counts for e in episodes]),
        episodes=episodes,
    )

run_row(cases, settings, *, oracle=False, concurrency=8, prices=None) async

One row: every case cast by the configured model, or by the oracle.

Source code in lpgcast/bench/run.py

async def run_row(
    cases: list[BenchCase],
    settings: LPGCastSettings,
    *,
    oracle: bool = False,
    concurrency: int = 8,
    prices: dict[str, dict[str, float]] | None = None,
) -> Row:
    """One row: every case cast by the configured model, or by the oracle."""
    clients = oracle_clients() if oracle else model_clients(settings)
    records = [
        await run_case(case, settings, clients, concurrency=concurrency)
        for case in cases
    ]
    label = ORACLE if oracle else settings.llm.model
    model = (
        {"model": ORACLE}
        if oracle
        else {
            key: value
            for key, value in cache_key_config(settings.llm).items()
            if key != "cache_format_version"
        }
    )
    row = Row(label=label, model=model, cases=records)
    if not oracle:
        row.usd = price(row.counts, (prices or {}).get(label))
    return row