lpgcast.bench.corpus

The benchmark corpus: texts, the schema they are cast against, and the graph each text states.

A case is data, never code: cases/<family>/<instance>/ holds case.yaml, episodes.jsonl and expected.yaml. case.yaml names its schema from schemas/, either one file or a list whose vertices and edges are merged. A family isolates one failure mode; its instances are that failure mode in different domains.

The expected graph is exact: every quote it gives occurs in the text, and every value is what the schema's field makes of it. test_bench checks both, and that an oracle replying with the expected graph scores perfectly.

Expected graphs are an interface: changing a case invalidates every recorded result, so :data:CORPUS_VERSION moves with it and :data:CORPUS_DIGEST pins the two together.

Attributes

BENCH_DIR = Path(__file__).parent module-attribute

CASES_DIR = BENCH_DIR / 'cases' module-attribute

CORPUS_DIGEST = '9378eb758efbd539a103b1819254704271f4f6e4f3447ae021297ec405ed23ee' module-attribute

CORPUS_VERSION = '1.1.0' module-attribute

SCHEMAS_DIR = BENCH_DIR / 'schemas' module-attribute

SPLITS_FILE = CASES_DIR / 'SPLITS.yaml' module-attribute

Split = Literal['dev', 'test'] module-attribute

Classes

BenchCase

Bases: BaseModel

One case, loaded.

Source code in lpgcast/bench/corpus.py

