Skip to content

schemacast.bench.saved

Score saved cast output against a corpus, without a model.

A saved cast is a JSON file holding a POST /cast response body, or a wrapper {"episodes": [...], "cast": <body>}: the episodes sent and the body returned. Each episode's record graph is scored as the run scores a fresh cast. What only a live cast knows — the first reply before repair, the slice, tokens, a re-cast, picks — is not measured and is reported as such.

Attributes

NOT_SAVED = 'not_saved' module-attribute

UNMEASURED_COUNTS = frozenset({'emitted_first', 'nonconforming_first', 'emitted_final', 'nonconforming_final', 'ungrounded_first', 'vertices_first', 'unknown_type_first', 'edges_first', 'undeclared_edge_first', 'properties_first', 'undeclared_property_first', 'unfaithful_value_first', 'sliced', 'gold_types', 'gold_types_seeded', 'gold_types_in_slice', 'gold_edges', 'gold_edges_in_slice', 'recast_vertices', 'recast_added', 'recast_minted', 'recast_duplicated', 'picks_expected', 'picks_offered', 'picks_correct', 'picks_linked', 'picks_wrong', 'picks_rejected', 'input_tokens', 'output_tokens', 'billed_input_tokens', 'billed_output_tokens'}) module-attribute

Classes

SavedCastError

Bases: ValueError

A file that is not a saved cast, or two casts of one episode.

Source code in schemacast/bench/saved.py 
66
67
class SavedCastError(ValueError):
    """A file that is not a saved cast, or two casts of one episode."""

SavedCasts

Episode outcomes read from saved casts, by source id.

Source code in schemacast/bench/saved.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
class SavedCasts:
    """Episode outcomes read from saved casts, by source id."""

    def __init__(self) -> None:
        self.outcomes: dict[str, EpisodeOutcome] = {}
        self.models: list[dict[str, Any]] = []

    def add(self, path: Path) -> None:
        """Read one saved cast.

        Raises:
            SavedCastError: not a cast body or wrapper, or an episode cast twice.
        """
        try:
            data = json.loads(path.read_text(encoding="utf-8"))
        except json.JSONDecodeError as error:
            raise SavedCastError(f"{path}: not JSON: {error.msg}") from error
        body = data.get("cast") if isinstance(data, dict) and "cast" in data else data
        if not isinstance(body, dict) or not {"mentions", "episodes"} <= set(body):
            raise SavedCastError(f"{path}: not a cast body (mentions, episodes)")
        try:
            graphs = {
                graph.source_id: graph
                for graph in (RecordGraph.model_validate(g) for g in body["mentions"])
            }
            results = {
                result.source_id: result
                for result in (
                    EpisodeResult.model_validate(r) for r in body["episodes"]
                )
            }
        except ValueError as error:
            raise SavedCastError(f"{path}: {error}") from error
        model = (body.get("manifest") or {}).get("model")
        if isinstance(model, dict):
            self.models.append(model)
        for source_id in [*results, *(g for g in graphs if g not in results)]:
            if source_id in self.outcomes:
                raise SavedCastError(f"{path}: episode {source_id!r} is cast twice")
            self.outcomes[source_id] = outcome(
                results.get(source_id)
                or EpisodeResult(source_id=source_id, status="ok"),
                graphs.get(source_id),
            )

    def label(self, paths: list[Path]) -> str:
        """The model the casts name, when they agree; else the first file's stem."""
        names = {str(model.get("model")) for model in self.models}
        if len(names) == 1 and len(self.models) == len(paths):
            return names.pop()
        return paths[0].stem

Attributes

models = [] instance-attribute
outcomes = {} instance-attribute

Methods:

__init__()
Source code in schemacast/bench/saved.py 
73
74
75
def __init__(self) -> None:
    self.outcomes: dict[str, EpisodeOutcome] = {}
    self.models: list[dict[str, Any]] = []
add(path)

Read one saved cast.

Raises:

Type Description
SavedCastError

