Skip to content

ontocast.tool.vector_store.core

Core contracts and models for ontology vector storage.

GraphAtom

Bases: BasePydanticModel

Embedding-ready ontology entity atom.

Source code in ontocast/tool/vector_store/core.py 
 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
class GraphAtom(BasePydanticModel):
    """Embedding-ready ontology entity atom."""

    atom_id: str = Field(
        description="Deterministic hash identifier for the atom content."
    )
    ontology_iri: str = Field(description="Source ontology IRI.")
    ontology_id: str | None = Field(
        default=None, description="Optional source ontology identifier."
    )
    ontology_hash: str | None = Field(
        default=None, description="Hash/version of the source ontology."
    )
    ontology_version: str | None = Field(
        default=None, description="Semantic version of the source ontology."
    )
    iri: str = Field(description="Focal entity IRI represented by this atom.")
    entity_role: str | None = Field(
        default=None,
        description="Role of focal entity in graph context: resource or predicate.",
    )
    core_representation: str = Field(
        description="High-precision natural language text (labels, types, descriptions)."
    )
    minimal_representation: str = Field(
        default="",
        description=(
            "IRI local name with camelCase/PascalCase split into space-separated terms; "
            "used for BM25 (keyword) indexing."
        ),
    )
    neighborhood_representation: str = Field(
        description="Neighborhood relation text for disambiguation context."
    )
    created_at: datetime = Field(
        default_factory=lambda: datetime.now(timezone.utc),
        description="Atom creation timestamp (UTC).",
    )
    score: float | None = Field(
        default=None,
        description="Optional similarity score populated by vector search.",
    )
    lexical_triggers: list[str] = Field(
        default_factory=list,
        description=(
            "Case-preserved literal tokens (symbols, notations, formula codes) "
            "used by the lexical-trigger retrieval lane for exact text matching."
        ),
    )
    symbol_surfaces: list[str] = Field(
        default_factory=list,
        description=(
            "Case-preserved declared symbol/notation surface forms "
            "(skos:notation, qudt:symbol, qudt:ucumCode). The embedded/BM25 "
            "text is case-folded, so these carry the only case-significant "
            "evidence at merge time — used to demote counterfeit symbol "
            "matches like prose 'meV' retrieving symbol 'MeV'."
        ),
    )

    @field_validator("entity_role", mode="before")
    @classmethod
    def _normalize_entity_role(cls, value: str | None) -> str | None:
        if value is None:
            return None
        return canonicalize_entity_role(str(value))

    @property
    def representation(self) -> str:
        """Combined embedding text view for generic consumers."""
        return combine_embedding_text(self)

representation property

Combined embedding text view for generic consumers.

OntologySearchHit

Bases: BasePydanticModel

Typed retrieval result that separates atom payload from ranking metadata.

Source code in ontocast/tool/vector_store/core.py 
116
117
118
119
120
class OntologySearchHit(BasePydanticModel):
    """Typed retrieval result that separates atom payload from ranking metadata."""

    atom: GraphAtom
    score: float = Field(description="Channel-specific retrieval score.")

OntologySearchHitsByChannel

Bases: BasePydanticModel

Per-query retrieval hits split by vector channel (dense core/neighborhood + optional BM25).

Source code in ontocast/tool/vector_store/core.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
class OntologySearchHitsByChannel(BasePydanticModel):
    """Per-query retrieval hits split by vector channel (dense core/neighborhood + optional BM25)."""

    core_hits: list[OntologySearchHit] = Field(
        default_factory=list,
        description="Top hits from the dense core vector channel.",
    )
    neighborhood_hits: list[OntologySearchHit] = Field(
        default_factory=list,
        description="Top hits from the dense neighborhood vector channel.",
    )
    bm25_hits: list[OntologySearchHit] = Field(
        default_factory=list,
        description="Top hits from the sparse BM25 lane (minimal IRI text).",
    )

VectorStoreManager

Bases: Tool

Abstract interface for vector store implementations.

