Skip to content

ontocast.onto.ontology_snapshot

Prompt-facing ontology snapshot: graph view with provenance, no catalog identity.

Assembled from catalog ontologies (O* → S). Not a versioned catalog subject; writeback (U → O*) targets real :class:Ontology instances by namespace ownership.

OntologySnapshot

Bases: BasePydanticModel

Ephemeral multi-source ontology context for LLM prompts.

Holds triples + prefix bindings and assembly provenance. Does not carry catalog iri / ontology_id / lineage — those belong only on :class:~ontocast.onto.ontology.Ontology.

Source code in ontocast/onto/ontology_snapshot.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
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
208
209
210
211
212
class OntologySnapshot(BasePydanticModel):
    """Ephemeral multi-source ontology context for LLM prompts.

    Holds triples + prefix bindings and assembly provenance. Does **not** carry
    catalog ``iri`` / ``ontology_id`` / lineage — those belong only on
    :class:`~ontocast.onto.ontology.Ontology`.
    """

    graph: LLMGraphWire = Field(
        default_factory=RDFGraph,
        description="Prompt ontology triples and namespace bindings.",
    )
    source_iris: list[str] = Field(
        default_factory=list,
        description="Catalog ontology IRIs that contributed to this snapshot.",
    )
    assembly_mode: OntologyAssemblyMode = Field(
        default=OntologyAssemblyMode.SELECTED_SINGLE_ONTOLOGY_LLM,
        description="How this snapshot was assembled.",
    )
    title: str | None = Field(default=None, description="Optional prompt title.")
    description: str | None = Field(
        default=None, description="Optional prompt description."
    )
    content_hash: str = Field(
        default="",
        description=(
            "Content hash of graph. Lazy: empty until refresh_content_hash() "
            "is called — canonical hashing is too expensive for the per-unit "
            "hot path."
        ),
    )

    #: Derived prompt text, keyed by graph identity so a reassigned graph misses.
    #: Populated only through :meth:`prompt_chapter`.
    _prompt_cache: dict[tuple[int, int, str, int | None], str] = PrivateAttr(
        default_factory=dict
    )

    def is_empty(self) -> bool:
        """True when the snapshot graph has no triples."""
        return len(self.graph) == 0

    def refresh_content_hash(self) -> None:
        """Recompute ``content_hash`` from the current graph."""
        self.content_hash = self.graph.hash() if len(self.graph) > 0 else ""

    def invalidate_prompt_cache(self) -> None:
        """Drop memoised prompt text.

        Call this after mutating :attr:`graph` in place. Reassigning ``graph``
        needs no call -- the cache key includes the graph's identity.
        """
        self._prompt_cache.clear()

    def prompt_chapter(self, profile: Any, *, max_triples: int | None = None) -> str:
        """Serialised ontology chapter for prompts, memoised per graph.

        Serialising the ontology is the single most expensive step in building a
        facts prompt, and under a shared document snapshot every unit -- and
        every render attempt within a unit -- would otherwise redo it on an
        identical graph, synchronously, on the event loop.

        The memo is keyed on the graph's identity, length and the profile's wire
        format. It is therefore correct for a snapshot whose graph is replaced,
        and *assumes* the graph is not mutated in place, which is the contract
        this class already documents ("ephemeral" context, read-only in the unit
        loops). Any code that does mutate it must call
        :meth:`invalidate_prompt_cache`.

        Args:
            profile: Graph format profile supplying the serialisation.

        Returns:
            str: The ``# ONTOLOGY`` chapter, including the index appendix.
        """
        from ontocast.prompt.ontology_context import build_ontology_index

        # max_triples is part of the key: the snapshot is shared by reference
        # across the whole fan-out, so without it the first budget seen would be
        # served to every later caller.
        key = (id(self.graph), len(self.graph), str(profile.format), max_triples)
        cached = self._prompt_cache.get(key)
        if cached is not None:
            return cached
        chapter = profile.format_ontology_chapter(
            self.graph,
            suffix=build_ontology_index(self.graph),
            max_triples=max_triples,
        )
        # Bound the memo: a snapshot only ever holds one live graph, so stale
        # entries are strictly dead weight after a reassignment.
        self._prompt_cache.clear()
        self._prompt_cache[key] = chapter
        return chapter

    def domain_prefix_pairs(self) -> list[tuple[str, str]]:
        """Domain prefix/namespace pairs from graph bindings (prompt hygiene)."""
        from ontocast.prompt.ontology_context import (
            extract_domain_prefix_pairs_from_graph,
        )

        return extract_domain_prefix_pairs_from_graph(self.graph)

    def describe_for_prompt(self) -> str:
        """Human-readable multi-source description for ontology-update prompts."""
        pairs = self.domain_prefix_pairs()
        prefix_lines = (
            "\n".join(f"  - `{p}:` <{ns}>" for p, ns in pairs)
            if pairs
            else "  (none declared beyond standard vocabularies)"
        )
        sources = (
            "\n".join(f"  - <{iri}>" for iri in self.source_iris)
            if self.source_iris
            else "  (none)"
        )
        title = self.title or "(untitled snapshot)"
        desc = self.description or ""
        return (
            f"Ontology context: {title}\n"
            f"Assembly mode: {self.assembly_mode.value}\n"
            f"Description: {desc}\n"
            f"Source catalog IRIs:\n{sources}\n"
            f"Domain prefixes:\n{prefix_lines}\n"
        )

    @classmethod
    def empty(
        cls,
        *,
        assembly_mode: OntologyAssemblyMode = OntologyAssemblyMode.SELECTED_VECTOR_SEARCH_ENSEMBLE,
        title: str | None = None,
        description: str | None = None,
    ) -> OntologySnapshot:
        """Build an empty snapshot (no triples, no sources)."""
        return cls(
            graph=RDFGraph(),
            source_iris=[],
            assembly_mode=assembly_mode,
            title=title,
            description=description,
            content_hash="",
        )

    @classmethod
    def from_ontology(
        cls,
        ontology: Ontology,
        *,
        assembly_mode: OntologyAssemblyMode,
        title: str | None = None,
        description: str | None = None,
    ) -> OntologySnapshot:
        """Assemble a snapshot from a single catalog ontology (graph copy)."""
        if ontology.is_null():
            return cls.empty(
                assembly_mode=assembly_mode,
                title=title or "Null ontology",
                description=description or "No catalog ontology selected.",
            )
        graph = ontology.graph.copy()
        source = [ontology.iri] if ontology.iri else []
        return cls(
            graph=graph,
            source_iris=source,
            assembly_mode=assembly_mode,
            title=title or ontology.title,
            description=description or ontology.description,
        )

    @classmethod
    def from_graph(
        cls,
        graph: RDFGraph,
        *,
        source_iris: list[str],
        assembly_mode: OntologyAssemblyMode,
        title: str | None = None,
        description: str | None = None,
        strip_headers: bool = True,
    ) -> OntologySnapshot:
        """Assemble a snapshot from a (possibly multi-source) graph."""
        working = graph.copy()
        if strip_headers:
            Ontology.strip_ontology_header_triples(working)
        return cls(
            graph=working,
            source_iris=list(source_iris),
            assembly_mode=assembly_mode,
            title=title,
            description=description,
        )

