ontocast.tool.vector_store

Vector store package for ontology patch retrieval.

The Qdrant and LanceDB managers are exported lazily: importing either pulls its backend SDK, and neither ships in OntoCast's base install. Everything else -- the abstract manager, the atom model, the embedding tools, and the in-memory backend -- is dependency-light and imported eagerly.

EmbeddingContractMismatchError

Bases: ValueError

Embedding vectors or store metadata disagree with the active embedding config.

Source code in ontocast/tool/vector_store/util.py

class EmbeddingContractMismatchError(ValueError):
    """Embedding vectors or store metadata disagree with the active embedding config."""

EmbeddingTool

Bases: Tool

Base embedding tool with provider-specific implementations.

Source code in ontocast/tool/vector_store/embedding.py

class EmbeddingTool(Tool):
    """Base embedding tool with provider-specific implementations."""

    config: EmbeddingConfig = Field(default_factory=EmbeddingConfig)

    @abc.abstractmethod
    def _embed_raw(self, texts: list[str]) -> list[list[float]]:
        """Return vectors for all given texts, prefixes already applied."""

    def embed(self, texts: list[str]) -> list[list[float]]:
        """Return vectors for all given texts as *documents*.

        Serialisation, where it is needed, belongs to whatever owns the model —
        the shared encoder for local checkpoints, nothing for remote providers.
        """
        if not texts:
            return []
        return self._embed_raw(self._apply(self.config.document_prefix, texts))

    def embed_query(self, texts: list[str]) -> list[list[float]]:
        """Return vectors for all given texts as *queries*.

        Asymmetric retrieval models are trained with distinct query and document
        instructions and lose accuracy when both sides are encoded identically. With
        empty prefixes — the default, suiting a symmetric paraphrase model — this is
        exactly :meth:`embed`.
        """
        if not texts:
            return []
        return self._embed_raw(self._apply(self.config.query_prefix, texts))

    @staticmethod
    def _apply(prefix: str, texts: list[str]) -> list[str]:
        return texts if not prefix else [f"{prefix}{text}" for text in texts]

    def embed_one(self, text: str) -> list[float]:
        """Return a vector for one query text."""
        vectors = self.embed_query([text])
        if not vectors:
            raise ValueError("Embedding provider returned no vectors for query text")
        return vectors[0]

    @classmethod
    def create(cls, config: EmbeddingConfig) -> "EmbeddingTool":
        """Factory for provider-specific embedding tools."""
        if config.provider == EmbeddingProvider.HUGGINGFACE:
            return HuggingFaceEmbeddingTool(config=config)
        if config.provider == EmbeddingProvider.OPENAI:
            return OpenAIEmbeddingTool(config=config)
        if config.provider == EmbeddingProvider.OLLAMA:
            return OllamaEmbeddingTool(config=config)
        raise ValueError(f"Unsupported embedding provider: {config.provider}")

create(config) classmethod

Factory for provider-specific embedding tools.

Source code in ontocast/tool/vector_store/embedding.py

@classmethod
def create(cls, config: EmbeddingConfig) -> "EmbeddingTool":
    """Factory for provider-specific embedding tools."""
    if config.provider == EmbeddingProvider.HUGGINGFACE:
        return HuggingFaceEmbeddingTool(config=config)
    if config.provider == EmbeddingProvider.OPENAI:
        return OpenAIEmbeddingTool(config=config)
    if config.provider == EmbeddingProvider.OLLAMA:
        return OllamaEmbeddingTool(config=config)
    raise ValueError(f"Unsupported embedding provider: {config.provider}")

embed(texts)

Return vectors for all given texts as documents.

Serialisation, where it is needed, belongs to whatever owns the model — the shared encoder for local checkpoints, nothing for remote providers.

Source code in ontocast/tool/vector_store/embedding.py

def embed(self, texts: list[str]) -> list[list[float]]:
    """Return vectors for all given texts as *documents*.

    Serialisation, where it is needed, belongs to whatever owns the model —
    the shared encoder for local checkpoints, nothing for remote providers.
    """
    if not texts:
        return []
    return self._embed_raw(self._apply(self.config.document_prefix, texts))

embed_one(text)

Return a vector for one query text.

Source code in ontocast/tool/vector_store/embedding.py

def embed_one(self, text: str) -> list[float]:
    """Return a vector for one query text."""
    vectors = self.embed_query([text])
    if not vectors:
        raise ValueError("Embedding provider returned no vectors for query text")
    return vectors[0]

embed_query(texts)

Return vectors for all given texts as queries.

Asymmetric retrieval models are trained with distinct query and document instructions and lose accuracy when both sides are encoded identically. With empty prefixes — the default, suiting a symmetric paraphrase model — this is exactly :meth:embed.

Source code in ontocast/tool/vector_store/embedding.py

def embed_query(self, texts: list[str]) -> list[list[float]]:
    """Return vectors for all given texts as *queries*.

    Asymmetric retrieval models are trained with distinct query and document
    instructions and lose accuracy when both sides are encoded identically. With
    empty prefixes — the default, suiting a symmetric paraphrase model — this is
    exactly :meth:`embed`.
    """
    if not texts:
        return []
    return self._embed_raw(self._apply(self.config.query_prefix, texts))

FastembedBm25SparseTool

Bases: Tool

BM25-style sparse text embeddings via fastembed (Qdrant-compatible).

Source code in ontocast/tool/vector_store/embedding.py

class FastembedBm25SparseTool(Tool):
    """BM25-style sparse text embeddings via fastembed (Qdrant-compatible)."""

    config: EmbeddingConfig = Field(default_factory=EmbeddingConfig)
    _embedder: Any = PrivateAttr(default=None)

    def _get_embedder(self) -> Any:
        if self._embedder is not None:
            return self._embedder
        fastembed_mod = require("fastembed", feature="BM25 sparse embeddings")
        sparse_cls = getattr(fastembed_mod, "SparseTextEmbedding", None)
        if sparse_cls is None:
            raise ImportError("fastembed.SparseTextEmbedding is not available")
        self._embedder = sparse_cls(model_name=self.config.bm25_model_name)
        return self._embedder

    def embed_sparse(self, texts: list[str]) -> list[SparseVector]:
        """Return Qdrant sparse vectors for indexing all given texts (thread-safe)."""
        if not texts:
            return []
        with _SPARSE_EMBED_LOCK:
            return self._embed_sparse_unlocked(texts)

    def embed_sparse_query(self, texts: list[str]) -> list[SparseVector]:
        """Return Qdrant sparse vectors for *querying* with all given texts.

        BM25 is asymmetric: documents carry term-frequency saturation weights, queries
        carry flat per-term weights, and the IDF factor is applied by the store. Encoding
        queries with the document encoder instead squares the term-frequency weighting and
        drops the query/document distinction entirely.
        """
        if not texts:
            return []
        with _SPARSE_EMBED_LOCK:
            return self._embed_sparse_unlocked(texts, query=True)

    def _embed_sparse_unlocked(
        self, texts: list[str], *, query: bool = False
    ) -> list[SparseVector]:
        model = self._get_embedder()
        encode = model.query_embed if query else model.embed
        out: list[SparseVector] = []
        for sparse_emb in encode(texts):
            payload = sparse_emb.as_object()
            indices_raw = payload["indices"]
            values_raw = payload["values"]
            indices_list = indices_raw.tolist()
            values_list = values_raw.tolist()
            out.append(
                SparseVector(
                    indices=[int(i) for i in indices_list],
                    values=[float(v) for v in values_list],
                )
            )
        if len(out) != len(texts):
            raise ValueError("BM25 embedder returned mismatched sparse vector count")
        return out

    def embed_one_sparse(self, text: str) -> SparseVector:
        vectors = self.embed_sparse_query([text])
        if not vectors:
            raise ValueError("BM25 embedder returned no sparse vector for query text")
        return vectors[0]

embed_sparse(texts)

Return Qdrant sparse vectors for indexing all given texts (thread-safe).

Source code in ontocast/tool/vector_store/embedding.py

def embed_sparse(self, texts: list[str]) -> list[SparseVector]:
    """Return Qdrant sparse vectors for indexing all given texts (thread-safe)."""
    if not texts:
        return []
    with _SPARSE_EMBED_LOCK:
        return self._embed_sparse_unlocked(texts)

embed_sparse_query(texts)

Return Qdrant sparse vectors for querying with all given texts.

BM25 is asymmetric: documents carry term-frequency saturation weights, queries carry flat per-term weights, and the IDF factor is applied by the store. Encoding queries with the document encoder instead squares the term-frequency weighting and drops the query/document distinction entirely.

Source code in ontocast/tool/vector_store/embedding.py

def embed_sparse_query(self, texts: list[str]) -> list[SparseVector]:
    """Return Qdrant sparse vectors for *querying* with all given texts.

    BM25 is asymmetric: documents carry term-frequency saturation weights, queries
    carry flat per-term weights, and the IDF factor is applied by the store. Encoding
    queries with the document encoder instead squares the term-frequency weighting and
    drops the query/document distinction entirely.
    """
    if not texts:
        return []
    with _SPARSE_EMBED_LOCK:
        return self._embed_sparse_unlocked(texts, query=True)

GraphAtom

Bases: BasePydanticModel

Embedding-ready ontology entity atom.

Source code in ontocast/tool/vector_store/core.py

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.

GraphAtomizer

Bases: Tool

Extract natural-language atoms around graph focal entities.

Two defaults narrow what an ontology contributes, both restorable:

• Focal IRIs in common W3C and DC vocabulary namespaces are skipped (see module-level exclusions); embed_standard_vocab_iris=True embeds them.
• An IRI is atomized only when this graph describes it — a subject-position triple or a label. index_undescribed_iris=True restores atomizing every URIRef, object-position references included.

Facts sources are restricted to facts_namespace only; neither narrowing applies to them.

Source code in ontocast/tool/vector_store/atomizer.py