not a cast body or wrapper, or an episode cast twice.
Source code in schemacast/bench/saved.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def add(self, path: Path) -> None:
    """Read one saved cast.

    Raises:
        SavedCastError: not a cast body or wrapper, or an episode cast twice.
    """
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as error:
        raise SavedCastError(f"{path}: not JSON: {error.msg}") from error
    body = data.get("cast") if isinstance(data, dict) and "cast" in data else data
    if not isinstance(body, dict) or not {"mentions", "episodes"} <= set(body):
        raise SavedCastError(f"{path}: not a cast body (mentions, episodes)")
    try:
        graphs = {
            graph.source_id: graph
            for graph in (RecordGraph.model_validate(g) for g in body["mentions"])
        }
        results = {
            result.source_id: result
            for result in (
                EpisodeResult.model_validate(r) for r in body["episodes"]
            )
        }
    except ValueError as error:
        raise SavedCastError(f"{path}: {error}") from error
    model = (body.get("manifest") or {}).get("model")
    if isinstance(model, dict):
        self.models.append(model)
    for source_id in [*results, *(g for g in graphs if g not in results)]:
        if source_id in self.outcomes:
            raise SavedCastError(f"{path}: episode {source_id!r} is cast twice")
        self.outcomes[source_id] = outcome(
            results.get(source_id)
            or EpisodeResult(source_id=source_id, status="ok"),
            graphs.get(source_id),
        )
label(paths)

The model the casts name, when they agree; else the first file's stem.

Source code in schemacast/bench/saved.py 
115
116
117
118
119
120
def label(self, paths: list[Path]) -> str:
    """The model the casts name, when they agree; else the first file's stem."""
    names = {str(model.get("model")) for model in self.models}
    if len(names) == 1 and len(self.models) == len(paths):
        return names.pop()
    return paths[0].stem

Functions:

outcome(result, graph)

An episode outcome as a fresh cast would hold it, from its saved parts.

Source code in schemacast/bench/saved.py 
123
124
125
126
127
128
129
130
131
132
133
134
def outcome(result: EpisodeResult, graph: RecordGraph | None) -> EpisodeOutcome:
    """An episode outcome as a fresh cast would hold it, from its saved parts."""
    if graph is None:
        return EpisodeOutcome(result=result)
    known = [r for r in graph.records if "known" in (r.evidence or {})]
    validation = Validation(
        mentions=[r for r in graph.records if r not in known],
        relations=list(graph.relations),
        known=known,
        fixes=list(result.fixes),
    )
    return EpisodeOutcome(result=result, validation=validation, graph=graph)

score_saved(cases, saved, label, *, corpus_ids=None)

One row: every episode of cases scored from saved.

An episode with no saved cast counts as failed and fully missed. A saved cast of an episode not in corpus_ids (default: those of cases) is listed in unmatched_casts.

Source code in schemacast/bench/saved.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def score_saved(
    cases: list[BenchCase],
    saved: SavedCasts,
    label: str,
    *,
    corpus_ids: set[str] | None = None,
) -> Row:
    """One row: every episode of *cases* scored from *saved*.

    An episode with no saved cast counts as failed and fully missed. A saved
    cast of an episode not in *corpus_ids* (default: those of *cases*) is
    listed in ``unmatched_casts``.
    """
    records: list[CaseRecord] = []
    for case in cases:
        episodes: list[EpisodeRecord] = []
        for episode in case.episodes:
            found = saved.outcomes.get(episode.source_id) or EpisodeOutcome(
                result=EpisodeResult(
                    source_id=episode.source_id, status="failed", failure=NOT_SAVED
                )
            )
            context = {item.source_id: item.text for item in episode.context}
            misses = Misses()
            counts = score_episode(
                found,
                case.expected[episode.source_id],
                case.schema_,
                episode.text,
                None,
                context,
                misses,
                episode.known,
            )
            episodes.append(
                EpisodeRecord(
                    source_id=episode.source_id,
                    status=found.result.status,
                    failure=found.result.failure,
                    counts=unmeasured(counts),
                    misses=misses,
                )
            )
        records.append(
            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,
            )
        )
    known = (
        corpus_ids
        if corpus_ids is not None
        else {e.source_id for case in cases for e in case.episodes}
    )
    return Row(
        label=label,
        model=saved.models[0] if saved.models else {},
        cases=records,
        saved=True,
        unmatched_casts=sorted(set(saved.outcomes) - known),
    )

unmeasured(counts)

counts with what a saved cast cannot supply set to zero.

Source code in schemacast/bench/saved.py 
137
138
139
def unmeasured(counts: Counts) -> Counts:
    """*counts* with what a saved cast cannot supply set to zero."""
    return counts.model_copy(update=dict.fromkeys(UNMEASURED_COUNTS, 0))