describe_for_prompt()

Human-readable multi-source description for ontology-update prompts.

Source code in ontocast/onto/ontology_snapshot.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def describe_for_prompt(self) -> str:
    """Human-readable multi-source description for ontology-update prompts."""
    pairs = self.domain_prefix_pairs()
    prefix_lines = (
        "\n".join(f"  - `{p}:` <{ns}>" for p, ns in pairs)
        if pairs
        else "  (none declared beyond standard vocabularies)"
    )
    sources = (
        "\n".join(f"  - <{iri}>" for iri in self.source_iris)
        if self.source_iris
        else "  (none)"
    )
    title = self.title or "(untitled snapshot)"
    desc = self.description or ""
    return (
        f"Ontology context: {title}\n"
        f"Assembly mode: {self.assembly_mode.value}\n"
        f"Description: {desc}\n"
        f"Source catalog IRIs:\n{sources}\n"
        f"Domain prefixes:\n{prefix_lines}\n"
    )

domain_prefix_pairs()

Domain prefix/namespace pairs from graph bindings (prompt hygiene).

Source code in ontocast/onto/ontology_snapshot.py 
116
117
118
119
120
121
122
def domain_prefix_pairs(self) -> list[tuple[str, str]]:
    """Domain prefix/namespace pairs from graph bindings (prompt hygiene)."""
    from ontocast.prompt.ontology_context import (
        extract_domain_prefix_pairs_from_graph,
    )

    return extract_domain_prefix_pairs_from_graph(self.graph)

empty(*, assembly_mode=OntologyAssemblyMode.SELECTED_VECTOR_SEARCH_ENSEMBLE, title=None, description=None) classmethod

Build an empty snapshot (no triples, no sources).

Source code in ontocast/onto/ontology_snapshot.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
@classmethod
def empty(
    cls,
    *,
    assembly_mode: OntologyAssemblyMode = OntologyAssemblyMode.SELECTED_VECTOR_SEARCH_ENSEMBLE,
    title: str | None = None,
    description: str | None = None,
) -> OntologySnapshot:
    """Build an empty snapshot (no triples, no sources)."""
    return cls(
        graph=RDFGraph(),
        source_iris=[],
        assembly_mode=assembly_mode,
        title=title,
        description=description,
        content_hash="",
    )

from_graph(graph, *, source_iris, assembly_mode, title=None, description=None, strip_headers=True) classmethod

Assemble a snapshot from a (possibly multi-source) graph.