Source code in ontocast/tool/vector_store/core.py 
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
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
class VectorStoreManager(Tool):
    """Abstract interface for vector store implementations."""

    store_config: VectorStoreConfig = Field(default_factory=VectorStoreConfig)
    embedding: EmbeddingTool | None = Field(default=None, exclude=True)
    sparse_embedding: FastembedBm25SparseTool | None = Field(default=None, exclude=True)

    @abc.abstractmethod
    async def initialize(self) -> None:
        """Prepare schema/collections in the backing vector store."""

    @abc.abstractmethod
    def index_ontology(self, ontology: Ontology) -> int:
        """Index an ontology and return number of indexed atoms."""

    @abc.abstractmethod
    def search_patches(
        self,
        query: str,
        top_k: int | None = None,
        filter_iri: str | None = None,
        filter_version: str | None = None,
        filter_hash: str | None = None,
    ) -> list[GraphAtom]:
        """Search ontology patches by query text (``top_k`` None → store default)."""

    @abc.abstractmethod
    def search_patch_hits(
        self,
        query: str,
        top_k: int | None = None,
        filter_iri: str | None = None,
        filter_version: str | None = None,
        filter_hash: str | None = None,
    ) -> list[OntologySearchHit]:
        """Search ontology atoms and return rank-fused scored hit objects."""

    @abc.abstractmethod
    def search_patch_hits_many(
        self,
        queries: list[str],
        top_k: int | None = None,
        filter_iri: str | None = None,
        filter_version: str | None = None,
        filter_hash: str | None = None,
    ) -> list[OntologySearchHitsByChannel]:
        """Search ontology atoms for many queries with split-channel outputs."""

    @abc.abstractmethod
    async def asearch_patch_hits_many(
        self,
        queries: list[str],
        top_k: int | None = None,
        filter_iri: str | None = None,
        filter_version: str | None = None,
        filter_hash: str | None = None,
    ) -> list[OntologySearchHitsByChannel]:
        """Async variant of :meth:`search_patch_hits_many`."""

    @abc.abstractmethod
    def fetch_vectors(
        self,
        atom_ids: list[str],
    ) -> dict[str, tuple[list[float], list[float]]]:
        """Batch-fetch dense core/neighborhood vectors for MMR."""

    async def afetch_vectors(
        self,
        atom_ids: list[str],
    ) -> dict[str, tuple[list[float], list[float]]]:
        """Async wrapper around :meth:`fetch_vectors`."""
        return await asyncio.to_thread(self.fetch_vectors, atom_ids)

    def fetch_atoms_by_ids(self, atom_ids: list[str]) -> list[GraphAtom]:
        """Batch-fetch atom payloads by ``atom_id`` (for lexical-trigger injection)."""
        raise NotImplementedError(
            f"{type(self).__name__} does not support fetch_atoms_by_ids"
        )

    def match_lexical_triggers(
        self, text: str, *, max_atoms: int | None = None
    ) -> list[GraphAtom]:
        """Match raw text against the lexical-trigger index and return atoms."""
        raise NotImplementedError(
            f"{type(self).__name__} does not support lexical trigger matching"
        )

    @abc.abstractmethod
    def delete_ontology(
        self,
        iri: str,
        version: str | None = None,
        ontology_hash: str | None = None,
    ) -> None:
        """Delete all indexed atoms for a specific ontology IRI."""

    def reindex_ontology(self, ontology: Ontology) -> int:
        """Replace all atoms for a given ontology and return indexed count."""
        self.delete_ontology(ontology.iri)
        return self.index_ontology(ontology)

    def list_indexed_ontology_iris(self) -> set[str]:
        """Return distinct ``ontology_iri`` values present in the ontology store."""
        raise NotImplementedError(
            f"{type(self).__name__} does not support listing indexed ontology IRIs"
        )

    def prune_orphan_ontology_iris(self, keep_iris: set[str]) -> list[str]:
        """Delete indexed atoms whose ``ontology_iri`` is not in ``keep_iris``.

        An empty ``keep_iris`` is refused rather than treated as "everything is
        an orphan". Pruning exists to follow IRI renames, and no rename makes
        every ontology disappear at once -- an empty catalog means the source of
        truth could not be read, and deleting the whole index on that basis is
        unrecoverable. Callers that genuinely want an empty store should call
        :meth:`wipe_store`.

        Returns the orphan IRIs that were deleted (sorted); empty when the
        prune was refused.
        """
        indexed = self.list_indexed_ontology_iris()
        if not keep_iris:
            if indexed:
                logger.warning(
                    "Refusing to prune %d indexed ontology IRI(s) against an empty "
                    "catalog -- this usually means the triple store could not be "
                    "read. Use wipe_store() to clear the index deliberately.",
                    len(indexed),
                )
            return []
        orphans = sorted(indexed - keep_iris)
        for iri in orphans:
            self.delete_ontology(iri)
        return orphans

    def close(self) -> None:
        """Release any backend connection held by this store.

        Default is a no-op: backends that open no long-lived handle (LanceDB
        connects per call) have nothing to release.
        """
        return None

    async def wipe_store(self) -> None:
        """Drop the currently configured ontology/facts collections or tables.

        Call :meth:`initialize` afterwards to recreate empty schema.
        """
        raise NotImplementedError(
            f"{type(self).__name__} does not support wiping the current store"
        )

    def apply_tenancy(
        self,
        tenant: str,
        project: str,
        *,
        sep: str = TENANCY_SEP,
    ) -> None:
        """Switch the active tenant/project partition when supported."""
        if not self.supports_tenancy_partition():
            raise NotImplementedError(
                f"{type(self).__name__} does not isolate data by tenant/project"
            )
        raise NotImplementedError(f"{type(self).__name__} must implement apply_tenancy")

    def supports_tenancy_partition(self) -> bool:
        """True if tenancy hooks isolate data by tenant/project."""
        return False

    async def clean_tenancy(self, tenant: str, project: str) -> None:
        """Drop or empty vector collections derived from ``tenant`` / ``project``."""
        raise NotImplementedError(
            f"{type(self).__name__} does not isolate vectors by tenant/project"
        )

