schemacast.bench.run

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

Attributes

ClientFactory = Callable[[BenchCase, SchemaCastSettings, PickScripts], LLMClient] module-attribute

PRICES_ENV = 'SCHEMACAST_BENCH_PRICES' module-attribute

PickScripts = dict[str, str] module-attribute

RECAST_CACHE_SUBDIR = 'recast' module-attribute

Classes

BenchReport

Bases: BaseModel

A benchmark run.

Source code in schemacast/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."
    )
    corpus_dir: str | None = Field(
        default=None,
        description="A corpus outside the package (``--corpus``): never publishable.",
    )
    split: str
    schemacast_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
            and self.corpus_dir is None
        )

Attributes

corpus_digest instance-attribute

corpus_dir = Field(default=None, description='A corpus outside the package (``--corpus``): never publishable.') class-attribute 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

prices_source = None class-attribute instance-attribute

publishable property

rows instance-attribute

schemacast_version = __version__ class-attribute instance-attribute

split instance-attribute

started instance-attribute

CaseRecord

Bases: BaseModel

One case's episodes and their sum.

Source code in schemacast/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 schemacast/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
    misses: Misses = Field(
        default_factory=Misses,
        description="Expected items not matched and produced items matching none.",
    )

Attributes

counts instance-attribute

failure instance-attribute

misses = Field(default_factory=Misses, description='Expected items not matched and produced items matching none.') class-attribute instance-attribute

source_id instance-attribute

status instance-attribute

Row

Bases: BaseModel

Every selected case cast by one model.

Source code in schemacast/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]
    saved: bool = Field(
        default=False,
        description="Scored from saved casts: the first reply, the slice, tokens, "
        "re-casts and picks are not measured.",
    )
    unmatched_casts: list[str] = Field(
        default_factory=list,
        description="Source ids of saved casts with no episode in the corpus; ignored.",
    )

    @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

saved = Field(default=False, description='Scored from saved casts: the first reply, the slice, tokens, re-casts and picks are not measured.') class-attribute instance-attribute

unmatched_casts = Field(default_factory=list, description='Source ids of saved casts with no episode in the corpus; ignored.') 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 schemacast/bench/run.py

def case_settings(case: BenchCase, base: SchemaCastSettings) -> SchemaCastSettings:
    """*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 schemacast/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 schemacast/bench/run.py

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

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

    return build

new_report(split, started, rows, prices_source, corpus=None)

A report over rows, stamped with the corpus it ran against: the bundled one, or the one under corpus (cases/ and schemas/).

Source code in schemacast/bench/run.py

def new_report(
    split: str,
    started: str,
    rows: list[Row],
    prices_source: str | None,
    corpus: Path | None = None,
) -> BenchReport:
    """A report over *rows*, stamped with the corpus it ran against: the
    bundled one, or the one under *corpus* (``cases/`` and ``schemas/``)."""
    digest = corpus_digest() if corpus is None else corpus_digest(corpus)
    return BenchReport(
        corpus_version=CORPUS_VERSION if corpus is None else "external",
        corpus_digest=digest,
        corpus_modified=digest != CORPUS_DIGEST,
        corpus_dir=None if corpus is None else str(corpus),
        split=split,
        started=started,
        prices_source=prices_source,
        rows=rows,
    )

oracle_clients()

Oracle clients, scripted per case.

Source code in schemacast/bench/run.py

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

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

    return build

pick_item(outcome, episode, spec, refs, schema)

The pick input for a cast episode's staged mentions the case gives candidates for, and the expected key per mention ref.

Source code in schemacast/bench/run.py

def pick_item(
    outcome: EpisodeOutcome,
    episode: Episode,
    spec: ExpectedPicks,
    refs: dict[str, str],
    schema: Any,
) -> tuple[PickItem, dict[str, dict[str, Any] | None]]:
    """The pick input for a cast episode's staged mentions the case gives
    candidates for, and the expected key per mention ref."""
    from graflo.link.records import EntityCandidate, StagedCandidates
    from graflo.link.resolve import resolve_records

    graph = outcome.graph
    resolution = resolve_records(schema, [graph] if graph is not None else [])
    by_handle = {handle: ref for ref, handle in refs.items()}

    def entity(item: Any) -> EntityCandidate:
        return EntityCandidate(
            vertex_type=item.type,
            key=item.key,
            label=item.label,
            lane=item.lane,
            matched=item.matched,
            relation=item.relation,
            anchor=by_handle.get(item.anchor, item.anchor) if item.anchor else None,
            role=item.role,
        )

    staged = []
    wanted: dict[str, dict[str, Any] | None] = {}
    for entry in resolution.staged.records:
        handle = refs.get(entry.record.ref)
        if handle not in spec.mentions:
            continue
        expected = spec.mentions[handle]
        staged.append(
            StagedCandidates(
                source_id=entry.source_id,
                record=entry.record,
                reason=entry.reason,
                candidates=[entity(c) for c in expected.candidates],
            )
        )
        wanted[entry.record.ref] = expected.pick
    item = PickItem(
        episode=episode,
        staged=staged,
        entities=[entity(e) for e in spec.entities],
        graph=graph,
    )
    return item, wanted

price(counts, entry)

USD for counts' tokens at entry's rates.

Source code in schemacast/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 schemacast/bench/run.py

def recast_settings(base: SchemaCastSettings) -> SchemaCastSettings:
    """*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 schemacast/bench/run.py

async def run_case(
    case: BenchCase,
    settings: SchemaCastSettings,
    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)
    picks: PickScripts = {}
    caster = SchemaCaster(
        case.schema_, settings=pinned, client=clients(case, pinned, picks)
    )
    recaster = None
    if case.recast:
        again = recast_settings(pinned)
        recaster = SchemaCaster(
            case.schema_, settings=again, client=clients(case, again, picks)
        )
    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,
                        picks,
                    )
                )
                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 schemacast/bench/run.py

async def run_row(
    cases: list[BenchCase],
    settings: SchemaCastSettings,
    *,
    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