Source code in ontocast/onto/ontology_snapshot.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
@classmethod
def from_graph(
    cls,
    graph: RDFGraph,
    *,
    source_iris: list[str],
    assembly_mode: OntologyAssemblyMode,
    title: str | None = None,
    description: str | None = None,
    strip_headers: bool = True,
) -> OntologySnapshot:
    """Assemble a snapshot from a (possibly multi-source) graph."""
    working = graph.copy()
    if strip_headers:
        Ontology.strip_ontology_header_triples(working)
    return cls(
        graph=working,
        source_iris=list(source_iris),
        assembly_mode=assembly_mode,
        title=title,
        description=description,
    )

from_ontology(ontology, *, assembly_mode, title=None, description=None) classmethod

Assemble a snapshot from a single catalog ontology (graph copy).

Source code in ontocast/onto/ontology_snapshot.py 
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
@classmethod
def from_ontology(
    cls,
    ontology: Ontology,
    *,
    assembly_mode: OntologyAssemblyMode,
    title: str | None = None,
    description: str | None = None,
) -> OntologySnapshot:
    """Assemble a snapshot from a single catalog ontology (graph copy)."""
    if ontology.is_null():
        return cls.empty(
            assembly_mode=assembly_mode,
            title=title or "Null ontology",
            description=description or "No catalog ontology selected.",
        )
    graph = ontology.graph.copy()
    source = [ontology.iri] if ontology.iri else []
    return cls(
        graph=graph,
        source_iris=source,
        assembly_mode=assembly_mode,
        title=title or ontology.title,
        description=description or ontology.description,
    )

invalidate_prompt_cache()

Drop memoised prompt text.

Call this after mutating :attr:graph in place. Reassigning graph needs no call -- the cache key includes the graph's identity.

Source code in ontocast/onto/ontology_snapshot.py 
67
68
69
70
71
72
73
def invalidate_prompt_cache(self) -> None:
    """Drop memoised prompt text.

    Call this after mutating :attr:`graph` in place. Reassigning ``graph``
    needs no call -- the cache key includes the graph's identity.
    """
    self._prompt_cache.clear()

is_empty()

True when the snapshot graph has no triples.

Source code in ontocast/onto/ontology_snapshot.py 
59
60
61
def is_empty(self) -> bool:
    """True when the snapshot graph has no triples."""
    return len(self.graph) == 0

prompt_chapter(profile, *, max_triples=None)

Serialised ontology chapter for prompts, memoised per graph.

Serialising the ontology is the single most expensive step in building a facts prompt, and under a shared document snapshot every unit -- and every render attempt within a unit -- would otherwise redo it on an identical graph, synchronously, on the event loop.

The memo is keyed on the graph's identity, length and the profile's wire format. It is therefore correct for a snapshot whose graph is replaced, and assumes the graph is not mutated in place, which is the contract this class already documents ("ephemeral" context, read-only in the unit loops). Any code that does mutate it must call :meth:invalidate_prompt_cache.

Parameters:

Name Type Description Default
profile Any

Graph format profile supplying the serialisation.

 required

Returns:

Name Type Description
str str

The # ONTOLOGY chapter, including the index appendix.
Source code in ontocast/onto/ontology_snapshot.py 
 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
def prompt_chapter(self, profile: Any, *, max_triples: int | None = None) -> str:
    """Serialised ontology chapter for prompts, memoised per graph.

    Serialising the ontology is the single most expensive step in building a
    facts prompt, and under a shared document snapshot every unit -- and
    every render attempt within a unit -- would otherwise redo it on an
    identical graph, synchronously, on the event loop.

    The memo is keyed on the graph's identity, length and the profile's wire
    format. It is therefore correct for a snapshot whose graph is replaced,
    and *assumes* the graph is not mutated in place, which is the contract
    this class already documents ("ephemeral" context, read-only in the unit
    loops). Any code that does mutate it must call
    :meth:`invalidate_prompt_cache`.

    Args:
        profile: Graph format profile supplying the serialisation.

    Returns:
        str: The ``# ONTOLOGY`` chapter, including the index appendix.
    """
    from ontocast.prompt.ontology_context import build_ontology_index

    # max_triples is part of the key: the snapshot is shared by reference
    # across the whole fan-out, so without it the first budget seen would be
    # served to every later caller.
    key = (id(self.graph), len(self.graph), str(profile.format), max_triples)
    cached = self._prompt_cache.get(key)
    if cached is not None:
        return cached
    chapter = profile.format_ontology_chapter(
        self.graph,
        suffix=build_ontology_index(self.graph),
        max_triples=max_triples,
    )
    # Bound the memo: a snapshot only ever holds one live graph, so stale
    # entries are strictly dead weight after a reassignment.
    self._prompt_cache.clear()
    self._prompt_cache[key] = chapter
    return chapter

refresh_content_hash()

Recompute content_hash from the current graph.

Source code in ontocast/onto/ontology_snapshot.py 
63
64
65
def refresh_content_hash(self) -> None:
    """Recompute ``content_hash`` from the current graph."""
    self.content_hash = self.graph.hash() if len(self.graph) > 0 else ""