schemacast.validate

Validate a reply against the slice it was asked for.

Complete on purpose: graflo's writer silently drops an edge whose key no edge declares and a vertex with no identity value, which is right for a writer and wrong for an extraction. Every such case is a finding here.

Deterministic fixes run first and are recorded: values are coerced to their field type (declared units stripped), an edge given in the reverse of its declared direction is flipped, and quotes are matched ignoring case and whitespace runs. A property's quote is read nearest its owner's. An identity value the text refers to but only the episode's context states is grounded there. A number written as digits, or a date, must be what its quote states. What remains are findings, each naming the item it removes; an item with a finding is left out, and the rest of the reply stands.

Validation decides no identity. A mention that introduces an entity but states no complete key gets one non-removing finding, so a repair can ask for a key the reply left out; the mention is kept either way, and resolution decides.

Kept mentions and relations are graflo records (PartialRecord, RecordRelation); each one's quote and span, and each property's, are its evidence (:func:schemacast.ground.evidence).

An episode's known entities are refs k1… (:func:schemacast.prompt.known_refs) an edge may end at. A vertex the reply gives such a ref is a known_ref finding. Each known entity a kept relation uses becomes a record in :attr:Validation.known, stating its key from the source: its evidence is {"known": role, "label": label}, with no quote or span.

Attributes

CONSEQUENTIAL = frozenset({'endpoint_invalid'}) module-attribute

FINDING_ORDER = ('unknown_type', 'known_ref', 'duplicate_ref', 'identity_insufficient', 'undeclared_edge', 'dangling_ref', 'undeclared_property', 'duplicate_property', 'coercion', 'value_not_in_quote', 'quote_not_found', 'endpoint_invalid') module-attribute

FindingKind = Literal['known_ref', 'duplicate_ref', 'unknown_type', 'undeclared_property', 'duplicate_property', 'coercion', 'value_not_in_quote', 'quote_not_found', 'identity_insufficient', 'dangling_ref', 'undeclared_edge', 'endpoint_invalid'] module-attribute

Scope = Literal['vertex', 'edge', 'property'] module-attribute

Classes

Finding

Bases: BaseModel

A defect in a reply, and the item it removes.

Source code in schemacast/validate.py

class Finding(BaseModel):
    """A defect in a reply, and the item it removes."""

    kind: FindingKind
    scope: Scope = Field(
        description="What is left out: a vertex, an edge or one property."
    )
    path: str = Field(
        description="Location in the reply, e.g. ``vertices[1].props[0]``."
    )
    ref: str | None = Field(default=None, description="Vertex ref involved, if any.")
    detail: str | None = Field(
        default=None, description="Sub-kind, e.g. the coercion kind."
    )
    message: str
    fragment: dict[str, Any] = Field(
        description="The offending item as the model wrote it."
    )
    removes: bool = Field(
        default=True,
        description="False when the item is kept and the finding only asks a repair.",
    )

Attributes

detail = Field(default=None, description='Sub-kind, e.g. the coercion kind.') class-attribute instance-attribute

fragment = Field(description='The offending item as the model wrote it.') class-attribute instance-attribute

kind instance-attribute

message instance-attribute

path = Field(description='Location in the reply, e.g. ``vertices[1].props[0]``.') class-attribute instance-attribute

ref = Field(default=None, description='Vertex ref involved, if any.') class-attribute instance-attribute

removes = Field(default=True, description='False when the item is kept and the finding only asks a repair.') class-attribute instance-attribute

scope = Field(description='What is left out: a vertex, an edge or one property.') class-attribute instance-attribute

Fix

Bases: BaseModel

A deterministic correction applied without asking the model.

Source code in schemacast/validate.py

class Fix(BaseModel):
    """A deterministic correction applied without asking the model."""

    kind: Literal[
        "coerced",
        "edge_reversed",
        "quote_normalized",
        "grounded_in_context",
    ]
    path: str
    message: str

Attributes

kind instance-attribute

message instance-attribute

path instance-attribute

Validation

Bases: BaseModel

The checked reply.

Source code in schemacast/validate.py

class Validation(BaseModel):
    """The checked reply."""

    mentions: list[PartialRecord] = Field(default_factory=list)
    relations: list[RecordRelation] = Field(default_factory=list)
    known: list[PartialRecord] = Field(
        default_factory=list,
        description="Known entities a kept relation uses (refs ``k1``…): their "
        "key from the source, never created, never grounded.",
    )
    unmapped: list[UnmappedText] = Field(default_factory=list)
    findings: list[Finding] = Field(default_factory=list)
    fixes: list[Fix] = Field(default_factory=list)

    @property
    def repairable(self) -> list[Finding]:
        """Findings worth showing the model: all but consequences of others."""
        return [f for f in self.findings if f.kind not in CONSEQUENTIAL]

    @property
    def removed(self) -> list[Finding]:
        """Findings that left an item out."""
        return [f for f in self.findings if f.removes]

    @property
    def kept(self) -> int:
        """Items kept: mentions, relations and properties. Used to pick the best attempt."""
        return (
            len(self.mentions)
            + len(self.relations)
            + sum(len(m.props) for m in self.mentions)
            + sum(len(r.props) for r in self.relations)
        )

    def graph(
        self, source_id: str, schema_ref: str, reference_time: Any = None
    ) -> RecordGraph:
        """The record graph this reply states; ``unmapped`` stays here."""
        return RecordGraph(
            source_id=source_id,
            schema_ref=schema_ref,
            reference_time=reference_time,
            records=[*self.mentions, *self.known],
            relations=self.relations,
        )

Attributes

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

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

kept property

Items kept: mentions, relations and properties. Used to pick the best attempt.

known = Field(default_factory=list, description='Known entities a kept relation uses (refs ``k1``…): their key from the source, never created, never grounded.') class-attribute instance-attribute

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

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

removed property

Findings that left an item out.

repairable property

Findings worth showing the model: all but consequences of others.

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

Methods:

graph(source_id, schema_ref, reference_time=None)

The record graph this reply states; unmapped stays here.

Source code in schemacast/validate.py

def graph(
    self, source_id: str, schema_ref: str, reference_time: Any = None
) -> RecordGraph:
    """The record graph this reply states; ``unmapped`` stays here."""
    return RecordGraph(
        source_id=source_id,
        schema_ref=schema_ref,
        reference_time=reference_time,
        records=[*self.mentions, *self.known],
        relations=self.relations,
    )

Functions:

validate(wire, slice_, text, context=(), known=())

Check wire against slice_, the episode text, its context and its known entities.

Source code in schemacast/validate.py

def validate(
    wire: WireExtraction,
    slice_: Schema,
    text: str,
    context: Sequence[ContextItem] = (),
    known: Sequence[KnownEntity] = (),
) -> Validation:
    """Check *wire* against *slice_*, the episode *text*, its *context* and its
    *known* entities."""
    checker = _Checker(slice_, text, context, known)
    refs = checker.vertices(wire)
    checker.edges(wire, refs)
    checker.known_records()
    checker.unmapped(wire)
    order = {kind: i for i, kind in enumerate(FINDING_ORDER)}
    checker.result.findings.sort(key=lambda f: order[f.kind])
    return checker.result