class GraphAtomizer(Tool):
    """Extract natural-language atoms around graph focal entities.

    Two defaults narrow what an ontology contributes, both restorable:

    * Focal IRIs in common W3C and DC vocabulary namespaces are skipped (see
      module-level exclusions); ``embed_standard_vocab_iris=True`` embeds them.
    * An IRI is atomized only when this graph *describes* it — a subject-position triple
      or a label. ``index_undescribed_iris=True`` restores atomizing every URIRef,
      object-position references included.

    Facts sources are restricted to ``facts_namespace`` only; neither narrowing applies
    to them.
    """

    embed_standard_vocab_iris: bool = Field(
        default=False,
        description="If True, do not exclude standard vocabulary namespace IRIs as focal entities.",
    )
    extra_excluded_namespace_prefixes: list[str] = Field(
        default_factory=list,
        description="Additional IRI prefixes excluded from focal entities (ontology sources).",
    )
    index_undescribed_iris: bool = Field(
        default=False,
        description=(
            "If True, atomize every IRI in the graph, including ones appearing only in "
            "object or predicate position. Default False: an ontology mints an atom "
            "only for terms it describes (a subject-position triple, or a label). A "
            "referenced IRI has no local text, so its atom is its mangled local name -- "
            "'a0e0l2i0m1h0t 3d0' for a QUDT dimension vector -- and such strings embed "
            "near the corpus centroid, making them hubs that rank against every query. "
            "Measured on the 8-module matsci catalog: 247 of 690 atoms (36%) were "
            "undescribed references, and dimension vectors alone took 51 of 140 dense "
            "retrieval slots on one document, crowding four ontologies out entirely. "
            "Referenced IRIs are still reachable -- induced-subgraph expansion walks "
            "into them from seeds; they just stop being seeds themselves. Changing this "
            "changes which atoms exist and requires a reindex."
        ),
    )
    minimal_representation_label_limit: int = Field(
        default=5,
        ge=0,
        description=(
            "Maximum declared surface forms (label/prefLabel/title/altLabel) folded "
            "into the sparse BM25 representation. A vocabulary may declare more "
            "aliases than this -- symbol aliases in particular sort last and are the "
            "first to be dropped -- so raising it widens what the sparse lane can "
            "match. Changing it changes stored vectors and requires a reindex."
        ),
    )
    label_predicates: list[str] = Field(
        default_factory=lambda: list(DEFAULT_LABEL_PREDICATES),
        description=(
            "Predicate IRIs whose literal objects are treated as declared "
            "labels, in descending priority. Changing this changes stored "
            "vectors and requires a reindex."
        ),
    )
    symbol_predicates: list[str] = Field(
        default_factory=lambda: list(DEFAULT_SYMBOL_PREDICATES),
        description=(
            "Predicate IRIs whose literal objects are treated as symbols/"
            "notations, collected against their own budget so they are not "
            "crowded out by multilingual labels. Should agree with "
            "VECTOR_STORE_INDUCED_SUBGRAPH_SYMBOL_PREDICATES, which controls "
            "the retrieval half of the same contract. Changing this changes "
            "stored vectors and requires a reindex."
        ),
    )
    lexical_trigger_enabled: bool = Field(
        default=True,
        description="Collect case-preserved lexical triggers on each atom.",
    )
    lexical_trigger_predicates: list[str] = Field(
        default_factory=lambda: [
            "http://www.w3.org/2004/02/skos/core#notation",
            "http://qudt.org/schema/qudt/symbol",
            "http://qudt.org/schema/qudt/ucumCode",
        ],
        description="Predicate IRIs whose literal objects become lexical triggers.",
    )
    lexical_trigger_heuristic_enabled: bool = Field(
        default=True,
        description=(
            "Promote code-shaped labels/altLabels when no notation is declared."
        ),
    )
    lexical_trigger_min_len: int = Field(default=2, ge=1)
    lexical_trigger_max_len: int = Field(default=24, ge=1)
    lexical_trigger_heuristic_max_per_entity: int = Field(default=2, ge=0)

    def _label_predicate_refs(self) -> list[URIRef]:
        """Configured label predicates as rdflib terms."""
        return [URIRef(iri) for iri in self.label_predicates]

    def _symbol_predicate_refs(self) -> list[URIRef]:
        """Configured symbol/notation predicates as rdflib terms."""
        return [URIRef(iri) for iri in self.symbol_predicates]

    class _VectorizationSource(Protocol):
        graph: RDFGraph
        iri: str
        ontology_id: str | None
        hash: str | None
        version: str | None

    def _merged_excluded_vocab_prefixes(self) -> frozenset[str]:
        extra = (
            _normalize_vocab_exclude_prefix(p)
            for p in self.extra_excluded_namespace_prefixes
        )
        return frozenset(STANDARD_VOCABULARY_NAMESPACE_PREFIXES).union(
            frozenset(p for p in extra if p)
        )

    def atomize(self, source: _VectorizationSource, depth: int = 1) -> list[GraphAtom]:
        """Generate deterministic atoms from local graph neighborhoods."""
        if depth < 0:
            raise ValueError("Atomizer depth must be >= 0")

        raw_graph = source.graph
        embedding_graph = strip_provenance_triples_for_embedding(raw_graph)
        focal_namespace = source.facts_namespace if isinstance(source, Facts) else None
        is_ontology_source = not isinstance(source, Facts)
        excluded_vocab: frozenset[str] | None = None
        if is_ontology_source and not self.embed_standard_vocab_iris:
            excluded_vocab = self._merged_excluded_vocab_prefixes()
        entities = self._collect_focal_entities(
            graph=embedding_graph,
            focal_namespace=focal_namespace,
            excluded_vocab_prefixes=excluded_vocab,
            # Facts are already confined to ``facts_namespace``, where every individual
            # is a subject; the describes-only rule targets ontology cross-references.
            require_description=is_ontology_source and not self.index_undescribed_iris,
        )
        predicate_uris = {p for (_, p, _) in embedding_graph if isinstance(p, URIRef)}
        declared_property_uris = {
            subject
            for property_type in _PROPERTY_TYPE_IRIS
            for subject in embedding_graph.subjects(RDF.type, property_type)
            if isinstance(subject, URIRef)
        }
        generated_at = datetime.now(timezone.utc)

        atoms_by_id: dict[str, GraphAtom] = {}
        seen_payload_keys: set[tuple[str, str, str, str | None, str | None]] = set()
        for entity in entities:
            role = role_from_declaration(
                is_declared_property=entity in declared_property_uris,
                is_predicate=entity in predicate_uris,
            )
            patch_graph = self._build_neighborhood_graph(
                graph=embedding_graph, root=entity, depth=depth
            )
            if len(patch_graph) == 0:
                continue

            core_representation = self._build_core_representation(
                entity=entity, graph=patch_graph, role=role
            )
            minimal_representation = self._build_minimal_representation(
                entity, embedding_graph
            )
            lexical_triggers = self._build_lexical_triggers(entity, embedding_graph)
            symbol_surfaces = self._collect_raw_literals(
                embedding_graph, entity, self._symbol_predicate_refs(), max_items=8
            )
            neighborhood_variants = self._build_neighborhood_variants(
                entity=entity, graph=patch_graph, entity_role=role
            )
            if not neighborhood_variants:
                neighborhood_variants = [""]
            # Keep first occurrence while removing repeated textual variants.
            neighborhood_variants = list(dict.fromkeys(neighborhood_variants))

            for variant_index, neighborhood_representation in enumerate(
                neighborhood_variants
            ):
                payload_key = (
                    source.iri,
                    str(entity),
                    core_representation,
                    neighborhood_representation,
                    role,
                )
                if payload_key in seen_payload_keys:
                    continue
                seen_payload_keys.add(payload_key)
                atom_key = (
                    f"{source.iri}|{source.hash}|{source.version}|{entity}|"
                    f"{variant_index}|{core_representation}|{neighborhood_representation}"
                )
                atom_id = render_text_hash(atom_key, digits=None)
                if atom_id in atoms_by_id:
                    continue
                atoms_by_id[atom_id] = GraphAtom(
                    atom_id=atom_id,
                    ontology_iri=source.iri,
                    ontology_id=source.ontology_id,
                    ontology_hash=source.hash,
                    ontology_version=source.version,
                    iri=str(entity),
                    entity_role=role,
                    core_representation=core_representation,
                    minimal_representation=minimal_representation,
                    neighborhood_representation=neighborhood_representation,
                    lexical_triggers=lexical_triggers,
                    symbol_surfaces=symbol_surfaces,
                    created_at=generated_at,
                )
        return list(atoms_by_id.values())

    def _build_neighborhood_graph(
        self, graph: RDFGraph, root: URIRef, depth: int
    ) -> RDFGraph:
        """Build a local subgraph by bounded BFS over URI/BNode neighbors."""
        result = RDFGraph()
        self._copy_namespaces(graph=graph, result=result)
        queue: deque[tuple[Node, int]] = deque([(root, 0)])
        visited: set[Node] = {root}

        while queue:
            node, node_depth = queue.popleft()

            for triple in graph.triples((node, None, None)):
                result.add(triple)
                _, _, obj = triple
                if node_depth < depth and isinstance(obj, (URIRef, BNode)):
                    if obj not in visited:
                        visited.add(obj)
                        queue.append((obj, node_depth + 1))

            for triple in graph.triples((None, None, node)):
                result.add(triple)
                subj, _, _ = triple
                if node_depth < depth and isinstance(subj, (URIRef, BNode)):
                    if subj not in visited:
                        visited.add(subj)
                        queue.append((subj, node_depth + 1))

        return result

    def _copy_namespaces(self, graph: RDFGraph, result: RDFGraph) -> None:
        """Preserve namespace bindings in derived patch graphs."""
        for prefix, namespace in graph.namespaces():
            if prefix:
                result.bind(prefix, namespace)

    def _describes(self, graph: RDFGraph, entity: URIRef) -> bool:
        """True when this graph says something *about* ``entity``, not merely with it.

        Subject-position triples are the primary evidence. A label alone also counts:
        a vocabulary may name a term it otherwise only references, and that name is
        exactly what retrieval needs.
        """
        for _ in graph.triples((entity, None, None)):
            return True
        return any(
            next(graph.objects(entity, predicate), None) is not None
            for predicate in self._label_predicate_refs()
        )

    def _collect_focal_entities(
        self,
        graph: RDFGraph,
        focal_namespace: str | None = None,
        excluded_vocab_prefixes: frozenset[str] | None = None,
        require_description: bool = False,
    ) -> list[URIRef]:
        ns_prefix = focal_namespace.rstrip("/") if focal_namespace is not None else None
        entities: set[URIRef] = set()
        for subj, pred, obj in graph:
            for term in (subj, pred, obj):
                if isinstance(term, URIRef):
                    if ns_prefix is None or str(term).startswith(ns_prefix):
                        entities.add(term)

        if ns_prefix is not None:
            entities = {e for e in entities if str(e).startswith(ns_prefix)}

        if excluded_vocab_prefixes:
            entities = {
                e
                for e in entities
                if not any(str(e).startswith(p) for p in excluded_vocab_prefixes)
            }

        if require_description:
            entities = {e for e in entities if self._describes(graph, e)}

        return sorted(entities, key=lambda entity: str(entity))

    def _parent_resource_phrase(self, graph: RDFGraph, parent: URIRef) -> str:
        """Local name plus optional label gloss when it adds information."""
        base = self._normalize_uri(parent)
        literals = self._collect_surface_forms(graph, parent, 1)
        if not literals:
            return base
        gloss = literals[0]
        if gloss == base:
            return base
        return f'{base} (also described as "{gloss}")'

    def _subclass_parent_index(self, graph: RDFGraph) -> dict[URIRef, set[URIRef]]:
        parent_to_children: dict[URIRef, set[URIRef]] = defaultdict(
            lambda: set[URIRef]()
        )
        for child, _, parent in graph.triples((None, RDFS.subClassOf, None)):
            if isinstance(child, URIRef) and isinstance(parent, URIRef):
                parent_to_children[parent].add(child)
        return parent_to_children

    def _incident_triples(
        self, graph: RDFGraph, entity: URIRef
    ) -> list[tuple[Node, Node, Node]]:
        raw: list[tuple[Node, Node, Node]] = []
        seen: set[tuple[Node, Node, Node]] = set()
        for triple in graph.triples((entity, None, None)):
            if triple not in seen:
                seen.add(triple)
                raw.append(triple)
        for triple in graph.triples((None, None, entity)):
            if triple not in seen:
                seen.add(triple)
                raw.append(triple)
        for triple in graph.triples((None, entity, None)):
            if triple not in seen:
                seen.add(triple)
                raw.append(triple)
        return stable_sorted_triples(raw)

    def _is_generic_type(self, type_uri: URIRef) -> bool:
        return type_uri in _GENERIC_TYPE_IRIS

    def _is_annotation_predicate(self, pred: URIRef) -> bool:
        return pred in _ANNOTATION_PREDICATES

    def _collect_domain_labels(
        self, entity: URIRef, graph: RDFGraph, max_items: int
    ) -> list[str]:
        labels: list[str] = []
        seen: set[str] = set()
        for _, _, o in sorted(
            graph.triples((entity, RDFS.domain, None)), key=lambda t: str(t[2])
        ):
            if not isinstance(o, URIRef):
                continue
            text = self._normalize_uri(o)
            if text not in seen:
                seen.add(text)
                labels.append(text)
            if len(labels) >= max_items:
                break
        return labels

    def _collect_range_labels(
        self, entity: URIRef, graph: RDFGraph, max_items: int
    ) -> list[str]:
        labels: list[str] = []
        seen: set[str] = set()
        for _, _, o in sorted(
            graph.triples((entity, RDFS.range, None)), key=lambda t: str(t[2])
        ):
            if not isinstance(o, URIRef):
                continue
            text = self._normalize_uri(o)
            if text not in seen:
                seen.add(text)
                labels.append(text)
            if len(labels) >= max_items:
                break
        return labels

    def _append_inverse_of_clues_for_property(
        self, prop_ref: URIRef, graph: RDFGraph, clues: list[str]
    ) -> None:
        for _, _, inv in sorted(
            graph.triples((prop_ref, OWL.inverseOf, None)),
            key=lambda tr: str(tr[2]),
        ):
            if isinstance(inv, URIRef):
                inv_phrase = self._parent_resource_phrase(graph, inv)
                clues.append(
                    f"{self._normalize_uri(prop_ref)} is the reverse of {inv_phrase}"
                )

    def _append_property_domain_range_clues_for_subject_resource(
        self,
        entity: URIRef,
        graph: RDFGraph,
        clues: list[str],
        *,
        max_properties: int,
        endpoint_label_cap: int,
    ) -> None:
        props_with_domain = sorted(
            {
                p
                for p, _, _ in graph.triples((None, RDFS.domain, entity))
                if isinstance(p, URIRef)
            },
            key=str,
        )[:max_properties]
        for prop in props_with_domain:
            prop_verb = self._normalize_uri(prop)  # bare verb for SPO
            ranges = self._collect_range_labels(
                prop, graph, max_items=endpoint_label_cap
            )
            for r_label in ranges or ["something"]:
                clues.append(f"it {prop_verb} {r_label}")
            self._append_inverse_of_clues_for_property(prop, graph, clues)

        props_with_range = sorted(
            {
                p
                for p, _, _ in graph.triples((None, RDFS.range, entity))
                if isinstance(p, URIRef)
            },
            key=str,
        )[:max_properties]
        for prop in props_with_range:
            prop_verb = self._normalize_uri(prop)  # bare verb for SPO
            domains = self._collect_domain_labels(
                prop, graph, max_items=endpoint_label_cap
            )
            for d_label in domains or ["something"]:
                clues.append(f"{d_label} {prop_verb} it")
            self._append_inverse_of_clues_for_property(prop, graph, clues)

    def _build_minimal_representation(
        self, entity: URIRef, graph: RDFGraph | None = None
    ) -> str:
        """Keyword-oriented text for the sparse BM25 lane.

        The IRI local name (camelCase/PascalCase split, see ``normalize_uri_local_name``)
        plus any human labels. Lexical match is the strongest available signal for
        technical vocabulary that appears near-verbatim in source text, but an IRI local
        name is often an opaque identifier — Wikidata-derived ``Q36834`` carries no
        tokens at all, and the term is only findable through its ``rdfs:label``.
        Descriptions are deliberately excluded: they would dominate term frequency
        without naming the entity.
        """
        local_name = normalize_uri_local_name(entity)
        if graph is None:
            return local_name
        labels = self._collect_surface_forms(
            graph,
            entity,
            self.minimal_representation_label_limit,
            lead_with_symbol=True,
        )
        parts = [local_name, *labels]
        seen: set[str] = set()
        tokens: list[str] = []
        for part in parts:
            normalized = normalize_text(part)
            if normalized and normalized not in seen:
                seen.add(normalized)
                tokens.append(normalized)
        return " ".join(tokens)

    def _build_core_representation(
        self, entity: URIRef, graph: RDFGraph, role: str
    ) -> str:
        labels = self._collect_surface_forms(graph, entity, 5)
        descriptions = self._collect_literals(
            graph,
            entity,
            [RDFS.comment, DCTERMS.description, SKOS.definition, SKOS.scopeNote],
            2,
        )
        informative_types = []
        for _, _, obj in sorted(
            graph.triples((entity, RDF.type, None)), key=lambda t: str(t[2])
        ):
            if not isinstance(obj, URIRef) or self._is_generic_type(obj):
                continue
            informative_types.append(self._normalize_uri(obj))
            if len(informative_types) >= 3:
                break

        entity_name = labels[0] if labels else self._normalize_uri(entity)
        parts: list[str] = [entity_name]

        if informative_types:
            parts[0] += f" ({', '.join(informative_types)})"

        if len(labels) > 1:
            parts.append(f"Also known as {', '.join(labels[1:])}")

        parts.extend(descriptions)

        domains = self._collect_domain_labels(entity, graph, max_items=4)
        ranges = self._collect_range_labels(entity, graph, max_items=4)
        if domains:
            parts.append(f"Applies to: {', '.join(domains)}")
        if ranges:
            parts.append(f"Values restricted to: {', '.join(ranges)}")

        return ". ".join(parts)

    def _collect_structural_clues(
        self, entity: URIRef, graph: RDFGraph, entity_role: str
    ) -> list[str]:
        clues: list[str] = []
        focal_is_property = entity_role == "predicate"

        for _, _, t in sorted(
            graph.triples((entity, RDF.type, None)), key=lambda tr: str(tr[2])
        ):
            if isinstance(t, URIRef) and not self._is_generic_type(t):
                clues.append(f"it is a {self._normalize_uri(t)}")

        if not focal_is_property:
            parent_to_children = self._subclass_parent_index(graph)

            for _, _, parent in sorted(
                graph.triples((entity, RDFS.subClassOf, None)),
                key=lambda tr: str(tr[2]),
            ):
                if isinstance(parent, URIRef):
                    clues.append(
                        f"it is a kind of {self._parent_resource_phrase(graph, parent)}"
                    )

            for child, _, _ in sorted(
                graph.triples((None, RDFS.subClassOf, entity)),
                key=lambda tr: str(tr[0]),
            ):
                if isinstance(child, URIRef):
                    clues.append(f"{self._normalize_uri(child)} is a kind of it")

            parents = [
                o
                for _, _, o in graph.triples((entity, RDFS.subClassOf, None))
                if isinstance(o, URIRef)
            ]
            for par in sorted(set(parents), key=str):
                siblings = sorted(
                    (
                        sib
                        for sib in parent_to_children.get(par, set[URIRef]())
                        if sib != entity
                    ),
                    key=str,
                )
                for sib in siblings[:6]:
                    clues.append(
                        f"{self._normalize_uri(sib)} is also a kind of "
                        f"{self._parent_resource_phrase(graph, par)}"
                    )

            self._append_property_domain_range_clues_for_subject_resource(
                entity=entity,
                graph=graph,
                clues=clues,
                max_properties=8,
                endpoint_label_cap=3,
            )

        for _, _, other in sorted(
            graph.triples((entity, OWL.equivalentClass, None)),
            key=lambda tr: str(tr[2]),
        ):
            if isinstance(other, URIRef):
                clues.append(f"it means the same as {self._normalize_uri(other)}")

        for _, _, other in sorted(
            graph.triples((entity, OWL.disjointWith, None)),
            key=lambda tr: str(tr[2]),
        ):
            if isinstance(other, URIRef):
                clues.append(f"it never overlaps with {self._normalize_uri(other)}")

        for _, _, other in sorted(
            graph.triples((entity, OWL.equivalentProperty, None)),
            key=lambda tr: str(tr[2]),
        ):
            if isinstance(other, URIRef):
                clues.append(f"it means the same as {self._normalize_uri(other)}")

        if focal_is_property:
            for _, _, parent in sorted(
                graph.triples((entity, RDFS.subPropertyOf, None)),
                key=lambda tr: str(tr[2]),
            ):
                if isinstance(parent, URIRef):
                    clues.append(
                        f"it is a narrower form of {self._parent_resource_phrase(graph, parent)}"
                    )

            for child, _, _ in sorted(
                graph.triples((None, RDFS.subPropertyOf, entity)),
                key=lambda tr: str(tr[0]),
            ):
                if isinstance(child, URIRef):
                    clues.append(
                        f"{self._normalize_uri(child)} is a narrower form of it"
                    )

            for _, _, inv in sorted(
                graph.triples((entity, OWL.inverseOf, None)),
                key=lambda tr: str(tr[2]),
            ):
                if isinstance(inv, URIRef):
                    clues.append(f"it is the reverse of {self._normalize_uri(inv)}")

            for d in self._collect_domain_labels(entity, graph, max_items=3):
                clues.append(f"it applies to {d}")
            for r in self._collect_range_labels(entity, graph, max_items=3):
                clues.append(f"it yields {r}")

        for subj, pred, obj in self._incident_triples(graph, entity):
            if not isinstance(pred, URIRef):
                continue
            if self._is_annotation_predicate(pred):
                continue
            if pred in _STRUCTURAL_PREDICATES:
                continue

            pred_phrase = self._normalize_uri(pred)

            if pred == entity:
                if isinstance(subj, URIRef) and isinstance(obj, URIRef):
                    clues.append(
                        f"{self._normalize_uri(subj)} it {self._normalize_uri(obj)}"
                    )
                continue

            if subj == entity:
                if not isinstance(obj, URIRef):
                    continue
                clues.append(f"it {pred_phrase} {self._normalize_uri(obj)}")
            elif obj == entity:
                if not isinstance(subj, URIRef):
                    continue
                clues.append(f"{self._normalize_uri(subj)} {pred_phrase} it")

        return sorted(set(clues))

    def _build_neighborhood_variants(
        self, entity: URIRef, graph: RDFGraph, entity_role: str
    ) -> list[str]:
        clues = self._collect_structural_clues(
            entity=entity, graph=graph, entity_role=entity_role
        )
        if not clues:
            return []
        # Temporary simplification: emit a single deterministic neighborhood view.
        return [". ".join(clues)]

    def _collect_literals(
        self, graph: RDFGraph, subject: URIRef, predicates: list[URIRef], max_items: int
    ) -> list[str]:
        """Collect literal surface forms, deterministically, in predicate priority order.

        ``graph.triples`` yields in unspecified order, so truncating its output at
        ``max_items`` picked an arbitrary subset of a term's labels: a term declaring
        more aliases than the cap allows would embed differently between runs over
        identical input, which makes retrieval measurements irreproducible. Values are
        sorted within each predicate before truncation; predicate order is still
        honoured, keeping ``rdfs:label`` ahead of ``skos:altLabel``.

        Args:
            graph: Graph to read literals from.
            subject: Subject whose literals are collected.
            predicates: Predicates to read, in descending priority.
            max_items: Maximum number of distinct values to return.

        Returns:
            list[str]: Normalized literal values, at most ``max_items``.
        """
        values: list[str] = []
        seen: set[str] = set()
        for predicate in predicates:
            candidates = sorted(
                {
                    (_language_rank(obj), normalized)
                    for _, _, obj in graph.triples((subject, predicate, None))
                    if isinstance(obj, Literal)
                    and (normalized := self._normalize_string(str(obj)))
                }
            )
            for _, normalized in candidates:
                if normalized in seen:
                    continue
                values.append(normalized)
                seen.add(normalized)
                if len(values) >= max_items:
                    return values
        return values

    def _collect_surface_forms(
        self,
        graph: RDFGraph,
        subject: URIRef,
        max_items: int,
        *,
        lead_with_symbol: bool = False,
    ) -> list[str]:
        """Declared labels plus QUDT symbols, with symbols guaranteed a slot.

        ``_collect_literals`` honours predicate priority, so appending the symbol
        predicates to the label list would let a term that declares many labels crowd
        the symbols out entirely — and QUDT units routinely declare one label per
        language. The two families are therefore collected against separate budgets and
        merged, so a unit stays findable by the symbol a reader actually types.

        Args:
            graph: Graph to read literals from.
            subject: Entity whose surface forms are collected.
            max_items: Maximum number of distinct values to return.
            lead_with_symbol: Put symbols first, for the sparse lexical lane. When
                ``False`` the primary label leads so the entity keeps a readable name.

        Returns:
            list[str]: Normalized surface forms, at most ``max_items``.
        """
        labels = self._collect_literals(
            graph, subject, self._label_predicate_refs(), max_items
        )
        symbols = self._collect_literals(
            graph, subject, self._symbol_predicate_refs(), max_items
        )
        if not symbols:
            return labels[:max_items]
        if lead_with_symbol:
            ordered = [*symbols, *labels]
        else:
            ordered = [*labels[:1], *symbols, *labels[1:]]

        merged: list[str] = []
        seen: set[str] = set()
        for value in ordered:
            if value in seen:
                continue
            seen.add(value)
            merged.append(value)
            if len(merged) >= max_items:
                break
        return merged

    def _resolved_lexical_trigger_predicates(self) -> list[URIRef]:
        return [URIRef(iri) for iri in self.lexical_trigger_predicates if iri.strip()]

    def _collect_raw_literals(
        self,
        graph: RDFGraph,
        subject: URIRef,
        predicates: list[URIRef],
        max_items: int,
    ) -> list[str]:
        """Collect literal values preserving original case (for lexical triggers)."""
        values: list[str] = []
        seen: set[str] = set()
        for predicate in predicates:
            candidates = sorted(
                {
                    (_language_rank(obj), str(obj).strip())
                    for _, _, obj in graph.triples((subject, predicate, None))
                    if isinstance(obj, Literal) and str(obj).strip()
                }
            )
            for _, raw in candidates:
                if raw in seen:
                    continue
                values.append(raw)
                seen.add(raw)
                if len(values) >= max_items:
                    return values
        return values

    def _build_lexical_triggers(self, entity: URIRef, graph: RDFGraph) -> list[str]:
        if not self.lexical_trigger_enabled:
            return []
        predicate_iris = self._resolved_lexical_trigger_predicates()
        declared = self._collect_raw_literals(
            graph, entity, predicate_iris, max_items=16
        )
        if declared:
            return dedupe_preserve_case(declared)

        if not self.lexical_trigger_heuristic_enabled:
            return []

        heuristic: list[str] = []
        for candidate in self._collect_raw_literals(
            graph,
            entity,
            [RDFS.label, SKOS.altLabel],
            max_items=self.lexical_trigger_heuristic_max_per_entity + 4,
        ):
            if looks_like_lexical_code(
                candidate,
                min_len=self.lexical_trigger_min_len,
                max_len=self.lexical_trigger_max_len,
            ):
                heuristic.append(candidate)
            if len(heuristic) >= self.lexical_trigger_heuristic_max_per_entity:
                break
        return dedupe_preserve_case(heuristic)

    def _normalize_uri(self, uri: URIRef) -> str:
        return normalize_uri_local_name(uri)

    def _normalize_string(self, text: str) -> str:
        return normalize_text(text)