afetch_vectors(atom_ids) async

Async wrapper around :meth:fetch_vectors.

Source code in ontocast/tool/vector_store/core.py 
206
207
208
209
210
211
async def afetch_vectors(
    self,
    atom_ids: list[str],
) -> dict[str, tuple[list[float], list[float]]]:
    """Async wrapper around :meth:`fetch_vectors`."""
    return await asyncio.to_thread(self.fetch_vectors, atom_ids)

apply_tenancy(tenant, project, *, sep=TENANCY_SEP)

Switch the active tenant/project partition when supported.

Source code in ontocast/tool/vector_store/core.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
def apply_tenancy(
    self,
    tenant: str,
    project: str,
    *,
    sep: str = TENANCY_SEP,
) -> None:
    """Switch the active tenant/project partition when supported."""
    if not self.supports_tenancy_partition():
        raise NotImplementedError(
            f"{type(self).__name__} does not isolate data by tenant/project"
        )
    raise NotImplementedError(f"{type(self).__name__} must implement apply_tenancy")

asearch_patch_hits_many(queries, top_k=None, filter_iri=None, filter_version=None, filter_hash=None) abstractmethod async

Async variant of :meth:search_patch_hits_many.

Source code in ontocast/tool/vector_store/core.py 
188
189
190
191
192
193
194
195
196
197
@abc.abstractmethod
async def asearch_patch_hits_many(
    self,
    queries: list[str],
    top_k: int | None = None,
    filter_iri: str | None = None,
    filter_version: str | None = None,
    filter_hash: str | None = None,
) -> list[OntologySearchHitsByChannel]:
    """Async variant of :meth:`search_patch_hits_many`."""

clean_tenancy(tenant, project) async

Drop or empty vector collections derived from tenant / project.

Source code in ontocast/tool/vector_store/core.py 
310
311
312
313
314
async def clean_tenancy(self, tenant: str, project: str) -> None:
    """Drop or empty vector collections derived from ``tenant`` / ``project``."""
    raise NotImplementedError(
        f"{type(self).__name__} does not isolate vectors by tenant/project"
    )

close()

Release any backend connection held by this store.

Default is a no-op: backends that open no long-lived handle (LanceDB connects per call) have nothing to release.

