ontocast.tool.chunk.schema_detect

Detect which document-type schema a document belongs to.

Section labels are only meaningful relative to a schema: a 10-Q scored against the academic schema recognises almost nothing and comes back unlabeled. When the caller supplies neither section_schema_id nor a matching document_type_hint, the schema has to be inferred from the document itself.

The schema catalog is a partition -- every document belongs to exactly one cell, with general as the residual. Only schemas carrying a document_profile are detection candidates; general deliberately has none, because six of its seven labels are strict subsets of other schemas.

Three tiers, cheapest first, each seeing only what the previous one could not decide:

1. Lexical (free, no model). Scores on exclusive evidence: a heading recognised by exactly one candidate schema counts, a heading recognised by several counts nothing. A shared heading such as References genuinely says nothing about which cell the document is in, so weighting it -- even fractionally -- only adds noise. Measured over the corpus in test/data/schema_corpus.json this classifies all nine cells correctly, with the tightest margin (a published trial protocol, which shares IMRaD headings with academic papers) at 2.0x.
2. Semantic (needs the chunker's embedding model). Each heading votes for its nearest schema by label-name similarity, damped by how close the runner-up is. Per-heading competition rather than per-schema mean similarity, because mean similarity is biased by how many label names a schema happens to have.
3. Content (last resort, heading-poor documents only). Body paragraphs against document_profile sentences. Measured accuracy is far below the heading tiers -- chemistry prose reads like a technical specification, and news behaves as a semantic attractor -- so it demands a much larger margin and is off unless explicitly enabled.

Every tier abstains rather than guessing. A wrong schema relabels an entire document silently, whereas an abstention falls back to the manifest default and leaves document_type_hint in charge.

SchemaDetection dataclass

An accepted detection, with the tier that made it and its evidence.

Source code in ontocast/tool/chunk/schema_detect.py

@dataclass(frozen=True)
class SchemaDetection:
    """An accepted detection, with the tier that made it and its evidence."""

    schema_id: str
    tier: str
    score: float
    margin: float
    share: float
    evidence: list[SchemaEvidence]

SchemaEvidence dataclass

Why one candidate schema scored what it did.

Source code in ontocast/tool/chunk/schema_detect.py

@dataclass(frozen=True)
class SchemaEvidence:
    """Why one candidate schema scored what it did."""

    schema_id: str
    score: float
    share: float
    examples: list[str] = field(default_factory=list)

TextEmbedder

Bases: Protocol

Embeds short texts, or returns None when no model is available.

Source code in ontocast/tool/chunk/schema_detect.py

class TextEmbedder(Protocol):
    """Embeds short texts, or returns ``None`` when no model is available."""

    def __call__(self, texts: list[str]) -> list[list[float]] | None: ...

candidate_schema_ids()

Schemas eligible to be detected, in manifest order.

A schema is a candidate only if it declares a document_profile. That is what keeps general -- the partition's residual cell -- from ever being a positive detection.

Source code in ontocast/tool/chunk/schema_detect.py

def candidate_schema_ids() -> tuple[str, ...]:
    """Schemas eligible to be detected, in manifest order.

    A schema is a candidate only if it declares a ``document_profile``. That is
    what keeps ``general`` -- the partition's residual cell -- from ever being a
    positive detection.
    """
    ids: list[str] = []
    for entry in load_manifest().schemas:
        if load_section_label_schema(entry.id).document_profile.strip():
            ids.append(entry.id)
    return tuple(ids)

detect_document_schema(headings, paragraphs=None, *, embed=None, allow_content_tier=False, min_score=2.0, min_margin=1.8, min_share=0.0, content_min_margin=4.0)

Infer the document-type schema, or None when the evidence is thin.

Parameters:

Name	Type	Description	Default
headings	list[str]	Raw heading lines in document order.	required
paragraphs	list[str] | None	Sampled body paragraphs, for the content tier.	None
embed	TextEmbedder | None	Embedding callable; None restricts detection to the lexical tier, exactly as missing semantic extras do.	None
allow_content_tier	bool	Permit the weak content tier on heading-poor documents.	False
min_score	float	Absolute evidence the winner must clear.	2.0
min_margin	float	Factor by which the winner must beat the runner-up.	1.8
min_share	float	Minimum fraction of voting items backing the winner.	0.0
content_min_margin	float	Stricter margin for the content tier.	4.0

Returns:

Type	Description
SchemaDetection | None	The accepted detection, or None to fall back to the caller's
SchemaDetection | None	default.

Source code in ontocast/tool/chunk/schema_detect.py

def detect_document_schema(
    headings: list[str],
    paragraphs: list[str] | None = None,
    *,
    embed: TextEmbedder | None = None,
    allow_content_tier: bool = False,
    min_score: float = 2.0,
    min_margin: float = 1.8,
    min_share: float = 0.0,
    content_min_margin: float = 4.0,
) -> SchemaDetection | None:
    """Infer the document-type schema, or ``None`` when the evidence is thin.

    Args:
        headings: Raw heading lines in document order.
        paragraphs: Sampled body paragraphs, for the content tier.
        embed: Embedding callable; ``None`` restricts detection to the lexical
            tier, exactly as missing semantic extras do.
        allow_content_tier: Permit the weak content tier on heading-poor
            documents.
        min_score: Absolute evidence the winner must clear.
        min_margin: Factor by which the winner must beat the runner-up.
        min_share: Minimum fraction of voting items backing the winner.
        content_min_margin: Stricter margin for the content tier.

    Returns:
        The accepted detection, or ``None`` to fall back to the caller's
        default.
    """
    lexical = score_headings_lexical(headings)
    detection = _accept(
        lexical,
        "lexical",
        min_score=min_score,
        min_margin=min_margin,
        min_share=min_share,
    )
    if detection is not None:
        return detection

    if embed is not None and headings:
        semantic = score_headings_semantic(headings, embed)
        detection = _accept(
            semantic,
            "semantic",
            min_score=min_score,
            min_margin=min_margin,
            min_share=min_share,
        )
        if detection is not None:
            return detection

    if (
        allow_content_tier
        and embed is not None
        and paragraphs
        and len(headings) <= MAX_HEADINGS_FOR_CONTENT_TIER
    ):
        content = score_content(paragraphs, embed)
        return _accept(
            content, "content", min_score=min_score, min_margin=content_min_margin
        )
    return None

score_content(paragraphs, embed)

Score candidate schemas by body prose against their document profiles.

Weak by construction -- see the module docstring -- and only meaningful for documents with essentially no headings.

Source code in ontocast/tool/chunk/schema_detect.py

def score_content(paragraphs: list[str], embed: TextEmbedder) -> list[SchemaEvidence]:
    """Score candidate schemas by body prose against their document profiles.

    Weak by construction -- see the module docstring -- and only meaningful for
    documents with essentially no headings.
    """
    candidates = [
        sid
        for sid in candidate_schema_ids()
        # News is a measured semantic attractor: any front matter drifts to it.
        if sid != "news"
    ]
    if not paragraphs or not candidates:
        return _rank({sid: 0.0 for sid in candidates}, {}, 0)

    profiles = embed(
        [load_section_label_schema(sid).document_profile.strip() for sid in candidates]
    )
    embedded = embed(paragraphs)
    if profiles is None or embedded is None:
        return _rank({sid: 0.0 for sid in candidates}, {}, 0)

    profile_vectors = [_normalise(vector) for vector in profiles]
    scores = {sid: 0.0 for sid in candidates}
    examples: dict[str, list[str]] = {sid: [] for sid in candidates}
    for paragraph, raw in zip(paragraphs, embedded):
        vector = _normalise(raw)
        ranked = sorted(
            (
                (_cosine(vector, profile_vectors[i]), sid)
                for i, sid in enumerate(candidates)
            ),
            reverse=True,
        )
        (best, winner), (second, _) = ranked[0], ranked[1]
        weight = min(1.0, max(0.0, (best - second) / SEMANTIC_VOTE_TAU))
        if weight <= 0.0:
            continue
        scores[winner] += weight
        examples[winner].append(" ".join(paragraph.split())[:60])
    return _rank(scores, examples, len(paragraphs))

score_headings_lexical(headings)

Score candidate schemas on headings only one of them recognises.

Parameters:

Name	Type	Description	Default
headings	list[str]	Raw heading lines in document order.	required

Returns:

Type	Description
list[SchemaEvidence]	Evidence per candidate schema, strongest first.

Source code in ontocast/tool/chunk/schema_detect.py

def score_headings_lexical(headings: list[str]) -> list[SchemaEvidence]:
    """Score candidate schemas on headings only one of them recognises.

    Args:
        headings: Raw heading lines in document order.

    Returns:
        Evidence per candidate schema, strongest first.
    """
    candidates = candidate_schema_ids()
    schemas = {sid: load_section_label_schema(sid) for sid in candidates}
    voting = _voting_headings(headings)
    scores = {sid: 0.0 for sid in candidates}
    examples: dict[str, list[str]] = {sid: [] for sid in candidates}

    for heading in voting:
        matched = {
            sid
            for sid in candidates
            if resolve_heading_label(heading, schemas[sid]) is not None
        }
        if len(matched) != 1:
            continue
        schema_id = next(iter(matched))
        scores[schema_id] += 1.0
        examples[schema_id].append(normalise_heading_line(heading))
    return _rank(scores, examples, len(voting))

score_headings_semantic(headings, embed)

Score candidate schemas by heading-to-label-name similarity.

Each heading casts a single vote for its nearest schema, scaled by how far that schema beat the runner-up, so a heading that is ambiguous between two schemas contributes almost nothing.

Source code in ontocast/tool/chunk/schema_detect.py

def score_headings_semantic(
    headings: list[str], embed: TextEmbedder
) -> list[SchemaEvidence]:
    """Score candidate schemas by heading-to-label-name similarity.

    Each heading casts a single vote for its nearest schema, scaled by how far
    that schema beat the runner-up, so a heading that is ambiguous between two
    schemas contributes almost nothing.
    """
    candidates = candidate_schema_ids()
    voting = _voting_headings(headings)
    if not voting:
        return _rank({sid: 0.0 for sid in candidates}, {}, 0)

    prototypes: dict[str, list[str]] = {
        sid: _label_prototypes(sid) for sid in candidates
    }
    flat = [text for sid in candidates for text in prototypes[sid]]
    embedded_prototypes = embed(flat)
    embedded_headings = embed(voting)
    if embedded_prototypes is None or embedded_headings is None:
        return _rank({sid: 0.0 for sid in candidates}, {}, 0)

    by_schema: dict[str, list[list[float]]] = {}
    cursor = 0
    for sid in candidates:
        size = len(prototypes[sid])
        by_schema[sid] = [
            _normalise(vector) for vector in embedded_prototypes[cursor : cursor + size]
        ]
        cursor += size

    scores = {sid: 0.0 for sid in candidates}
    examples: dict[str, list[str]] = {sid: [] for sid in candidates}
    for heading, raw in zip(voting, embedded_headings):
        vector = _normalise(raw)
        best_per_schema = [
            (max(_cosine(vector, proto) for proto in by_schema[sid]), sid)
            for sid in candidates
            if by_schema[sid]
        ]
        if len(best_per_schema) < 2:
            continue
        best_per_schema.sort(reverse=True)
        (best, winner), (second, _) = best_per_schema[0], best_per_schema[1]
        weight = min(1.0, max(0.0, (best - second) / SEMANTIC_VOTE_TAU))
        if weight <= 0.0:
            continue
        scores[winner] += weight
        examples[winner].append(normalise_heading_line(heading))
    return _rank(scores, examples, len(voting))