atomize(source, depth=1)

Generate deterministic atoms from local graph neighborhoods.

Source code in ontocast/tool/vector_store/atomizer.py

def atomize(self, source: _VectorizationSource, depth: int = 1) -> list[GraphAtom]:
    """Generate deterministic atoms from local graph neighborhoods."""
    if depth < 0:
        raise ValueError("Atomizer depth must be >= 0")

    raw_graph = source.graph
    embedding_graph = strip_provenance_triples_for_embedding(raw_graph)
    focal_namespace = source.facts_namespace if isinstance(source, Facts) else None
    is_ontology_source = not isinstance(source, Facts)
    excluded_vocab: frozenset[str] | None = None
    if is_ontology_source and not self.embed_standard_vocab_iris:
        excluded_vocab = self._merged_excluded_vocab_prefixes()
    entities = self._collect_focal_entities(
        graph=embedding_graph,
        focal_namespace=focal_namespace,
        excluded_vocab_prefixes=excluded_vocab,
        # Facts are already confined to ``facts_namespace``, where every individual
        # is a subject; the describes-only rule targets ontology cross-references.
        require_description=is_ontology_source and not self.index_undescribed_iris,
    )
    predicate_uris = {p for (_, p, _) in embedding_graph if isinstance(p, URIRef)}
    declared_property_uris = {
        subject
        for property_type in _PROPERTY_TYPE_IRIS
        for subject in embedding_graph.subjects(RDF.type, property_type)
        if isinstance(subject, URIRef)
    }
    generated_at = datetime.now(timezone.utc)

    atoms_by_id: dict[str, GraphAtom] = {}
    seen_payload_keys: set[tuple[str, str, str, str | None, str | None]] = set()
    for entity in entities:
        role = role_from_declaration(
            is_declared_property=entity in declared_property_uris,
            is_predicate=entity in predicate_uris,
        )
        patch_graph = self._build_neighborhood_graph(
            graph=embedding_graph, root=entity, depth=depth
        )
        if len(patch_graph) == 0:
            continue

        core_representation = self._build_core_representation(
            entity=entity, graph=patch_graph, role=role
        )
        minimal_representation = self._build_minimal_representation(
            entity, embedding_graph
        )
        lexical_triggers = self._build_lexical_triggers(entity, embedding_graph)
        symbol_surfaces = self._collect_raw_literals(
            embedding_graph, entity, self._symbol_predicate_refs(), max_items=8
        )
        neighborhood_variants = self._build_neighborhood_variants(
            entity=entity, graph=patch_graph, entity_role=role
        )
        if not neighborhood_variants:
            neighborhood_variants = [""]
        # Keep first occurrence while removing repeated textual variants.
        neighborhood_variants = list(dict.fromkeys(neighborhood_variants))

        for variant_index, neighborhood_representation in enumerate(
            neighborhood_variants
        ):
            payload_key = (
                source.iri,
                str(entity),
                core_representation,
                neighborhood_representation,
                role,
            )
            if payload_key in seen_payload_keys:
                continue
            seen_payload_keys.add(payload_key)
            atom_key = (
                f"{source.iri}|{source.hash}|{source.version}|{entity}|"
                f"{variant_index}|{core_representation}|{neighborhood_representation}"
            )
            atom_id = render_text_hash(atom_key, digits=None)
            if atom_id in atoms_by_id:
                continue
            atoms_by_id[atom_id] = GraphAtom(
                atom_id=atom_id,
                ontology_iri=source.iri,
                ontology_id=source.ontology_id,
                ontology_hash=source.hash,
                ontology_version=source.version,
                iri=str(entity),
                entity_role=role,
                core_representation=core_representation,
                minimal_representation=minimal_representation,
                neighborhood_representation=neighborhood_representation,
                lexical_triggers=lexical_triggers,
                symbol_surfaces=symbol_surfaces,
                created_at=generated_at,
            )
    return list(atoms_by_id.values())