Source code in ontocast/tool/vector_store/core.py 
275
276
277
278
279
280
281
def close(self) -> None:
    """Release any backend connection held by this store.

    Default is a no-op: backends that open no long-lived handle (LanceDB
    connects per call) have nothing to release.
    """
    return None

delete_ontology(iri, version=None, ontology_hash=None) abstractmethod

Delete all indexed atoms for a specific ontology IRI.

Source code in ontocast/tool/vector_store/core.py 
227
228
229
230
231
232
233
234
@abc.abstractmethod
def delete_ontology(
    self,
    iri: str,
    version: str | None = None,
    ontology_hash: str | None = None,
) -> None:
    """Delete all indexed atoms for a specific ontology IRI."""

fetch_atoms_by_ids(atom_ids)

Batch-fetch atom payloads by atom_id (for lexical-trigger injection).

Source code in ontocast/tool/vector_store/core.py 
213
214
215
216
217
def fetch_atoms_by_ids(self, atom_ids: list[str]) -> list[GraphAtom]:
    """Batch-fetch atom payloads by ``atom_id`` (for lexical-trigger injection)."""
    raise NotImplementedError(
        f"{type(self).__name__} does not support fetch_atoms_by_ids"
    )

fetch_vectors(atom_ids) abstractmethod

Batch-fetch dense core/neighborhood vectors for MMR.

Source code in ontocast/tool/vector_store/core.py 
199
200
201
202
203
204
@abc.abstractmethod
def fetch_vectors(
    self,
    atom_ids: list[str],
) -> dict[str, tuple[list[float], list[float]]]:
    """Batch-fetch dense core/neighborhood vectors for MMR."""

index_ontology(ontology) abstractmethod

Index an ontology and return number of indexed atoms.

Source code in ontocast/tool/vector_store/core.py 
151
152
153
@abc.abstractmethod
def index_ontology(self, ontology: Ontology) -> int:
    """Index an ontology and return number of indexed atoms."""

initialize() abstractmethod async

Prepare schema/collections in the backing vector store.

Source code in ontocast/tool/vector_store/core.py 
147
148
149
@abc.abstractmethod
async def initialize(self) -> None:
    """Prepare schema/collections in the backing vector store."""

list_indexed_ontology_iris()

Return distinct ontology_iri values present in the ontology store.

Source code in ontocast/tool/vector_store/core.py 
241
242
243
244
245
def list_indexed_ontology_iris(self) -> set[str]:
    """Return distinct ``ontology_iri`` values present in the ontology store."""
    raise NotImplementedError(
        f"{type(self).__name__} does not support listing indexed ontology IRIs"
    )

match_lexical_triggers(text, *, max_atoms=None)

Match raw text against the lexical-trigger index and return atoms.

Source code in ontocast/tool/vector_store/core.py 
219
220
221
222
223
224
225
def match_lexical_triggers(
    self, text: str, *, max_atoms: int | None = None
) -> list[GraphAtom]:
    """Match raw text against the lexical-trigger index and return atoms."""
    raise NotImplementedError(
        f"{type(self).__name__} does not support lexical trigger matching"
    )

prune_orphan_ontology_iris(keep_iris)

Delete indexed atoms whose ontology_iri is not in keep_iris.

An empty keep_iris is refused rather than treated as "everything is an orphan". Pruning exists to follow IRI renames, and no rename makes every ontology disappear at once -- an empty catalog means the source of truth could not be read, and deleting the whole index on that basis is unrecoverable. Callers that genuinely want an empty store should call :meth:wipe_store.

Returns the orphan IRIs that were deleted (sorted); empty when the prune was refused.