class BenchCase(BaseModel):
    """One case, loaded."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    name: str
    family: str
    instance: str
    split: Split
    description: str
    schema_: Schema = Field(alias="schema")
    recast: bool
    settings: CaseSettings
    episodes: list[Episode]
    expected: dict[str, ExpectedEpisode]
    digest: str = Field(
        description="SHA-256 of the case's files, its split and its schemas: "
        "two runs compare on a case only when this matches."
    )

Attributes

description instance-attribute

digest = Field(description="SHA-256 of the case's files, its split and its schemas: two runs compare on a case only when this matches.") class-attribute instance-attribute

episodes instance-attribute

expected instance-attribute

family instance-attribute

instance instance-attribute

model_config = ConfigDict(arbitrary_types_allowed=True) class-attribute instance-attribute

name instance-attribute

recast instance-attribute

schema_ = Field(alias='schema') class-attribute instance-attribute

settings instance-attribute

split instance-attribute

CaseMeta

Bases: BaseModel

case.yaml.

Source code in lpgcast/bench/corpus.py

class CaseMeta(BaseModel):
    """``case.yaml``."""

    model_config = ConfigDict(extra="forbid")

    description: str
    schema_: str | list[str] = Field(alias="schema")
    recast: bool = Field(
        default=False,
        description="Cast a second time with a fresh model call and compare.",
    )
    settings: CaseSettings = Field(default_factory=CaseSettings)

Attributes

description instance-attribute

model_config = ConfigDict(extra='forbid') class-attribute instance-attribute

recast = Field(default=False, description='Cast a second time with a fresh model call and compare.') class-attribute instance-attribute

schema_ = Field(alias='schema') class-attribute instance-attribute

settings = Field(default_factory=CaseSettings) class-attribute instance-attribute

CaseSettings

Bases: BaseModel

Settings a case pins, e.g. a slice budget tight enough to select.

Source code in lpgcast/bench/corpus.py

class CaseSettings(BaseModel):
    """Settings a case pins, e.g. a slice budget tight enough to select."""

    model_config = ConfigDict(extra="forbid")

    slice_max_elements: int | None = None
    slice_max_tokens: int | None = None

Attributes

model_config = ConfigDict(extra='forbid') class-attribute instance-attribute

slice_max_elements = None class-attribute instance-attribute

slice_max_tokens = None class-attribute instance-attribute

ExpectedEdge

Bases: BaseModel

A relation the text states, between two vertex handles.

Source code in lpgcast/bench/corpus.py

class ExpectedEdge(BaseModel):
    """A relation the text states, between two vertex handles."""

    model_config = ConfigDict(extra="forbid", populate_by_name=True)

    relation: str
    from_: str = Field(alias="from")
    to: str
    quote: str
    props: dict[str, ExpectedProp] = Field(default_factory=dict)

    @field_validator("props", mode="before")
    @classmethod
    def _wrap_scalars(cls, value: Any) -> Any:
        return _props(value)

Attributes

from_ = Field(alias='from') class-attribute instance-attribute

model_config = ConfigDict(extra='forbid', populate_by_name=True) class-attribute instance-attribute

props = Field(default_factory=dict) class-attribute instance-attribute

quote instance-attribute

relation instance-attribute

to instance-attribute

ExpectedEpisode

Bases: ExpectedGraph

What one episode states, and what a cast of it must not write.

forbidden holds items an obedient or careless model would produce: what an injected instruction asks for, or an entity the text gives no identity. Its edges may use handles of either graph.

Source code in lpgcast/bench/corpus.py

class ExpectedEpisode(ExpectedGraph):
    """What one episode states, and what a cast of it must not write.

    ``forbidden`` holds items an obedient or careless model would produce: what
    an injected instruction asks for, or an entity the text gives no identity.
    Its edges may use handles of either graph.
    """

    forbidden: ExpectedGraph = Field(default_factory=ExpectedGraph)

Attributes

forbidden = Field(default_factory=ExpectedGraph) class-attribute instance-attribute

ExpectedGraph

Bases: BaseModel

Vertices keyed by a handle local to the episode, and edges between handles.

Source code in lpgcast/bench/corpus.py

class ExpectedGraph(BaseModel):
    """Vertices keyed by a handle local to the episode, and edges between handles."""

    model_config = ConfigDict(extra="forbid")

    vertices: dict[str, ExpectedVertex] = Field(default_factory=dict)
    edges: list[ExpectedEdge] = Field(default_factory=list)

Attributes

edges = Field(default_factory=list) class-attribute instance-attribute

model_config = ConfigDict(extra='forbid') class-attribute instance-attribute

vertices = Field(default_factory=dict) class-attribute instance-attribute

ExpectedProp

Bases: BaseModel

A property value as the schema's field holds it, and the text it is read from.

quote defaults to the value as written; give it when the text writes the value differently (200 kW for 200.0, 4 March 2024 for a date).

Source code in lpgcast/bench/corpus.py

class ExpectedProp(BaseModel):
    """A property value as the schema's field holds it, and the text it is read from.

    ``quote`` defaults to the value as written; give it when the text writes the
    value differently (``200 kW`` for ``200.0``, ``4 March 2024`` for a date).
    """

    model_config = ConfigDict(extra="forbid")

    value: Any
    quote: str | None = None
    context: str | None = Field(
        default=None,
        description="Source id of the context item the quote is in; only an "
        "identity value may be read from context.",
    )

    @property
    def evidence(self) -> str:
        """The quote, or the value as written."""
        if self.quote is not None:
            return self.quote
        return str(self.value)

Attributes

context = Field(default=None, description='Source id of the context item the quote is in; only an identity value may be read from context.') class-attribute instance-attribute

evidence property

The quote, or the value as written.

model_config = ConfigDict(extra='forbid') class-attribute instance-attribute

quote = None class-attribute instance-attribute

value instance-attribute

ExpectedVertex

Bases: BaseModel

An entity the text states or refers to.

Source code in lpgcast/bench/corpus.py

class ExpectedVertex(BaseModel):
    """An entity the text states or refers to."""

    model_config = ConfigDict(extra="forbid")

    type: str
    quote: str = Field(description="Where the text names the entity.")
    mention: bool = Field(
        default=False,
        description="The text only refers to it: never written, looked up by identity.",
    )
    props: dict[str, ExpectedProp] = Field(default_factory=dict)

    @field_validator("props", mode="before")
    @classmethod
    def _wrap_scalars(cls, value: Any) -> Any:
        return _props(value)

Attributes

mention = Field(default=False, description='The text only refers to it: never written, looked up by identity.') class-attribute instance-attribute

model_config = ConfigDict(extra='forbid') class-attribute instance-attribute

props = Field(default_factory=dict) class-attribute instance-attribute

quote = Field(description='Where the text names the entity.') class-attribute instance-attribute

type instance-attribute

Functions:

corpus_digest(bench_dir=BENCH_DIR)

SHA-256 over every file of cases/ and schemas/, paths included.

Source code in lpgcast/bench/corpus.py

def corpus_digest(bench_dir: Path = BENCH_DIR) -> str:
    """SHA-256 over every file of ``cases/`` and ``schemas/``, paths included."""
    digest = hashlib.sha256()
    for sub in ("cases", "schemas"):
        for path in sorted((bench_dir / sub).rglob("*")):
            if path.is_file() and "__pycache__" not in path.parts:
                digest.update(path.relative_to(bench_dir).as_posix().encode("utf-8"))
                digest.update(path.read_bytes())
    return digest.hexdigest()

load_case(directory, *, cases_dir=CASES_DIR, schemas_dir=SCHEMAS_DIR, splits=None)

Load one case directory.

Raises:

Type	Description
ValueError	the case has no split, or its expected graph and episodes name different source ids.

Source code in lpgcast/bench/corpus.py

def load_case(
    directory: Path,
    *,
    cases_dir: Path = CASES_DIR,
    schemas_dir: Path = SCHEMAS_DIR,
    splits: dict[str, Split] | None = None,
) -> BenchCase:
    """Load one case directory.

    Raises:
        ValueError: the case has no split, or its expected graph and episodes
            name different source ids.
    """
    name = directory.relative_to(cases_dir).as_posix()
    family, _, instance = name.partition("/")
    assignment = load_splits(cases_dir / "SPLITS.yaml") if splits is None else splits
    if name not in assignment:
        raise ValueError(f"{name}: not assigned to dev or test in SPLITS.yaml")
    meta = CaseMeta.model_validate(_read_yaml(directory / "case.yaml"))
    episodes = list(read_episodes(directory / "episodes.jsonl"))
    raw = _read_yaml(directory / "expected.yaml") or {}
    expected = {
        source_id: ExpectedEpisode.model_validate(item or {})
        for source_id, item in raw.items()
    }
    ids = [episode.source_id for episode in episodes]
    if sorted(ids) != sorted(expected) or len(set(ids)) != len(ids):
        raise ValueError(
            f"{name}: episodes {sorted(ids)} and expected {sorted(expected)} differ"
        )
    parts = [meta.schema_] if isinstance(meta.schema_, str) else meta.schema_
    digest = hashlib.sha256(f"{name}\0{assignment[name]}".encode())
    for path in [
        directory / "case.yaml",
        directory / "episodes.jsonl",
        directory / "expected.yaml",
        *(schemas_dir / f"{part}.yaml" for part in parts),
    ]:
        digest.update(path.name.encode("utf-8"))
        digest.update(path.read_bytes())
    return BenchCase(
        name=name,
        family=family,
        instance=instance or family,
        split=assignment[name],
        description=meta.description,
        schema=load_schema_parts(meta.schema_, schemas_dir),
        recast=meta.recast,
        settings=meta.settings,
        episodes=episodes,
        expected=expected,
        digest=digest.hexdigest(),
    )

load_corpus(cases_dir=CASES_DIR, schemas_dir=SCHEMAS_DIR)

Every case, sorted by name.

Source code in lpgcast/bench/corpus.py

def load_corpus(
    cases_dir: Path = CASES_DIR, schemas_dir: Path = SCHEMAS_DIR
) -> list[BenchCase]:
    """Every case, sorted by name."""
    splits = load_splits(cases_dir / "SPLITS.yaml")
    return [
        load_case(
            path.parent, cases_dir=cases_dir, schemas_dir=schemas_dir, splits=splits
        )
        for path in sorted(cases_dir.rglob("case.yaml"))
    ]

load_schema_parts(names, schemas_dir=SCHEMAS_DIR)

One schema from schemas/<name>.yaml files.

Vertices and edges are concatenated in order; metadata is the last file's.

Source code in lpgcast/bench/corpus.py

def load_schema_parts(
    names: str | list[str], schemas_dir: Path = SCHEMAS_DIR
) -> Schema:
    """One schema from ``schemas/<name>.yaml`` files.

    Vertices and edges are concatenated in order; metadata is the last file's.
    """
    parts = [names] if isinstance(names, str) else names
    vertices: list[Any] = []
    edges: list[Any] = []
    metadata: Any = None
    for name in parts:
        data = _read_yaml(schemas_dir / f"{name}.yaml")
        core = data["core_schema"]
        vertices += core["vertex_config"]["vertices"]
        edges += core["edge_config"]["edges"]
        metadata = data.get("metadata", metadata)
    return Schema.model_validate(
        {
            "metadata": metadata,
            "core_schema": {
                "vertex_config": {"vertices": vertices},
                "edge_config": {"edges": edges},
            },
        }
    )

load_splits(path=SPLITS_FILE)

Case name → dev or test.

Source code in lpgcast/bench/corpus.py

def load_splits(path: Path = SPLITS_FILE) -> dict[str, Split]:
    """Case name → ``dev`` or ``test``."""
    raw = _read_yaml(path) or {}
    return {name: half for half in ("dev", "test") for name in raw.get(half) or []}