ontocast.onto.ontology_condense

Best-effort condensing of an ontology graph before it becomes prompt text.

Only the vector-retrieval mode ever bounded how much ontology reached the LLM. selected_single_ontology (the default) and fixed_single_ontology serialize the whole selected catalog ontology into every prompt, and the facts fan-out serializes the union of every ontology artifact -- all with no cap, no sampling and no truncation. On a large catalog that is the context blow-up; nothing else in the pipeline notices.

This module trims a graph toward a triple budget by dropping the least-load-bearing triples first, in the order established by :data:~ontocast.onto.graph_prune.BFS_PREDICATE_PRIORITY. It is deliberately best-effort: it will never drop labels, types, hierarchy or domain/range to hit a number. A budget that cannot be met without cutting into those is reported as a warning and the graph is passed through oversized, because silently removing the schema the model needed produces a bad extraction that looks like a bad model -- the most expensive failure mode this pipeline has.

Attributes

CLIP_MARKER = '…' module-attribute

CONTRACT_TEXT_PREDICATES = frozenset({SKOS.scopeNote, SKOS.definition}) module-attribute

GLOSS_PREDICATES = BFS_PREDICATE_PRIORITY[-1] module-attribute

LOAD_BEARING_PREDICATES = frozenset().union(*BFS_PREDICATE_PRIORITY[:-1]) module-attribute

NAMING_TEXT_PREDICATES = frozenset({RDFS.label, SKOS.prefLabel, SKOS.altLabel}) module-attribute

PROSE_TEXT_PREDICATES = frozenset({RDFS.comment, SKOS.example, SKOS.note, SKOS.editorialNote, SKOS.historyNote}) module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

CondenseReport

Bases: BaseModel

What condensing did, for telemetry and for explaining a warning.

Source code in ontocast/onto/ontology_condense.py

class CondenseReport(BaseModel):
    """What condensing did, for telemetry and for explaining a warning."""

    triples_before: int = Field(description="Triple count on entry")
    triples_after: int = Field(description="Triple count after condensing")
    max_triples: int | None = Field(
        default=None, description="Budget applied; None means no budget"
    )
    dropped_noise: int = Field(default=0, description="Header/list plumbing removed")
    dropped_structural: int = Field(
        default=0, description="Generic types, stub restrictions, orphan bnodes"
    )
    dropped_glosses: int = Field(
        default=0, description="Comments, definitions, scope notes, alt labels"
    )
    over_budget: bool = Field(
        default=False,
        description="Still above budget after condensing; passed through oversized",
    )
    text_chars_before: int = Field(
        default=0, description="Summed length of capped text literals on entry"
    )
    text_chars_after: int = Field(
        default=0, description="Summed length of capped text literals after capping"
    )
    literals_clipped: int = Field(
        default=0, description="Text literals shortened to a per-role cap"
    )
    text_over_budget: bool = Field(
        default=False,
        description="Still above the total text budget after every tightening stage",
    )

    @property
    def changed(self) -> bool:
        """Whether anything was removed or shortened at all."""
        return bool(self.triples_after != self.triples_before or self.literals_clipped)

    def as_metrics(self) -> dict[str, int | bool | None]:
        """Flat mapping for the retrieval-metrics payload."""
        return {
            "triples_before": self.triples_before,
            "triples_after": self.triples_after,
            "max_triples": self.max_triples,
            "dropped_noise": self.dropped_noise,
            "dropped_structural": self.dropped_structural,
            "dropped_glosses": self.dropped_glosses,
            "over_budget": self.over_budget,
            "text_chars_before": self.text_chars_before,
            "text_chars_after": self.text_chars_after,
            "literals_clipped": self.literals_clipped,
            "text_over_budget": self.text_over_budget,
        }

Attributes

changed property

Whether anything was removed or shortened at all.

dropped_glosses = Field(default=0, description='Comments, definitions, scope notes, alt labels') class-attribute instance-attribute

dropped_noise = Field(default=0, description='Header/list plumbing removed') class-attribute instance-attribute

dropped_structural = Field(default=0, description='Generic types, stub restrictions, orphan bnodes') class-attribute instance-attribute