HuggingFaceEmbeddingTool

Bases: EmbeddingTool

Local HuggingFace/SentenceTransformer embeddings.

Source code in ontocast/tool/vector_store/embedding.py

class HuggingFaceEmbeddingTool(EmbeddingTool):
    """Local HuggingFace/SentenceTransformer embeddings."""

    _embedder: SharedEncoder | None = PrivateAttr(default=None)

    def _get_embedder(self) -> SharedEncoder:
        if self._embedder is not None:
            return self._embedder
        # Shared process-wide with entity clustering and semantic chunking, which
        # default to the same or a configurable checkpoint. The handle owns the
        # lock, so every one of those consumers is serialised on the same model
        # without any of them having to know about the others.
        self._embedder = get_shared_encoder(
            self.config.model_name,
            feature=(
                "Local HuggingFace embeddings. For a light install, set "
                "EMBEDDING_PROVIDER=openai or =ollama to embed via an API instead"
            ),
        )
        return self._embedder

    def _embed_raw(self, texts: list[str]) -> list[list[float]]:
        vectors = self._get_embedder().encode(
            texts, convert_to_numpy=True, show_progress_bar=len(texts) > 100
        )
        return [vector.tolist() for vector in vectors]

OllamaEmbeddingTool

Bases: _LangChainEmbeddingTool

Ollama embeddings using either LangChain or direct API fallback.

Source code in ontocast/tool/vector_store/embedding.py

class OllamaEmbeddingTool(_LangChainEmbeddingTool):
    """Ollama embeddings using either LangChain or direct API fallback."""

    def _build_embedder(self) -> Embeddings:
        OllamaEmbeddings = require(
            "langchain_ollama.embeddings", feature="Ollama embeddings"
        ).OllamaEmbeddings
        return OllamaEmbeddings(
            model=self.config.model_name,
            base_url=self.config.base_url,
        )

    def _embed_raw(self, texts: list[str]) -> list[list[float]]:
        try:
            return super()._embed_raw(texts)
        except Exception as exc:
            # Log the real cause: a bad base URL, an auth failure and an absent
            # langchain integration all reach the fallback identically, and if
            # the HTTP path then fails too the user is shown an httpx error
            # unrelated to what actually went wrong.
            logger.debug("Ollama langchain embedding failed, using HTTP: %s", exc)
            return self._embed_via_http(texts)

    def _embed_via_http(self, texts: list[str]) -> list[list[float]]:
        base_url = self.config.base_url or "http://localhost:11434"
        endpoint = f"{base_url.rstrip('/')}/api/embeddings"
        vectors: list[list[float]] = []
        with httpx.Client(timeout=30.0) as client:
            for text in texts:
                response = client.post(
                    endpoint,
                    json={"model": self.config.model_name, "prompt": text},
                )
                response.raise_for_status()
                payload = response.json()
                vector = payload.get("embedding")
                if not isinstance(vector, list):
                    raise ValueError(
                        "Ollama embedding response missing 'embedding' vector"
                    )
                vectors.append(vector)
        return vectors

OntologyPatchRetriever

Bases: Tool

Combines vector retrieval into one composite ontology graph.

Source code in ontocast/tool/vector_store/patch_retriever.py