Source code in ontocast/tool/vector_store/core.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
def prune_orphan_ontology_iris(self, keep_iris: set[str]) -> list[str]:
    """Delete indexed atoms whose ``ontology_iri`` is not in ``keep_iris``.

    An empty ``keep_iris`` is refused rather than treated as "everything is
    an orphan". Pruning exists to follow IRI renames, and no rename makes
    every ontology disappear at once -- an empty catalog means the source of
    truth could not be read, and deleting the whole index on that basis is
    unrecoverable. Callers that genuinely want an empty store should call
    :meth:`wipe_store`.

    Returns the orphan IRIs that were deleted (sorted); empty when the
    prune was refused.
    """
    indexed = self.list_indexed_ontology_iris()
    if not keep_iris:
        if indexed:
            logger.warning(
                "Refusing to prune %d indexed ontology IRI(s) against an empty "
                "catalog -- this usually means the triple store could not be "
                "read. Use wipe_store() to clear the index deliberately.",
                len(indexed),
            )
        return []
    orphans = sorted(indexed - keep_iris)
    for iri in orphans:
        self.delete_ontology(iri)
    return orphans

reindex_ontology(ontology)

Replace all atoms for a given ontology and return indexed count.

Source code in ontocast/tool/vector_store/core.py 
236
237
238
239
def reindex_ontology(self, ontology: Ontology) -> int:
    """Replace all atoms for a given ontology and return indexed count."""
    self.delete_ontology(ontology.iri)
    return self.index_ontology(ontology)

search_patch_hits(query, top_k=None, filter_iri=None, filter_version=None, filter_hash=None) abstractmethod

Search ontology atoms and return rank-fused scored hit objects.

Source code in ontocast/tool/vector_store/core.py 
166
167
168
169
170
171
172
173
174
175
@abc.abstractmethod
def search_patch_hits(
    self,
    query: str,
    top_k: int | None = None,
    filter_iri: str | None = None,
    filter_version: str | None = None,
    filter_hash: str | None = None,
) -> list[OntologySearchHit]:
    """Search ontology atoms and return rank-fused scored hit objects."""

search_patch_hits_many(queries, top_k=None, filter_iri=None, filter_version=None, filter_hash=None) abstractmethod

Search ontology atoms for many queries with split-channel outputs.

Source code in ontocast/tool/vector_store/core.py 
177
178
179
180
181
182
183
184
185
186
@abc.abstractmethod
def search_patch_hits_many(
    self,
    queries: list[str],
    top_k: int | None = None,
    filter_iri: str | None = None,
    filter_version: str | None = None,
    filter_hash: str | None = None,
) -> list[OntologySearchHitsByChannel]:
    """Search ontology atoms for many queries with split-channel outputs."""

search_patches(query, top_k=None, filter_iri=None, filter_version=None, filter_hash=None) abstractmethod

Search ontology patches by query text (top_k None → store default).

Source code in ontocast/tool/vector_store/core.py 
155
156
157
158
159
160
161
162
163
164
@abc.abstractmethod
def search_patches(
    self,
    query: str,
    top_k: int | None = None,
    filter_iri: str | None = None,
    filter_version: str | None = None,
    filter_hash: str | None = None,
) -> list[GraphAtom]:
    """Search ontology patches by query text (``top_k`` None → store default)."""

supports_tenancy_partition()

True if tenancy hooks isolate data by tenant/project.

Source code in ontocast/tool/vector_store/core.py 
306
307
308
def supports_tenancy_partition(self) -> bool:
    """True if tenancy hooks isolate data by tenant/project."""
    return False

wipe_store() async

Drop the currently configured ontology/facts collections or tables.

Call :meth:initialize afterwards to recreate empty schema.

Source code in ontocast/tool/vector_store/core.py 
283
284
285
286
287
288
289
290
async def wipe_store(self) -> None:
    """Drop the currently configured ontology/facts collections or tables.

    Call :meth:`initialize` afterwards to recreate empty schema.
    """
    raise NotImplementedError(
        f"{type(self).__name__} does not support wiping the current store"
    )

canonicalize_entity_role(role)

Normalize role labels to vector-store vocabulary.

Source code in ontocast/tool/vector_store/core.py 
29
30
31
32
33
34
35
36
37
38
39
40
def canonicalize_entity_role(role: str | None) -> str | None:
    """Normalize role labels to vector-store vocabulary."""
    if role is None:
        return None
    normalized = role.strip().lower()
    if normalized in VECTOR_ENTITY_ROLES:
        return normalized
    if normalized in {"property", "predicate"}:
        return ROLE_PREDICATE
    if normalized in {"class", "instance", "resource"}:
        return ROLE_RESOURCE
    return None