literals_clipped = Field(default=0, description='Text literals shortened to a per-role cap') class-attribute instance-attribute

max_triples = Field(default=None, description='Budget applied; None means no budget') class-attribute instance-attribute

over_budget = Field(default=False, description='Still above budget after condensing; passed through oversized') class-attribute instance-attribute

text_chars_after = Field(default=0, description='Summed length of capped text literals after capping') class-attribute instance-attribute

text_chars_before = Field(default=0, description='Summed length of capped text literals on entry') class-attribute instance-attribute

text_over_budget = Field(default=False, description='Still above the total text budget after every tightening stage') class-attribute instance-attribute

triples_after = Field(description='Triple count after condensing') class-attribute instance-attribute

triples_before = Field(description='Triple count on entry') class-attribute instance-attribute

Methods:

as_metrics()

Flat mapping for the retrieval-metrics payload.

Source code in ontocast/onto/ontology_condense.py

def as_metrics(self) -> dict[str, int | bool | None]:
    """Flat mapping for the retrieval-metrics payload."""
    return {
        "triples_before": self.triples_before,
        "triples_after": self.triples_after,
        "max_triples": self.max_triples,
        "dropped_noise": self.dropped_noise,
        "dropped_structural": self.dropped_structural,
        "dropped_glosses": self.dropped_glosses,
        "over_budget": self.over_budget,
        "text_chars_before": self.text_chars_before,
        "text_chars_after": self.text_chars_after,
        "literals_clipped": self.literals_clipped,
        "text_over_budget": self.text_over_budget,
    }

TextCaps

Bases: BaseModel

Per-role character caps on the text literals reaching a prompt.

Nothing else in the pipeline bounds a single literal, so chapter size is otherwise proportional to how chatty a catalog's authors were rather than to how many terms it offers. These caps make it proportional to the term count, which the retrieval budget already controls. On a tersely authored catalog they are a no-op; that is the intended shape -- a bound, not a reduction.

Clipping rather than dropping is what keeps a usage contract available at a predictable price: the first sentence of a scope note is the part that says when a term applies. Where a cap fires, the cut is by content rather than by position -- the opening sentence and any clause stating when the term applies, then stop -- so a verbose catalog pays for its terms and not for its prose style. A cap left unset leaves every literal exactly as authored.

Source code in ontocast/onto/ontology_condense.py

class TextCaps(BaseModel):
    """Per-role character caps on the text literals reaching a prompt.

    Nothing else in the pipeline bounds a single literal, so chapter size is
    otherwise proportional to how chatty a catalog's authors were rather than to
    how many terms it offers. These caps make it proportional to the term count,
    which the retrieval budget already controls. On a tersely authored catalog
    they are a no-op; that is the intended shape -- a bound, not a reduction.

    Clipping rather than dropping is what keeps a usage contract available at a
    predictable price: the first sentence of a scope note is the part that says
    when a term applies. Where a cap fires, the cut is by content rather than
    by position -- the opening sentence and any clause stating when the term
    applies, then stop -- so a verbose catalog pays for its terms and not for
    its prose style. A cap left unset leaves every literal exactly as authored.
    """

    naming: int | None = Field(
        default=None,
        ge=1,
        description="Cap on rdfs:label / skos:prefLabel / skos:altLabel.",
    )
    contract: int | None = Field(
        default=None,
        ge=1,
        description="Cap on skos:scopeNote / skos:definition.",
    )
    prose: int | None = Field(
        default=None, ge=1, description="Cap on rdfs:comment and other notes."
    )
    total_budget: int | None = Field(
        default=None,
        ge=1,
        description=(
            "Ceiling on the summed length of all text literals in the chapter. "
            "Backstop for a catalog that defeats the per-role caps by holding "
            "very many short terms."
        ),
    )

    @property
    def active(self) -> bool:
        """Whether any cap is set at all."""
        return any(
            cap is not None
            for cap in (self.naming, self.contract, self.prose, self.total_budget)
        )

    def cap_for(self, predicate: URIRef) -> int | None:
        """The cap governing ``predicate``, or None when it governs no role."""
        if predicate in NAMING_TEXT_PREDICATES:
            return self.naming
        if predicate in CONTRACT_TEXT_PREDICATES:
            return self.contract
        if predicate in PROSE_TEXT_PREDICATES:
            return self.prose
        return None