class OntologyPatchRetriever(Tool):
    """Combines vector retrieval into one composite ontology graph."""

    vector_store: VectorStoreManager = Field(exclude=True)
    sparql_tool: Any | None = Field(default=None, exclude=True)
    # Typed ``Any`` for the same reason as ``sparql_tool``: OntologyManager holds a
    # back-reference to this class, so a concrete annotation would be a cycle.
    ontology_manager: Any | None = Field(default=None, exclude=True)
    patch: PatchRetrievalConfig = Field(
        default_factory=PatchRetrievalConfig,
        exclude=True,
    )
    _last_retrieval_metrics: dict[str, Any] = PrivateAttr(default_factory=dict)
    _surface_index: CatalogSurfaceIndex | None = PrivateAttr(default=None)
    # Whole-module graphs for the small-module closure, keyed by ontology IRI.
    # The catalog is stable for the life of a run, and every content unit hits
    # the same handful of modules — refetching per unit multiplies catalog
    # reads by the unit count for no new information. ``None`` caches a miss.
    _small_module_cache: dict[str, Ontology | None] = PrivateAttr(default_factory=dict)
    # Tenancy the cache was filled under. The retriever outlives a tenancy
    # switch, and serving one tenant's modules to another would be a leak.
    _small_module_cache_scope: str = PrivateAttr(default="")

    @property
    def last_retrieval_metrics(self) -> dict[str, Any]:
        return self._last_retrieval_metrics

    def _match_query_unit_signals(self, trigger_source: str) -> dict[str, str]:
        """Match number-adjacent unit tokens against catalog surface forms.

        Additive, outside the semantic atom budget (precedent: the lexical
        trigger lane). Returns ``{entity_iri: ontology_iri}``; empty when the
        lane is disabled or nothing matches.
        """
        if not self.vector_store.store_config.query_unit_signals_enabled:
            return {}
        manager = self.ontology_manager
        if manager is None or not trigger_source:
            return {}
        tokens = number_adjacent_tokens(trigger_source)
        if not tokens:
            return {}
        if self._surface_index is None:
            # Symbol/notation predicates come from configuration rather than
            # being compiled into query_signals; built lazily because the
            # store config is not available at PrivateAttr default time.
            self._surface_index = CatalogSurfaceIndex(
                symbol_predicates=[
                    URIRef(iri)
                    for iri in (
                        self.vector_store.store_config.induced_subgraph_symbol_predicates
                    )
                ]
            )
        matched = self._surface_index.match(tokens, manager.ontologies)
        if matched:
            logger.info(
                "Query unit signals matched %d entity(ies) from tokens %s",
                len(matched),
                sorted(tokens),
            )
        return matched

    @staticmethod
    def _schema_axiom_graph(
        merged_context: tuple[RDFGraph, dict[str, str]] | None,
        catalog: list[Ontology] | None,
    ) -> RDFGraph | None:
        """Pick the graph to read ``rdfs:domain``/``rdfs:range`` axioms from.

        Whichever of the two expansion paths materialized the ontologies wins;
        neither being available means the induced-subgraph call is fetching on
        its own and there is nothing local to close over.
        """
        if merged_context is not None:
            return merged_context[0]
        if catalog:
            combined = RDFGraph()
            for ontology in catalog:
                combined += ontology.graph
            return combined
        return None

    async def _apply_small_module_closure(
        self, graph: RDFGraph, hit_ontology_iris: list[str]
    ) -> None:
        """Merge whole small modules into the snapshot (header-stripped).

        A vocabulary small enough to fit entirely (e.g. a qualified-quantity
        module of ~20 terms) is included wholesale once any of its atoms is
        admitted: partial inclusion of a tiny module is what pushes the
        renderer to improvise near-miss property names.
        """
        closure_max = self.patch.small_module_closure_max_triples
        if closure_max <= 0:
            return
        modules = await self._asmall_module_candidates(hit_ontology_iris)
        closed: list[str] = []
        for onto_iri, ontology in modules:
            if len(ontology.graph) > closure_max:
                continue
            module_graph = Ontology.strip_ontology_header_triples(ontology.graph.copy())
            _drop_module_contribution(graph, module_graph)
            graph += module_graph
            for prefix, namespace_uri in ontology.graph.namespaces():
                graph.bind(prefix, namespace_uri)
            closed.append(onto_iri)
        if closed:
            self._last_retrieval_metrics["module_closure_iris"] = closed

    async def _asmall_module_candidates(
        self, hit_ontology_iris: list[str]
    ) -> list[tuple[str, Ontology]]:
        """Resolve hit ontologies to full graphs, manager first, store second.

        The in-memory manager is empty in every deployment that keeps its
        catalog in a triple store and fetches per query — which is the normal
        server configuration, and where this closure silently did nothing.
        """
        store_config = getattr(self.vector_store, "store_config", None)
        scope = str(getattr(store_config, "ontology_table", "") or "")
        if scope != self._small_module_cache_scope:
            self._small_module_cache.clear()
            self._small_module_cache_scope = scope

        wanted = sorted(set(hit_ontology_iris))
        resolved: list[tuple[str, Ontology]] = []
        missing: list[str] = []
        manager = self.ontology_manager
        for onto_iri in wanted:
            if onto_iri in self._small_module_cache:
                cached = self._small_module_cache[onto_iri]
                if cached is not None:
                    resolved.append((onto_iri, cached))
                continue
            ontology = (
                manager.get_freshest_terminal_ontology_by_iri(onto_iri)
                if manager is not None
                else None
            )
            if ontology is None or ontology.is_null():
                missing.append(onto_iri)
            else:
                self._small_module_cache[onto_iri] = ontology
                resolved.append((onto_iri, ontology))

        store = self.sparql_tool.triple_store_manager if self.sparql_tool else None
        if missing and store is not None:
            try:
                fetched = await store.afetch_ontologies_by_iri(missing)
            except Exception as exc:
                # Do NOT cache on this path. A None entry is a permanent
                # negative (see the miss-caching note above), so memoizing a
                # transient store error would silently strip the small-module
                # closure from every later unit in the process.
                logger.warning(
                    "Small-module closure catalog fetch failed (not cached, "
                    "will retry on the next unit): %s",
                    exc,
                )
                return sorted(resolved, key=lambda item: item[0])
            by_iri = {o.iri: o for o in fetched if o.iri and not o.is_null()}
            for onto_iri in missing:
                found = by_iri.get(onto_iri)
                self._small_module_cache[onto_iri] = found
                if found is not None:
                    resolved.append((onto_iri, found))
        return sorted(resolved, key=lambda item: item[0])

    async def _acandidate_context(
        self,
        *,
        entity_uris: list[str],
        ontology_iris: list[str],
        ontology_version_filters: dict[str, set[str]] | None,
        ontology_hash_filters: dict[str, set[str]] | None,
        depth: int,
    ) -> tuple[RDFGraph, dict[str, str]]:
        """Build the working graph from a CONSTRUCT instead of merging catalogs.

        Version and hash filters are applied to the *headers*, so the CONSTRUCT is
        restricted to exactly the named graphs the merge path would have selected.

        Prefix bindings cannot come from a CONSTRUCT, so they are rebuilt from the
        catalog's author-prefix table; standard vocabulary prefixes are bound
        downstream by :func:`_bind_common_vocab_prefixes` as on the merge path.

        Returns:
            tuple: ``(candidate_graph, prefix_map)``.
        """
        manager = self.ontology_manager
        assert manager is not None and self.sparql_tool is not None
        store = self.sparql_tool.triple_store_manager
        headers = select_relevant_ontologies(
            dedupe_terminal_ontologies(await manager.aget_catalog_headers()),
            ontology_iris,
            ontology_version_filters,
            ontology_hash_filters,
        )
        if not headers:
            return RDFGraph(), {}

        graph_irefs = _sparql_irefs([header.graph_uri for header in headers])
        seed_irefs = _sparql_irefs(entity_uris)
        if not graph_irefs or not seed_irefs:
            return RDFGraph(), {}

        candidate = RDFGraph()
        for chunk in _chunked(seed_irefs, _MAX_VALUES_TERMS):
            partial = await store.aconstruct(
                build_candidate_subgraph_query(chunk, graph_irefs, depth=depth)
            )
            candidate += partial

        prefix_map: dict[str, str] = {}
        for header in headers:
            namespace = str(header.namespace)
            prefix = manager.author_prefix_for_namespace(namespace)
            if prefix:
                prefix_map[prefix] = namespace
        prefix_map = filter_overbroad_namespace_map(prefix_map)
        for prefix, namespace in prefix_map.items():
            candidate.bind(prefix, Namespace(namespace))
        # Mirror the merge path: author @prefix names persisted as sh:declare
        # triples (pulled by the candidate CONSTRUCT's header branch) win over
        # stem-derived recovery, exactly as ontology_from_named_graph binds them
        # for merged catalog graphs.
        declared = candidate.bind_declared_prefixes()
        known_before_declared = set(prefix_map.values())
        for namespace, prefix in declared.items():
            if namespace not in known_before_declared:
                prefix_map[prefix] = namespace
        # Graphs served from a triple store carry no author @prefix bindings, so
        # stem-derived prefixes fill any remaining gap (see
        # ontology_from_named_graph). Recover the same implicit stems here so
        # both context paths advertise identical namespaces.
        candidate.bind_implicit_namespaces()
        known_namespaces = set(prefix_map.values())
        for prefix, namespace_uri in candidate.namespaces():
            ns = str(namespace_uri)
            if (
                not prefix
                or ns in known_namespaces
                or ns in RDFLIB_DEFAULT_NAMESPACE_URIS
            ):
                continue
            prefix_map[prefix] = ns
        return candidate, prefix_map

    async def _aresolve_merged_context(
        self,
        *,
        entity_uris: list[str],
        ontology_iris: list[str],
        catalog: list[Ontology] | None,
        ontology_version_filters: dict[str, set[str]] | None,
        ontology_hash_filters: dict[str, set[str]] | None,
        depth: int,
        candidate_pushdown: bool,
    ) -> tuple[RDFGraph, dict[str, str]] | None:
        """Resolve the merged ontology context through the catalog, or ``None``.

        Returning ``None`` leaves the induced-subgraph call on its own fetch path,
        which is what happens when no catalog is registered or a read fails.
        ``catalog`` being set means the reference-expansion fallback already
        materialized everything, so there is nothing left to save here.

        Args:
            ontology_iris: Ontology IRIs surviving reference expansion.
            catalog: Ontologies already materialized by the fallback path, if any.
            ontology_version_filters: Allowed versions per ontology IRI.
            ontology_hash_filters: Allowed hashes per ontology IRI.

        Returns:
            tuple | None: ``(merged_graph, prefix_map)``, or ``None`` to fall back.
        """
        manager = self.ontology_manager
        if manager is None or catalog is not None:
            return None
        store = self.sparql_tool.triple_store_manager if self.sparql_tool else None
        use_pushdown = (
            candidate_pushdown
            and store is not None
            and store.supports_sparql_construct()
        )
        try:
            if use_pushdown:
                merged = await self._acandidate_context(
                    entity_uris=entity_uris,
                    ontology_iris=ontology_iris,
                    ontology_version_filters=ontology_version_filters,
                    ontology_hash_filters=ontology_hash_filters,
                    depth=depth,
                )
                mode = "sparql_candidate"
            else:
                selected = select_relevant_ontologies(
                    await manager.aget_ontologies_by_iri(ontology_iris),
                    ontology_iris,
                    ontology_version_filters,
                    ontology_hash_filters,
                )
                merged = await manager.aget_merged_graph(selected)
                mode = "merged_catalog"
        except Exception as exc:
            logger.warning(
                "Catalog context via OntologyManager failed (%s); "
                "falling back to a direct triple-store read",
                exc,
            )
            return None
        self._last_retrieval_metrics.update(manager.catalog_cache_stats())
        self._last_retrieval_metrics["catalog_context_mode"] = mode
        self._last_retrieval_metrics["catalog_context_triples"] = len(merged[0])
        return merged

    def _effective_top_k(self, top_k: int | None) -> int:
        if top_k is not None:
            return top_k
        return self.vector_store.store_config.top_k

    def _resolve_subgraph_budget(
        self,
        subgraph_depth: int | None,
        max_total_triples: int | None,
        estimated_triples_per_query: int | None,
    ) -> tuple[int, int, int]:
        """Fill unset induced-subgraph budget arguments from configuration."""
        sc = self.vector_store.store_config
        return (
            sc.induced_subgraph_depth if subgraph_depth is None else subgraph_depth,
            (
                sc.induced_subgraph_max_total_triples
                if max_total_triples is None
                else max_total_triples
            ),
            (
                sc.induced_subgraph_estimated_triples_per_query
                if estimated_triples_per_query is None
                else estimated_triples_per_query
            ),
        )

    def retrieve(
        self,
        query: str,
        top_k: int | None = None,
        expand_sparql: bool = True,
        subgraph_depth: int | None = None,
        max_total_triples: int | None = None,
        estimated_triples_per_query: int | None = None,
    ) -> tuple[RDFGraph, list[str]]:
        """Retrieve top-k hits for one query and optional induced subgraph; returns source ontology IRIs."""
        try:
            asyncio.get_running_loop()
        except RuntimeError:
            return asyncio.run(
                self.aretrieve(
                    query=query,
                    top_k=top_k,
                    expand_sparql=expand_sparql,
                    subgraph_depth=subgraph_depth,
                    max_total_triples=max_total_triples,
                    estimated_triples_per_query=estimated_triples_per_query,
                )
            )
        raise RuntimeError(
            "retrieve() cannot be called from async code; use await aretrieve()"
        )

    def retrieve_ensemble(
        self,
        queries: list[str],
        top_k: int | None = None,
        expand_sparql: bool = True,
        subgraph_depth: int | None = None,
        max_total_triples: int | None = None,
        estimated_triples_per_query: int | None = None,
        trigger_text: str | None = None,
    ) -> tuple[RDFGraph, list[str]]:
        """Sync: one induced graph and source IRIs for the union of vector hits over ``queries``."""
        try:
            asyncio.get_running_loop()
        except RuntimeError:
            return asyncio.run(
                self.aretrieve_ensemble(
                    queries=queries,
                    top_k=top_k,
                    expand_sparql=expand_sparql,
                    subgraph_depth=subgraph_depth,
                    max_total_triples=max_total_triples,
                    estimated_triples_per_query=estimated_triples_per_query,
                    trigger_text=trigger_text,
                )
            )
        raise RuntimeError(
            "retrieve_ensemble() is not allowed inside async code; use aretrieve_ensemble()"
        )

    async def aretrieve(
        self,
        query: str,
        top_k: int | None = None,
        expand_sparql: bool = True,
        subgraph_depth: int | None = None,
        max_total_triples: int | None = None,
        estimated_triples_per_query: int | None = None,
        trigger_text: str | None = None,
    ) -> tuple[RDFGraph, list[str]]:
        """Async single-query variant of :meth:`aretrieve_ensemble`."""
        return await self.aretrieve_ensemble(
            queries=[query],
            top_k=top_k,
            expand_sparql=expand_sparql,
            subgraph_depth=subgraph_depth,
            max_total_triples=max_total_triples,
            estimated_triples_per_query=estimated_triples_per_query,
            trigger_text=trigger_text,
        )

    async def aretrieve_ensemble(
        self,
        queries: list[str],
        top_k: int | None = None,
        expand_sparql: bool = True,
        subgraph_depth: int | None = None,
        max_total_triples: int | None = None,
        estimated_triples_per_query: int | None = None,
        trigger_text: str | None = None,
    ) -> tuple[RDFGraph, list[str]]:
        """Vector search over all ``queries`` once, score-filter, dedupe, single subgraph expansion.

        ``subgraph_depth`` / ``max_total_triples`` / ``estimated_triples_per_query``
        default to the configured values (``ONTOLOGY_PATCH_INDUCED_SUBGRAPH_*``).
        They previously carried literal defaults of 1 / 300 / 24, which
        contradicted the config defaults of 2 / 1200 / 24: the pipeline passed
        config explicitly and was unaffected, but any other caller of this
        public API silently got a 4x smaller snapshot than the deployment was
        configured for.
        """
        self._last_retrieval_metrics = {}
        subgraph_depth, max_total_triples, estimated_triples_per_query = (
            self._resolve_subgraph_budget(
                subgraph_depth, max_total_triples, estimated_triples_per_query
            )
        )
        trigger_source = (trigger_text or "").strip()
        if not queries and not trigger_source:
            return RDFGraph(), []

        eff_top_k = self._effective_top_k(top_k)
        hits_by_query: list[OntologySearchHitsByChannel] = []
        if queries:
            hits_by_query = await self.vector_store.asearch_patch_hits_many(
                queries=queries,
                top_k=eff_top_k,
            )
        sc = self.vector_store.store_config
        pc = self.patch
        eff_max_atoms = pc.effective_max_atoms(len(queries))
        merged = _filter_and_merge_patch_hits(
            hits_by_query,
            store_config=sc,
            patch_config=pc,
            per_query_core_score_ratio=pc.per_query_core_score_ratio,
            per_query_neighborhood_score_ratio=pc.per_query_neighborhood_score_ratio,
            per_query_bm25_score_ratio=pc.per_query_bm25_score_ratio,
            min_core_query_best_score=pc.min_core_query_best_score,
            min_neighborhood_query_best_score=pc.min_neighborhood_query_best_score,
            min_bm25_query_best_score=pc.min_bm25_query_best_score,
            min_merged_max_score=pc.min_merged_max_score,
            max_atoms_total=0,
        )
        atoms_after_dedupe = len(merged)
        merged = [atom for atom in merged if not _is_ontology_declaration_atom(atom)]

        if merged and pc.merged_score_ratio > 0.0:
            merged_top = float(merged[0].score or 0.0)
            merged_floor = merged_top * pc.merged_score_ratio
            merged = [
                atom for atom in merged if float(atom.score or 0.0) >= merged_floor
            ]

        ranked_before_cut = list(merged)

        if merged and pc.mmr_lambda < 1.0:
            merged = _normalize_relevance_scores(merged)
            vectors = await self.vector_store.afetch_vectors(
                [atom.atom_id for atom in merged]
            )
            core_w, neigh_w = normalized_core_neighborhood_weights(sc)
            merged = _mmr_rerank(
                merged,
                vectors,
                mmr_lambda=pc.mmr_lambda,
                max_atoms=eff_max_atoms,
                core_weight=core_w,
                neighborhood_weight=neigh_w,
            )
        elif pc.cross_query_merge_mode in (
            CrossQueryMergeMode.MAX_SCORE,
            CrossQueryMergeMode.SUM_SCORE,
        ):
            merged = _select_atoms_round_robin_by_ontology(
                merged,
                per_ontology_seed_quota=pc.per_ontology_seed_quota,
                max_atoms=eff_max_atoms,
                per_ontology_atom_floor=pc.per_ontology_atom_floor,
                per_role_atom_floor=pc.per_role_atom_floor,
            )
        elif eff_max_atoms > 0:
            merged = merged[:eff_max_atoms]

        trigger_source = trigger_source or " ".join(queries)
        trigger_atoms = await asyncio.to_thread(
            self.vector_store.match_lexical_triggers, trigger_source
        )
        merged, trigger_promoted, trigger_appended = _merge_lexical_trigger_atoms(
            merged, trigger_atoms, fusion=sc.lexical_trigger_fusion
        )
        # After the trigger merge: an exact-case trigger hit is positive
        # evidence and exempts the atom; what remains penalizable is the
        # case-folded BM25/dense residue.
        merged, symbol_case_penalized = _demote_case_mismatched_symbol_atoms(
            merged,
            trigger_source,
            policy=sc.symbol_case_mismatch_policy,
            demote_factor=sc.symbol_case_mismatch_demote_factor,
        )

        if not merged:
            self._last_retrieval_metrics = {
                "query_count": len(queries),
                "top_k": eff_top_k,
                "effective_max_atoms": eff_max_atoms,
                "atoms_after_dedupe": atoms_after_dedupe,
                "atoms_final": 0,
                "seed_iris": [],
                "lexical_trigger_hits": len(trigger_atoms),
                "lexical_trigger_atom_ids": [a.atom_id for a in trigger_atoms],
                "lexical_trigger_promoted": trigger_promoted,
                "lexical_trigger_appended": trigger_appended,
                "symbol_case_penalized": symbol_case_penalized,
            }
            if pc.dump_ontology_ranks:
                self._last_retrieval_metrics["ontology_rank_diagnostics"] = (
                    build_ontology_rank_diagnostics(
                        hits_by_query, ranked_before_cut, []
                    )
                )
            return RDFGraph(), []

        source_iris = _source_iris_from_atoms(merged)
        seeds_by_ontology: dict[str, int] = defaultdict(int)
        for atom in merged:
            if atom.ontology_iri:
                seeds_by_ontology[atom.ontology_iri] += 1

        self._last_retrieval_metrics = {
            "query_count": len(queries),
            "top_k": eff_top_k,
            "effective_max_atoms": eff_max_atoms,
            "merge_mode": pc.cross_query_merge_mode.value,
            "atoms_after_dedupe": atoms_after_dedupe,
            "atoms_final": len(merged),
            "seed_iris": [atom.iri for atom in merged if atom.iri],
            "source_ontology_iris": source_iris,
            "seeds_by_ontology": dict(seeds_by_ontology),
            "lexical_trigger_hits": len(trigger_atoms),
            "lexical_trigger_atom_ids": [a.atom_id for a in trigger_atoms],
            "lexical_trigger_iris": [a.iri for a in trigger_atoms if a.iri],
            "lexical_trigger_promoted": trigger_promoted,
            "lexical_trigger_appended": trigger_appended,
            "symbol_case_penalized": symbol_case_penalized,
        }
        if pc.dump_ontology_ranks:
            self._last_retrieval_metrics["ontology_rank_diagnostics"] = (
                build_ontology_rank_diagnostics(
                    hits_by_query, ranked_before_cut, merged
                )
            )

        if not expand_sparql or self.sparql_tool is None:
            return RDFGraph(), source_iris

        entity_uris, entity_relevance, entity_roles = _ranked_entity_weights(merged)
        signal_entities = self._match_query_unit_signals(trigger_source)
        for signal_iri, signal_onto_iri in sorted(signal_entities.items()):
            if signal_iri in entity_relevance:
                continue
            entity_uris.append(signal_iri)
            entity_relevance[signal_iri] = sc.lexical_trigger_score
            entity_roles[signal_iri] = "resource"
        if signal_entities:
            self._last_retrieval_metrics["query_signal_iris"] = sorted(
                signal_entities.keys()
            )
        hit_ontology_iris = sorted(
            {atom.ontology_iri for atom in merged if atom.ontology_iri}
            | set(signal_entities.values())
        )
        ontology_version_filters: dict[str, set[str]] = {}
        ontology_hash_filters: dict[str, set[str]] = {}
        for atom in merged:
            if atom.ontology_iri and atom.ontology_version:
                ontology_version_filters.setdefault(atom.ontology_iri, set()).add(
                    str(atom.ontology_version)
                )
            if atom.ontology_iri and atom.ontology_hash:
                ontology_hash_filters.setdefault(atom.ontology_iri, set()).add(
                    atom.ontology_hash
                )

        ontology_iris = hit_ontology_iris
        catalog: list[Ontology] | None = None
        triple_store_manager = self.sparql_tool.triple_store_manager
        if triple_store_manager is not None:
            ontology_iris, catalog, expansion_metrics = await _aexpand_ontology_iris(
                triple_store_manager, entity_uris, hit_ontology_iris
            )
            expanded = sorted(set(ontology_iris) - set(hit_ontology_iris))
            if expanded:
                self._last_retrieval_metrics["expanded_ontology_iris"] = expanded
            self._last_retrieval_metrics.update(expansion_metrics)

        merged_context = await self._aresolve_merged_context(
            entity_uris=entity_uris,
            ontology_iris=ontology_iris,
            catalog=catalog,
            ontology_version_filters=ontology_version_filters or None,
            ontology_hash_filters=ontology_hash_filters or None,
            depth=subgraph_depth,
            candidate_pushdown=sc.induced_subgraph_candidate_pushdown,
        )

        schema_graph = self._schema_axiom_graph(merged_context, catalog)
        if schema_graph is not None:
            closure = _schema_closure_entities(
                schema_graph,
                entity_uris,
                max_entities=pc.schema_closure_max_entities,
                ancestor_depth=pc.schema_closure_ancestor_depth,
                seed_relevance=entity_relevance,
            )
            if closure:
                closure_score = _closure_floor_score(entity_relevance)
                for closure_iri, closure_role in closure.items():
                    entity_uris.append(closure_iri)
                    entity_relevance[closure_iri] = closure_score
                    entity_roles[closure_iri] = closure_role
                self._last_retrieval_metrics["schema_closure_iris"] = sorted(closure)

        hub_seed_count = sc.induced_subgraph_hub_seed_count
        ancestor_depth = sc.induced_subgraph_ancestor_closure_depth
        entity_groups: dict[str, str] = {
            atom.iri: atom.ontology_iri
            for atom in merged
            if atom.iri and atom.ontology_iri
        }
        for signal_iri, signal_onto_iri in signal_entities.items():
            entity_groups.setdefault(signal_iri, signal_onto_iri)
        symbol_predicates = tuple(
            URIRef(iri) for iri in sc.induced_subgraph_symbol_predicates
        )

        graph = await self.sparql_tool.aget_induced_subgraph(
            ontologies=catalog,
            merged=merged_context,
            entity_uris=entity_uris,
            entity_relevance=entity_relevance,
            entity_roles=entity_roles,
            ontology_iris=ontology_iris,
            depth=subgraph_depth,
            max_total_triples=max_total_triples,
            estimated_triples_per_query=estimated_triples_per_query,
            ontology_version_filters=ontology_version_filters or None,
            ontology_hash_filters=ontology_hash_filters or None,
            hub_seed_count=hub_seed_count,
            ancestor_closure_depth=ancestor_depth,
            type_promotion_score_factor=(
                sc.induced_subgraph_type_promotion_score_factor
            ),
            seed_order=sc.induced_subgraph_seed_order.value,
            entity_groups=entity_groups,
            extra_description_predicates=symbol_predicates,
        )
        await self._apply_small_module_closure(graph, hit_ontology_iris)

        self._last_retrieval_metrics["snapshot_triple_count"] = len(graph)
        self._last_retrieval_metrics["ontology_iris_for_expansion"] = ontology_iris
        self._last_retrieval_metrics.update(self.sparql_tool.last_finalize_metrics)

        _bind_common_vocab_prefixes(graph)
        return graph, source_iris

aretrieve(query, top_k=None, expand_sparql=True, subgraph_depth=None, max_total_triples=None, estimated_triples_per_query=None, trigger_text=None) async

Async single-query variant of :meth:aretrieve_ensemble.

Source code in ontocast/tool/vector_store/patch_retriever.py

async def aretrieve(
    self,
    query: str,
    top_k: int | None = None,
    expand_sparql: bool = True,
    subgraph_depth: int | None = None,
    max_total_triples: int | None = None,
    estimated_triples_per_query: int | None = None,
    trigger_text: str | None = None,
) -> tuple[RDFGraph, list[str]]:
    """Async single-query variant of :meth:`aretrieve_ensemble`."""
    return await self.aretrieve_ensemble(
        queries=[query],
        top_k=top_k,
        expand_sparql=expand_sparql,
        subgraph_depth=subgraph_depth,
        max_total_triples=max_total_triples,
        estimated_triples_per_query=estimated_triples_per_query,
        trigger_text=trigger_text,
    )

aretrieve_ensemble(queries, top_k=None, expand_sparql=True, subgraph_depth=None, max_total_triples=None, estimated_triples_per_query=None, trigger_text=None) async

Vector search over all queries once, score-filter, dedupe, single subgraph expansion.

subgraph_depth / max_total_triples / estimated_triples_per_query default to the configured values (ONTOLOGY_PATCH_INDUCED_SUBGRAPH_*). They previously carried literal defaults of 1 / 300 / 24, which contradicted the config defaults of 2 / 1200 / 24: the pipeline passed config explicitly and was unaffected, but any other caller of this public API silently got a 4x smaller snapshot than the deployment was configured for.