Attributes

active property

Whether any cap is set at all.

contract = Field(default=None, ge=1, description='Cap on skos:scopeNote / skos:definition.') class-attribute instance-attribute

naming = Field(default=None, ge=1, description='Cap on rdfs:label / skos:prefLabel / skos:altLabel.') class-attribute instance-attribute

prose = Field(default=None, ge=1, description='Cap on rdfs:comment and other notes.') class-attribute instance-attribute

total_budget = Field(default=None, ge=1, description='Ceiling on the summed length of all text literals in the chapter. Backstop for a catalog that defeats the per-role caps by holding very many short terms.') class-attribute instance-attribute

Methods:

cap_for(predicate)

The cap governing predicate, or None when it governs no role.

Source code in ontocast/onto/ontology_condense.py

def cap_for(self, predicate: URIRef) -> int | None:
    """The cap governing ``predicate``, or None when it governs no role."""
    if predicate in NAMING_TEXT_PREDICATES:
        return self.naming
    if predicate in CONTRACT_TEXT_PREDICATES:
        return self.contract
    if predicate in PROSE_TEXT_PREDICATES:
        return self.prose
    return None

Functions:

clip_text(text, cap)

Clip text to cap characters, marking the cut.

The retained text is at most cap characters; :data:CLIP_MARKER is appended on top of it, so a clipped literal reads as clipped. A cap of None, or text already within it, is returned unchanged -- byte-identical, so a disabled cap cannot perturb a prompt or its cache key.

Where the text has more than one sentence the cut is content-aware (see :func:_contract_head) and may retain well under cap; a single sentence longer than the cap falls back to a word boundary, which is the only cut available inside it.

Parameters:

Name	Type	Description	Default
text	str	Literal text to bound.	required
cap	int | None	Maximum retained characters, or None to leave text alone.	required

Returns:

Type	Description
str	Either text itself or a clipped copy ending in the marker.

Source code in ontocast/onto/ontology_condense.py

def clip_text(text: str, cap: int | None) -> str:
    """Clip ``text`` to ``cap`` characters, marking the cut.

    The retained text is at most ``cap`` characters; :data:`CLIP_MARKER` is
    appended on top of it, so a clipped literal reads as clipped. A ``cap`` of
    None, or text already within it, is returned unchanged -- byte-identical, so
    a disabled cap cannot perturb a prompt or its cache key.

    Where the text has more than one sentence the cut is content-aware (see
    :func:`_contract_head`) and may retain well under ``cap``; a single
    sentence longer than the cap falls back to a word boundary, which is the
    only cut available inside it.

    Args:
        text: Literal text to bound.
        cap: Maximum retained characters, or None to leave ``text`` alone.

    Returns:
        Either ``text`` itself or a clipped copy ending in the marker.
    """
    if cap is None or len(text) <= cap:
        return text
    head = _contract_head(text, cap)
    if head:
        return head + CLIP_MARKER
    head = text[:cap].rsplit(" ", 1)[0].rstrip()
    if not head:
        head = text[:cap].rstrip()
    return head + CLIP_MARKER

condense_graph_for_prompt(graph, max_triples, text_caps=None)

Trim graph toward max_triples, dropping the least useful triples first.

Passes are applied in increasing order of harm, stopping as soon as the graph fits: header/list noise, then structural scaffolding, then glosses. Structure that lets the model name and place a term is never dropped.

Text literals are bounded first and unconditionally: the triple budget is a count and says nothing about how long a single rdfs:comment may be, so a graph well under it can still carry an unbounded chapter. text_caps makes chapter size a function of term count rather than of prose volume.

Parameters:

Name	Type	Description	Default
graph	RDFGraph	Ontology graph destined for a prompt. Not mutated.	required
max_triples	int | None	Triple budget, or None to disable condensing entirely.	required
text_caps	TextCaps | None	Per-role character caps on text literals, or None/all-unset to leave every literal as authored.	None