Source code in ontocast/tool/vector_store/patch_retriever.py

async def aretrieve_ensemble(
    self,
    queries: list[str],
    top_k: int | None = None,
    expand_sparql: bool = True,
    subgraph_depth: int | None = None,
    max_total_triples: int | None = None,
    estimated_triples_per_query: int | None = None,
    trigger_text: str | None = None,
) -> tuple[RDFGraph, list[str]]:
    """Vector search over all ``queries`` once, score-filter, dedupe, single subgraph expansion.

    ``subgraph_depth`` / ``max_total_triples`` / ``estimated_triples_per_query``
    default to the configured values (``ONTOLOGY_PATCH_INDUCED_SUBGRAPH_*``).
    They previously carried literal defaults of 1 / 300 / 24, which
    contradicted the config defaults of 2 / 1200 / 24: the pipeline passed
    config explicitly and was unaffected, but any other caller of this
    public API silently got a 4x smaller snapshot than the deployment was
    configured for.
    """
    self._last_retrieval_metrics = {}
    subgraph_depth, max_total_triples, estimated_triples_per_query = (
        self._resolve_subgraph_budget(
            subgraph_depth, max_total_triples, estimated_triples_per_query
        )
    )
    trigger_source = (trigger_text or "").strip()
    if not queries and not trigger_source:
        return RDFGraph(), []

    eff_top_k = self._effective_top_k(top_k)
    hits_by_query: list[OntologySearchHitsByChannel] = []
    if queries:
        hits_by_query = await self.vector_store.asearch_patch_hits_many(
            queries=queries,
            top_k=eff_top_k,
        )
    sc = self.vector_store.store_config
    pc = self.patch
    eff_max_atoms = pc.effective_max_atoms(len(queries))
    merged = _filter_and_merge_patch_hits(
        hits_by_query,
        store_config=sc,
        patch_config=pc,
        per_query_core_score_ratio=pc.per_query_core_score_ratio,
        per_query_neighborhood_score_ratio=pc.per_query_neighborhood_score_ratio,
        per_query_bm25_score_ratio=pc.per_query_bm25_score_ratio,
        min_core_query_best_score=pc.min_core_query_best_score,
        min_neighborhood_query_best_score=pc.min_neighborhood_query_best_score,
        min_bm25_query_best_score=pc.min_bm25_query_best_score,
        min_merged_max_score=pc.min_merged_max_score,
        max_atoms_total=0,
    )
    atoms_after_dedupe = len(merged)
    merged = [atom for atom in merged if not _is_ontology_declaration_atom(atom)]

    if merged and pc.merged_score_ratio > 0.0:
        merged_top = float(merged[0].score or 0.0)
        merged_floor = merged_top * pc.merged_score_ratio
        merged = [
            atom for atom in merged if float(atom.score or 0.0) >= merged_floor
        ]

    ranked_before_cut = list(merged)

    if merged and pc.mmr_lambda < 1.0:
        merged = _normalize_relevance_scores(merged)
        vectors = await self.vector_store.afetch_vectors(
            [atom.atom_id for atom in merged]
        )
        core_w, neigh_w = normalized_core_neighborhood_weights(sc)
        merged = _mmr_rerank(
            merged,
            vectors,
            mmr_lambda=pc.mmr_lambda,
            max_atoms=eff_max_atoms,
            core_weight=core_w,
            neighborhood_weight=neigh_w,
        )
    elif pc.cross_query_merge_mode in (
        CrossQueryMergeMode.MAX_SCORE,
        CrossQueryMergeMode.SUM_SCORE,
    ):
        merged = _select_atoms_round_robin_by_ontology(
            merged,
            per_ontology_seed_quota=pc.per_ontology_seed_quota,
            max_atoms=eff_max_atoms,
            per_ontology_atom_floor=pc.per_ontology_atom_floor,
            per_role_atom_floor=pc.per_role_atom_floor,
        )
    elif eff_max_atoms > 0:
        merged = merged[:eff_max_atoms]

    trigger_source = trigger_source or " ".join(queries)
    trigger_atoms = await asyncio.to_thread(
        self.vector_store.match_lexical_triggers, trigger_source
    )
    merged, trigger_promoted, trigger_appended = _merge_lexical_trigger_atoms(
        merged, trigger_atoms, fusion=sc.lexical_trigger_fusion
    )
    # After the trigger merge: an exact-case trigger hit is positive
    # evidence and exempts the atom; what remains penalizable is the
    # case-folded BM25/dense residue.
    merged, symbol_case_penalized = _demote_case_mismatched_symbol_atoms(
        merged,
        trigger_source,
        policy=sc.symbol_case_mismatch_policy,
        demote_factor=sc.symbol_case_mismatch_demote_factor,
    )

    if not merged:
        self._last_retrieval_metrics = {
            "query_count": len(queries),
            "top_k": eff_top_k,
            "effective_max_atoms": eff_max_atoms,
            "atoms_after_dedupe": atoms_after_dedupe,
            "atoms_final": 0,
            "seed_iris": [],
            "lexical_trigger_hits": len(trigger_atoms),
            "lexical_trigger_atom_ids": [a.atom_id for a in trigger_atoms],
            "lexical_trigger_promoted": trigger_promoted,
            "lexical_trigger_appended": trigger_appended,
            "symbol_case_penalized": symbol_case_penalized,
        }
        if pc.dump_ontology_ranks:
            self._last_retrieval_metrics["ontology_rank_diagnostics"] = (
                build_ontology_rank_diagnostics(
                    hits_by_query, ranked_before_cut, []
                )
            )
        return RDFGraph(), []

    source_iris = _source_iris_from_atoms(merged)
    seeds_by_ontology: dict[str, int] = defaultdict(int)
    for atom in merged:
        if atom.ontology_iri:
            seeds_by_ontology[atom.ontology_iri] += 1

    self._last_retrieval_metrics = {
        "query_count": len(queries),
        "top_k": eff_top_k,
        "effective_max_atoms": eff_max_atoms,
        "merge_mode": pc.cross_query_merge_mode.value,
        "atoms_after_dedupe": atoms_after_dedupe,
        "atoms_final": len(merged),
        "seed_iris": [atom.iri for atom in merged if atom.iri],
        "source_ontology_iris": source_iris,
        "seeds_by_ontology": dict(seeds_by_ontology),
        "lexical_trigger_hits": len(trigger_atoms),
        "lexical_trigger_atom_ids": [a.atom_id for a in trigger_atoms],
        "lexical_trigger_iris": [a.iri for a in trigger_atoms if a.iri],
        "lexical_trigger_promoted": trigger_promoted,
        "lexical_trigger_appended": trigger_appended,
        "symbol_case_penalized": symbol_case_penalized,
    }
    if pc.dump_ontology_ranks:
        self._last_retrieval_metrics["ontology_rank_diagnostics"] = (
            build_ontology_rank_diagnostics(
                hits_by_query, ranked_before_cut, merged
            )
        )

    if not expand_sparql or self.sparql_tool is None:
        return RDFGraph(), source_iris

    entity_uris, entity_relevance, entity_roles = _ranked_entity_weights(merged)
    signal_entities = self._match_query_unit_signals(trigger_source)
    for signal_iri, signal_onto_iri in sorted(signal_entities.items()):
        if signal_iri in entity_relevance:
            continue
        entity_uris.append(signal_iri)
        entity_relevance[signal_iri] = sc.lexical_trigger_score
        entity_roles[signal_iri] = "resource"
    if signal_entities:
        self._last_retrieval_metrics["query_signal_iris"] = sorted(
            signal_entities.keys()
        )
    hit_ontology_iris = sorted(
        {atom.ontology_iri for atom in merged if atom.ontology_iri}
        | set(signal_entities.values())
    )
    ontology_version_filters: dict[str, set[str]] = {}
    ontology_hash_filters: dict[str, set[str]] = {}
    for atom in merged:
        if atom.ontology_iri and atom.ontology_version:
            ontology_version_filters.setdefault(atom.ontology_iri, set()).add(
                str(atom.ontology_version)
            )
        if atom.ontology_iri and atom.ontology_hash:
            ontology_hash_filters.setdefault(atom.ontology_iri, set()).add(
                atom.ontology_hash
            )

    ontology_iris = hit_ontology_iris
    catalog: list[Ontology] | None = None
    triple_store_manager = self.sparql_tool.triple_store_manager
    if triple_store_manager is not None:
        ontology_iris, catalog, expansion_metrics = await _aexpand_ontology_iris(
            triple_store_manager, entity_uris, hit_ontology_iris
        )
        expanded = sorted(set(ontology_iris) - set(hit_ontology_iris))
        if expanded:
            self._last_retrieval_metrics["expanded_ontology_iris"] = expanded
        self._last_retrieval_metrics.update(expansion_metrics)

    merged_context = await self._aresolve_merged_context(
        entity_uris=entity_uris,
        ontology_iris=ontology_iris,
        catalog=catalog,
        ontology_version_filters=ontology_version_filters or None,
        ontology_hash_filters=ontology_hash_filters or None,
        depth=subgraph_depth,
        candidate_pushdown=sc.induced_subgraph_candidate_pushdown,
    )

    schema_graph = self._schema_axiom_graph(merged_context, catalog)
    if schema_graph is not None:
        closure = _schema_closure_entities(
            schema_graph,
            entity_uris,
            max_entities=pc.schema_closure_max_entities,
            ancestor_depth=pc.schema_closure_ancestor_depth,
            seed_relevance=entity_relevance,
        )
        if closure:
            closure_score = _closure_floor_score(entity_relevance)
            for closure_iri, closure_role in closure.items():
                entity_uris.append(closure_iri)
                entity_relevance[closure_iri] = closure_score
                entity_roles[closure_iri] = closure_role
            self._last_retrieval_metrics["schema_closure_iris"] = sorted(closure)

    hub_seed_count = sc.induced_subgraph_hub_seed_count
    ancestor_depth = sc.induced_subgraph_ancestor_closure_depth
    entity_groups: dict[str, str] = {
        atom.iri: atom.ontology_iri
        for atom in merged
        if atom.iri and atom.ontology_iri
    }
    for signal_iri, signal_onto_iri in signal_entities.items():
        entity_groups.setdefault(signal_iri, signal_onto_iri)
    symbol_predicates = tuple(
        URIRef(iri) for iri in sc.induced_subgraph_symbol_predicates
    )

    graph = await self.sparql_tool.aget_induced_subgraph(
        ontologies=catalog,
        merged=merged_context,
        entity_uris=entity_uris,
        entity_relevance=entity_relevance,
        entity_roles=entity_roles,
        ontology_iris=ontology_iris,
        depth=subgraph_depth,
        max_total_triples=max_total_triples,
        estimated_triples_per_query=estimated_triples_per_query,
        ontology_version_filters=ontology_version_filters or None,
        ontology_hash_filters=ontology_hash_filters or None,
        hub_seed_count=hub_seed_count,
        ancestor_closure_depth=ancestor_depth,
        type_promotion_score_factor=(
            sc.induced_subgraph_type_promotion_score_factor
        ),
        seed_order=sc.induced_subgraph_seed_order.value,
        entity_groups=entity_groups,
        extra_description_predicates=symbol_predicates,
    )
    await self._apply_small_module_closure(graph, hit_ontology_iris)

    self._last_retrieval_metrics["snapshot_triple_count"] = len(graph)
    self._last_retrieval_metrics["ontology_iris_for_expansion"] = ontology_iris
    self._last_retrieval_metrics.update(self.sparql_tool.last_finalize_metrics)

    _bind_common_vocab_prefixes(graph)
    return graph, source_iris

retrieve(query, top_k=None, expand_sparql=True, subgraph_depth=None, max_total_triples=None, estimated_triples_per_query=None)

Retrieve top-k hits for one query and optional induced subgraph; returns source ontology IRIs.

Source code in ontocast/tool/vector_store/patch_retriever.py

def retrieve(
    self,
    query: str,
    top_k: int | None = None,
    expand_sparql: bool = True,
    subgraph_depth: int | None = None,
    max_total_triples: int | None = None,
    estimated_triples_per_query: int | None = None,
) -> tuple[RDFGraph, list[str]]:
    """Retrieve top-k hits for one query and optional induced subgraph; returns source ontology IRIs."""
    try:
        asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(
            self.aretrieve(
                query=query,
                top_k=top_k,
                expand_sparql=expand_sparql,
                subgraph_depth=subgraph_depth,
                max_total_triples=max_total_triples,
                estimated_triples_per_query=estimated_triples_per_query,
            )
        )
    raise RuntimeError(
        "retrieve() cannot be called from async code; use await aretrieve()"
    )

retrieve_ensemble(queries, top_k=None, expand_sparql=True, subgraph_depth=None, max_total_triples=None, estimated_triples_per_query=None, trigger_text=None)

Source code in ontocast/tool/vector_store/patch_retriever.py

def retrieve_ensemble(
    self,
    queries: list[str],
    top_k: int | None = None,
    expand_sparql: bool = True,
    subgraph_depth: int | None = None,
    max_total_triples: int | None = None,
    estimated_triples_per_query: int | None = None,
    trigger_text: str | None = None,
) -> tuple[RDFGraph, list[str]]:
    """Sync: one induced graph and source IRIs for the union of vector hits over ``queries``."""
    try:
        asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(
            self.aretrieve_ensemble(
                queries=queries,
                top_k=top_k,
                expand_sparql=expand_sparql,
                subgraph_depth=subgraph_depth,
                max_total_triples=max_total_triples,
                estimated_triples_per_query=estimated_triples_per_query,
                trigger_text=trigger_text,
            )
        )
    raise RuntimeError(
        "retrieve_ensemble() is not allowed inside async code; use aretrieve_ensemble()"
    )

OntologySearchHit

Bases: BasePydanticModel

Typed retrieval result that separates atom payload from ranking metadata.

Source code in ontocast/tool/vector_store/core.py

class OntologySearchHit(BasePydanticModel):
    """Typed retrieval result that separates atom payload from ranking metadata."""

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

OpenAIEmbeddingTool

Bases: _LangChainEmbeddingTool

OpenAI embeddings via langchain-openai.

Source code in ontocast/tool/vector_store/embedding.py

class OpenAIEmbeddingTool(_LangChainEmbeddingTool):
    """OpenAI embeddings via langchain-openai."""

    def _build_embedder(self) -> Embeddings:
        api_key = (
            SecretStr(self.config.api_key) if self.config.api_key is not None else None
        )
        OpenAIEmbeddings = require(
            "langchain_openai", feature="OpenAI embeddings"
        ).OpenAIEmbeddings
        return OpenAIEmbeddings(
            model=self.config.model_name,
            api_key=api_key,
            base_url=self.config.base_url,
        )

VectorStoreManager

Bases: Tool

Abstract interface for vector store implementations.

Source code in ontocast/tool/vector_store/core.py

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

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

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

@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

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

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

@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

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

@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

@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

@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

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

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

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

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

@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

@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

@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

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

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"
    )

__getattr__(name)

Resolve the backend managers on first access.

Source code in ontocast/tool/vector_store/__init__.py

def __getattr__(name: str) -> Any:
    """Resolve the backend managers on first access."""
    target = _LAZY_EXPORTS.get(name)
    if target is None:
        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
    import importlib

    module_name, attribute = target
    value = getattr(importlib.import_module(module_name), attribute)
    globals()[name] = value
    return value

create_vector_store_manager(tool_config, embedding, sparse_embedding=None)

Return a vector store manager for the configured backend.

Selection is driven by VectorStoreConfig.backend. The default, :attr:~ontocast.onto.enum.VectorStoreBackend.AUTO, infers the backend from whichever connection setting is populated and otherwise resolves to :attr:~ontocast.onto.enum.VectorStoreBackend.NONE, returning None. A deployment that configures neither Qdrant nor LanceDB has no vector retrieval: ontology context comes from a single working ontology, which is the default :class:~ontocast.onto.enum.OntologyContextMode.

Parameters:

Name	Type	Description	Default
tool_config	ToolConfig	The resolved tool configuration.	required
embedding	EmbeddingTool	Dense embedding provider.	required
sparse_embedding	FastembedBm25SparseTool | None	BM25 sparse provider, required by both backends.	None

Returns:

Type	Description
VectorStoreManager | None	A manager for the selected backend, or None when the backend is
VectorStoreManager | None	explicitly disabled.

Raises:

Type	Description
ValueError	If an explicitly requested backend is not configured, or if Qdrant's vector_size contradicts the embedding dimension.

Source code in ontocast/tool/vector_store/factory.py

def create_vector_store_manager(
    tool_config: ToolConfig,
    embedding: EmbeddingTool,
    sparse_embedding: FastembedBm25SparseTool | None = None,
) -> VectorStoreManager | None:
    """Return a vector store manager for the configured backend.

    Selection is driven by ``VectorStoreConfig.backend``. The default,
    :attr:`~ontocast.onto.enum.VectorStoreBackend.AUTO`, infers the backend from
    whichever connection setting is populated and otherwise resolves to
    :attr:`~ontocast.onto.enum.VectorStoreBackend.NONE`, returning ``None``.
    A deployment that configures neither Qdrant nor LanceDB has **no** vector
    retrieval: ontology context comes from a single working ontology, which is
    the default :class:`~ontocast.onto.enum.OntologyContextMode`.

    Args:
        tool_config: The resolved tool configuration.
        embedding: Dense embedding provider.
        sparse_embedding: BM25 sparse provider, required by both backends.

    Returns:
        A manager for the selected backend, or ``None`` when the backend is
        explicitly disabled.

    Raises:
        ValueError: If an explicitly requested backend is not configured, or if
            Qdrant's ``vector_size`` contradicts the embedding dimension.
    """
    backend = _resolve_backend(tool_config)

    if backend is VectorStoreBackend.NONE:
        return None

    if backend is VectorStoreBackend.QDRANT:
        q_vs = tool_config.qdrant.vector_size
        emb_dim = tool_config.embedding.dimension
        if q_vs is not None and q_vs != emb_dim:
            raise ValueError(
                "QdrantConfig.vector_size must match "
                "EmbeddingConfig.dimension when set "
                f"(got vector_size={q_vs}, embedding.dimension={emb_dim})"
            )
        from ontocast.tool.vector_store.qdrant import QdrantVectorStoreManager

        return QdrantVectorStoreManager(
            store_config=tool_config.vector_store,
            qdrant_config=tool_config.qdrant,
            embedding=embedding,
            sparse_embedding=sparse_embedding,
        )

    from ontocast.tool.vector_store.lancedb import LanceDBVectorStoreManager

    return LanceDBVectorStoreManager(
        store_config=tool_config.vector_store,
        lancedb_config=tool_config.lancedb,
        embedding=embedding,
        sparse_embedding=sparse_embedding,
    )