Returns:

Type	Description
RDFGraph	The condensed graph (or graph itself when nothing was done) and a
CondenseReport

Source code in ontocast/onto/ontology_condense.py

def condense_graph_for_prompt(
    graph: RDFGraph,
    max_triples: int | None,
    text_caps: TextCaps | None = None,
) -> tuple[RDFGraph, CondenseReport]:
    """Trim ``graph`` toward ``max_triples``, dropping the least useful triples first.

    Passes are applied in increasing order of harm, stopping as soon as the graph
    fits: header/list noise, then structural scaffolding, then glosses. Structure
    that lets the model name and place a term is never dropped.

    Text literals are bounded first and unconditionally: the triple budget is a
    count and says nothing about how long a single ``rdfs:comment`` may be, so a
    graph well under it can still carry an unbounded chapter. ``text_caps`` makes
    chapter size a function of term count rather than of prose volume.

    Args:
        graph: Ontology graph destined for a prompt. Not mutated.
        max_triples: Triple budget, or ``None`` to disable condensing entirely.
        text_caps: Per-role character caps on text literals, or ``None``/all-unset
            to leave every literal as authored.

    Returns:
        The condensed graph (or ``graph`` itself when nothing was done) and a
        :class:`CondenseReport` describing what happened.
    """
    before = len(graph)
    report = CondenseReport(
        triples_before=before, triples_after=before, max_triples=max_triples
    )
    capping = text_caps is not None and text_caps.active
    fits = max_triples is None or before <= max_triples
    if fits and not capping:
        return graph, report

    working = graph.copy()
    if capping:
        assert text_caps is not None
        _cap_text_literals(working, text_caps, report)
    if fits:
        report.triples_after = len(working)
        return working, report

    report.dropped_noise = _drop_predicates(working, NOISY_EXPANSION_PREDICATES)
    if len(working) > max_triples:
        structural_before = len(working)
        strip_redundant_generic_types(working)
        prune_degenerate_restriction_bnodes(working)
        prune_orphaned_bnode_subjects(working)
        report.dropped_structural = structural_before - len(working)

    if len(working) > max_triples:
        report.dropped_glosses = _drop_predicates(working, GLOSS_PREDICATES)

    report.triples_after = len(working)
    report.over_budget = len(working) > max_triples

    if report.over_budget:
        logger.warning(
            "Ontology context still exceeds the prompt budget after condensing "
            "(%d > %d triples; was %d). Passing it through rather than dropping "
            "labels, types, hierarchy or domain/range, which the model needs to "
            "use the schema at all. Reduce the catalog, split it, or switch to "
            "ONTOLOGY_CONTEXT_MODE=selected_vector_search_ontology, which "
            "retrieves only the relevant part.",
            report.triples_after,
            max_triples,
            before,
        )
    else:
        logger.info(
            "Condensed ontology context %d -> %d triples for a %d budget "
            "(noise=%d, structural=%d, glosses=%d).",
            before,
            report.triples_after,
            max_triples,
            report.dropped_noise,
            report.dropped_structural,
            report.dropped_glosses,
        )

    return working, report

split_sentences(text)

Split text into sentences, keeping each one's terminator.

Abbreviation-aware only to the extent that a period inside e.g. or a single capital initial must not end a sentence; anything more would be guessing at a catalog's writing style.

Source code in ontocast/onto/ontology_condense.py

def split_sentences(text: str) -> list[str]:
    """Split ``text`` into sentences, keeping each one's terminator.

    Abbreviation-aware only to the extent that a period inside ``e.g.`` or a
    single capital initial must not end a sentence; anything more would be
    guessing at a catalog's writing style.
    """
    sentences: list[str] = []
    start = 0
    for match in _SENTENCE_BOUNDARY_RE.finditer(text):
        head = text[start : match.start()]
        if _ABBREVIATION_TAIL_RE.search(head):
            continue
        stripped = head.strip()
        if stripped:
            sentences.append(stripped)
        start = match.end()
    tail = text[start:].strip()
    if tail:
        sentences.append(tail)
    return sentences