ontocast.tool.vector_store

Vector store package for ontology patch retrieval.

EmbeddingContractMismatchError

Bases: ValueError

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

Typical causes: switching EMBEDDING_MODEL_NAME without recreating the Qdrant collection, or a provider returning an unexpected vector length.

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

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

    Typical causes: switching ``EMBEDDING_MODEL_NAME`` without recreating the Qdrant
    collection, or a provider returning an unexpected vector length.
    """

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(self, texts: list[str]) -> list[list[float]]:
        """Return vectors for all given texts."""

    def embed_one(self, text: str) -> list[float]:
        """Return a vector for one text."""
        vectors = self.embed([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) abstractmethod

Return vectors for all given texts.

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

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

embed_one(text)

Return a vector for one text.

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

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

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
        try:
            fastembed_mod = importlib.import_module("fastembed")
        except ImportError as error:
            raise ImportError(
                "BM25 sparse embeddings require fastembed. "
                "Install it with: uv add 'fastembed[all]'"
            ) from error
        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[qdrant_models.SparseVector]:
        """Return Qdrant sparse vectors for all given texts."""
        if not texts:
            return []
        model = self._get_embedder()
        out: list[qdrant_models.SparseVector] = []
        for sparse_emb in model.embed(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(
                qdrant_models.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) -> qdrant_models.SparseVector:
        vectors = self.embed_sparse([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 all given texts.

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

def embed_sparse(self, texts: list[str]) -> list[qdrant_models.SparseVector]:
    """Return Qdrant sparse vectors for all given texts."""
    if not texts:
        return []
    model = self._get_embedder()
    out: list[qdrant_models.SparseVector] = []
    for sparse_emb in model.embed(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(
            qdrant_models.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

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

    @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.

By default, ontology atomization skips focal IRIs in common W3C and DC vocabulary namespaces (see module-level exclusions). Set embed_standard_vocab_iris=True to restore legacy behavior (embed every URIRef in the graph). Facts sources are still restricted to facts_namespace only; vocabulary exclusion does not apply to them.

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

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

    By default, ontology atomization skips focal IRIs in common W3C and DC vocabulary
    namespaces (see module-level exclusions). Set ``embed_standard_vocab_iris=True`` to
    restore legacy behavior (embed every URIRef in the graph). Facts sources are still
    restricted to ``facts_namespace`` only; vocabulary exclusion does not apply 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).",
    )

    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
        excluded_vocab: frozenset[str] | None = None
        if not isinstance(source, Facts) 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,
        )
        predicate_uris = {p for (_, p, _) in embedding_graph if isinstance(p, 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_predicate_usage(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)
            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,
                    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 _collect_focal_entities(
        self,
        graph: RDFGraph,
        focal_namespace: str | None = None,
        excluded_vocab_prefixes: frozenset[str] | None = None,
    ) -> 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)
            }

        return cast(
            list[URIRef],
            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_literals(
            graph,
            parent,
            [RDFS.label, SKOS.prefLabel, DCTERMS.title, SKOS.altLabel],
            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_ref = cast(URIRef, prop)
            prop_verb = self._normalize_uri(prop_ref)  # bare verb for SPO
            ranges = self._collect_range_labels(
                prop_ref, 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_ref, 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_ref = cast(URIRef, prop)
            prop_verb = self._normalize_uri(prop_ref)  # bare verb for SPO
            domains = self._collect_domain_labels(
                prop_ref, 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_ref, graph, clues)

    def _build_minimal_representation(self, entity: URIRef) -> str:
        """IRI local name as keyword-oriented tokens: split camelCase/PascalCase, etc.

        Compact text for sparse BM25 (no labels or gloss); only the focal entity IRI
        is tokenized (see ``normalize_uri_local_name``).
        """
        return normalize_uri_local_name(entity)

    def _build_core_representation(
        self, entity: URIRef, graph: RDFGraph, role: str
    ) -> str:
        labels = self._collect_literals(
            graph,
            entity,
            [RDFS.label, SKOS.prefLabel, DCTERMS.title, SKOS.altLabel],
            5,
        )
        descriptions = self._collect_literals(
            graph, entity, [RDFS.comment, DCTERMS.description, SKOS.definition], 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):
                par_ref = cast(URIRef, par)
                siblings = sorted(
                    (
                        sib
                        for sib in parent_to_children.get(par_ref, set[URIRef]())
                        if sib != entity
                    ),
                    key=str,
                )
                for sib in siblings[:6]:
                    sib_ref = cast(URIRef, sib)
                    clues.append(
                        f"{self._normalize_uri(sib_ref)} is also a kind of "
                        f"{self._parent_resource_phrase(graph, par_ref)}"
                    )

            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]:
        values: list[str] = []
        seen: set[str] = set()
        for predicate in predicates:
            for _, _, obj in graph.triples((subject, predicate, None)):
                if not isinstance(obj, Literal):
                    continue
                normalized = self._normalize_string(str(obj))
                if not normalized or normalized in seen:
                    continue
                values.append(normalized)
                seen.add(normalized)
                if len(values) >= max_items:
                    return values
        return values

    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
    excluded_vocab: frozenset[str] | None = None
    if not isinstance(source, Facts) 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,
    )
    predicate_uris = {p for (_, p, _) in embedding_graph if isinstance(p, 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_predicate_usage(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)
        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,
                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: Any = PrivateAttr(default=None)

    def _get_embedder(self) -> Any:
        if self._embedder is not None:
            return self._embedder
        try:
            sentence_transformers = importlib.import_module("sentence_transformers")
        except ImportError as error:
            raise ImportError(
                "HuggingFace embeddings require sentence-transformers. "
                "Install it with: uv add sentence-transformers"
            ) from error
        self._embedder = sentence_transformers.SentenceTransformer(
            self.config.model_name
        )
        return self._embedder

    def embed(self, texts: list[str]) -> list[list[float]]:
        if not texts:
            return []
        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:
        return OllamaEmbeddings(
            model=self.config.model_name,
            base_url=self.config.base_url,
        )

    def embed(self, texts: list[str]) -> list[list[float]]:
        if not texts:
            return []
        try:
            return super().embed(texts)
        except Exception:
            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: QdrantVectorStore = Field(exclude=True)
    sparql_tool: Any | None = Field(default=None, exclude=True)
    patch: PatchRetrievalConfig = Field(
        default_factory=PatchRetrievalConfig,
        exclude=True,
    )

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

    def retrieve(
        self,
        query: str,
        top_k: int | None = None,
        expand_sparql: bool = True,
        subgraph_depth: int = 1,
        max_total_triples: int = 300,
        estimated_triples_per_query: int = 24,
    ) -> 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 = 1,
        max_total_triples: int = 300,
        estimated_triples_per_query: int = 24,
    ) -> 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,
                )
            )
        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 = 1,
        max_total_triples: int = 300,
        estimated_triples_per_query: int = 24,
    ) -> 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,
        )

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

        Hits are filtered per query and per channel relative to each channel's best
        score (see ``PatchRetrievalConfig`` per-query ratio fields for core,
        neighborhood, and BM25), then merged by rank fusion so channels with
        different score distributions all contribute. Optional per-channel
        min-best filters and ``min_merged_max_score`` reject weak or irrelevant
        candidates.

        Returns the merged RDF graph (possibly disconnected across ontologies) and sorted
        distinct ontology IRIs that contributed vector hits.
        """
        if not queries:
            return RDFGraph(), []
        eff_top_k = self._effective_top_k(top_k)
        hits_by_query = await self.vector_store.asearch_patch_hits_many(
            queries=queries,
            top_k=eff_top_k,
        )
        qc = self.vector_store.config
        pc = self.patch
        merged = _filter_and_merge_patch_hits(
            hits_by_query,
            qdrant_config=qc,
            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,
        )
        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
            ]
        if merged and pc.mmr_lambda < 1.0:
            vectors = await self.vector_store.afetch_vectors(
                [atom.atom_id for atom in merged]
            )
            merged = _mmr_rerank(
                merged,
                vectors,
                mmr_lambda=pc.mmr_lambda,
                max_atoms=pc.max_atoms,
                core_weight=qc.fusion_core_weight,
                neighborhood_weight=qc.fusion_neighborhood_weight,
            )
        elif pc.max_atoms > 0:
            merged = merged[: pc.max_atoms]
        source_iris = _source_iris_from_atoms(merged)

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

        if not merged:
            return RDFGraph(), []

        entity_uris, entity_relevance = _ranked_entity_weights(merged)
        ontology_iris = sorted(
            {atom.ontology_iri for atom in merged if atom.ontology_iri}
        )
        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
                )

        graph = await self.sparql_tool.aget_induced_subgraph(
            entity_uris=entity_uris,
            entity_relevance=entity_relevance,
            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,
        )
        _bind_common_vocab_prefixes(graph)
        return graph, source_iris

aretrieve(query, top_k=None, expand_sparql=True, subgraph_depth=1, max_total_triples=300, estimated_triples_per_query=24) 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 = 1,
    max_total_triples: int = 300,
    estimated_triples_per_query: int = 24,
) -> 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,
    )

aretrieve_ensemble(queries, top_k=None, expand_sparql=True, subgraph_depth=1, max_total_triples=300, estimated_triples_per_query=24) async

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

Hits are filtered per query and per channel relative to each channel's best score (see PatchRetrievalConfig per-query ratio fields for core, neighborhood, and BM25), then merged by rank fusion so channels with different score distributions all contribute. Optional per-channel min-best filters and min_merged_max_score reject weak or irrelevant candidates.

Returns the merged RDF graph (possibly disconnected across ontologies) and sorted distinct ontology IRIs that contributed vector hits.

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 = 1,
    max_total_triples: int = 300,
    estimated_triples_per_query: int = 24,
) -> tuple[RDFGraph, list[str]]:
    """Vector search over all ``queries`` once, score-filter, dedupe, single subgraph expansion.

    Hits are filtered per query and per channel relative to each channel's best
    score (see ``PatchRetrievalConfig`` per-query ratio fields for core,
    neighborhood, and BM25), then merged by rank fusion so channels with
    different score distributions all contribute. Optional per-channel
    min-best filters and ``min_merged_max_score`` reject weak or irrelevant
    candidates.

    Returns the merged RDF graph (possibly disconnected across ontologies) and sorted
    distinct ontology IRIs that contributed vector hits.
    """
    if not queries:
        return RDFGraph(), []
    eff_top_k = self._effective_top_k(top_k)
    hits_by_query = await self.vector_store.asearch_patch_hits_many(
        queries=queries,
        top_k=eff_top_k,
    )
    qc = self.vector_store.config
    pc = self.patch
    merged = _filter_and_merge_patch_hits(
        hits_by_query,
        qdrant_config=qc,
        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,
    )
    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
        ]
    if merged and pc.mmr_lambda < 1.0:
        vectors = await self.vector_store.afetch_vectors(
            [atom.atom_id for atom in merged]
        )
        merged = _mmr_rerank(
            merged,
            vectors,
            mmr_lambda=pc.mmr_lambda,
            max_atoms=pc.max_atoms,
            core_weight=qc.fusion_core_weight,
            neighborhood_weight=qc.fusion_neighborhood_weight,
        )
    elif pc.max_atoms > 0:
        merged = merged[: pc.max_atoms]
    source_iris = _source_iris_from_atoms(merged)

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

    if not merged:
        return RDFGraph(), []

    entity_uris, entity_relevance = _ranked_entity_weights(merged)
    ontology_iris = sorted(
        {atom.ontology_iri for atom in merged if atom.ontology_iri}
    )
    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
            )

    graph = await self.sparql_tool.aget_induced_subgraph(
        entity_uris=entity_uris,
        entity_relevance=entity_relevance,
        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,
    )
    _bind_common_vocab_prefixes(graph)
    return graph, source_iris

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

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 = 1,
    max_total_triples: int = 300,
    estimated_triples_per_query: int = 24,
) -> 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=1, max_total_triples=300, estimated_triples_per_query=24)

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 = 1,
    max_total_triples: int = 300,
    estimated_triples_per_query: int = 24,
) -> 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,
            )
        )
    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
        )
        return OpenAIEmbeddings(
            model=self.config.model_name,
            openai_api_key=api_key,
            openai_api_base=self.config.base_url,
        )

QdrantVectorStore

Bases: VectorStoreTool

Stores ontology atoms in Qdrant and supports similarity lookup.

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

class QdrantVectorStore(VectorStoreTool):
    """Stores ontology atoms in Qdrant and supports similarity lookup."""

    config: QdrantConfig = Field(default_factory=QdrantConfig)
    embedding: EmbeddingTool = Field(..., exclude=True)
    sparse_embedding: FastembedBm25SparseTool | None = Field(default=None, exclude=True)
    atomizer: GraphAtomizer = Field(default_factory=GraphAtomizer, exclude=True)
    _client: QdrantClient | None = PrivateAttr(default=None)

    @property
    def embedding_config(self) -> EmbeddingConfig:
        return self.embedding.config

    def _require_sparse_embedding_tool(self) -> FastembedBm25SparseTool:
        if self.sparse_embedding is None:
            raise ValueError(
                "BM25 sparse embedding is required for vector search but "
                "sparse_embedding was not wired"
            )
        return self.sparse_embedding

    def _encode_single_query_vectors(
        self, query: str
    ) -> tuple[list[float], list[float], qdrant_models.SparseVector]:
        triples = self._encode_query_vectors_batch([query])
        return triples[0]

    def _encode_query_vectors_batch(
        self, queries: list[str]
    ) -> list[tuple[list[float], list[float], qdrant_models.SparseVector]]:
        n = len(queries)
        if n == 0:
            return []
        dense_vecs = self.embedding.embed(queries)
        if len(dense_vecs) != n:
            raise ValueError(
                "Embedding provider returned mismatched vectors for queries"
            )
        for i, vec in enumerate(dense_vecs):
            self._require_embedding_vector_length(vec, role=f"Query embedding[{i}]")
        sparse_vecs = self._require_sparse_embedding_tool().embed_sparse(queries)
        if len(sparse_vecs) != n:
            raise ValueError(
                "BM25 embedder returned mismatched sparse vectors for queries"
            )
        return [(dense_vecs[i], dense_vecs[i], sparse_vecs[i]) for i in range(n)]

    def _normalized_fusion_weights(self) -> tuple[float, float, float]:
        """Weights for core / neighborhood / BM25 reciprocal-rank fusion (sum to 1)."""
        cw = self.config.fusion_core_weight
        nw = self.config.fusion_neighborhood_weight
        bw = self.config.fusion_bm25_weight
        total = cw + nw + bw
        if total <= 0.0:
            return (1.0 / 3.0, 1.0 / 3.0, 1.0 / 3.0)
        return (cw / total, nw / total, bw / total)

    @property
    def client(self) -> QdrantClient:
        if self._client is None:
            if self.config.uri is None:
                raise ValueError(
                    "Qdrant URI is required to initialize vector store client"
                )
            self._client = QdrantClient(
                url=self.config.uri,
                api_key=self.config.api_key,
                grpc_port=self.config.grpc_port,
                prefer_grpc=self.config.use_grpc,
            )
        return self._client

    def _ontology_collection_name(self) -> str:
        name = self.config.ontology_collection
        if name is None:
            raise ValueError(
                "Qdrant ontology_collection is unset; ensure QdrantConfig validation"
                " ran or call apply_tenancy before vector operations"
            )
        return name

    def supports_tenancy_partition(self) -> bool:
        return True

    async def initialize(self) -> None:
        """Create ontology/facts collections and payload indexes if missing."""
        ontology_col = self.config.ontology_collection
        facts_col = self.config.facts_collection
        assert ontology_col is not None
        assert facts_col is not None
        self._ensure_named_vector_collection(ontology_col)
        self._ensure_named_vector_collection(facts_col)

        self._ensure_payload_index(
            collection_name=ontology_col, field_name="ontology_iri"
        )
        self._ensure_payload_index(
            collection_name=ontology_col, field_name="ontology_version"
        )
        self._ensure_payload_index(
            collection_name=ontology_col, field_name="ontology_hash"
        )
        self._ensure_payload_index(collection_name=ontology_col, field_name="iri")

    async def clean_tenancy(
        self,
        tenant: str,
        project: str,
        *,
        sep: str = TENANCY_SEP,
    ) -> None:
        """Delete Qdrant collections named for ``tenant`` / ``project``."""
        t, p = tenant.strip(), project.strip()
        for name in (
            tenant_project_ontologies_name(t, p, sep=sep),
            tenant_project_facts_name(t, p, sep=sep),
        ):
            if self.client.collection_exists(collection_name=name):
                self.client.delete_collection(collection_name=name)
                logger.info("Deleted Qdrant collection %s", name)

    def apply_tenancy(
        self,
        tenant: str,
        project: str,
        *,
        sep: str = TENANCY_SEP,
    ) -> None:
        """Point config at collections for ``tenant`` / ``project``.

        Call :meth:`initialize` after.
        """
        t, p = tenant.strip(), project.strip()
        self.config.ontology_collection = tenant_project_ontologies_name(t, p, sep=sep)
        self.config.facts_collection = tenant_project_facts_name(t, p, sep=sep)

    def _dense_dimension(self) -> int:
        """Dense vector length for ``VectorParams`` and dense query validation."""
        return self.config.vector_size or self.embedding_config.dimension

    def _metadata_embedding_dimension(self) -> int:
        """Dense dimension stored in collection metadata."""
        return self._dense_dimension()

    def _embedding_model_fingerprint(self) -> str:
        ec = self.embedding_config
        dense_part = f"dense:{ec.provider.value}:{ec.model_name}"
        return f"{dense_part}|bm25={ec.bm25_model_name}"

    def _embedding_fingerprint_matches(self, stored: str) -> bool:
        return stored == self._embedding_model_fingerprint()

    def _collection_embedding_metadata(self, metadata_dim: int) -> dict[str, Any]:
        return {
            QDRANT_META_EMBEDDING_DIMENSION: metadata_dim,
            QDRANT_META_EMBEDDING_MODEL: self._embedding_model_fingerprint(),
        }

    def _coerce_metadata_int(self, value: Any, *, field: str, collection: str) -> int:
        if type(value) is bool:
            raise ValueError(
                f"Qdrant collection '{collection}' metadata {field!r} has invalid type"
            )
        if isinstance(value, int):
            return value
        if isinstance(value, float) and value.is_integer():
            return int(value)
        if isinstance(value, str):
            try:
                return int(value.strip(), 10)
            except ValueError as exc:
                raise ValueError(
                    f"Qdrant collection '{collection}' metadata {field!r} "
                    "is not an integer"
                ) from exc
        raise ValueError(
            f"Qdrant collection '{collection}' metadata {field!r} has invalid type"
        )

    def _validate_existing_embedding_contract(
        self, collection: str, info: qdrant_models.CollectionInfo
    ) -> None:
        raw = info.config.metadata
        if raw is None:
            meta = {}
        elif isinstance(raw, Mapping):
            meta = dict(raw)
        else:
            raise ValueError(
                f"Qdrant collection '{collection}' has unsupported metadata type "
                f"{type(raw).__name__}"
            )
        dim_key = QDRANT_META_EMBEDDING_DIMENSION
        model_key = QDRANT_META_EMBEDDING_MODEL
        if dim_key not in meta or model_key not in meta:
            raise EmbeddingContractMismatchError(
                f"Qdrant collection '{collection}' is missing OntoCast "
                f"embedding metadata ({dim_key!r}, {model_key!r}). "
                "Drop and recreate the collection. " + _embedding_contract_help()
            )
        stored_dim = self._coerce_metadata_int(
            meta[dim_key], field=dim_key, collection=collection
        )
        stored_model = meta[model_key]
        if not isinstance(stored_model, str):
            raise ValueError(
                f"Qdrant collection '{collection}' metadata {model_key!r} "
                "must be a string"
            )
        expected_meta_dim = self._metadata_embedding_dimension()
        if stored_dim != expected_meta_dim or not self._embedding_fingerprint_matches(
            stored_model
        ):
            raise EmbeddingContractMismatchError(
                f"Qdrant collection '{collection}' embedding contract mismatch: "
                f"collection has dimension={stored_dim}, model={stored_model!r}; "
                f"current config expects dimension={expected_meta_dim}, "
                f"model={self._embedding_model_fingerprint()!r}. "
                + _embedding_contract_help()
            )

    def _vectors_and_sparse_for_create(
        self,
    ) -> tuple[
        dict[str, qdrant_models.VectorParams],
        dict[str, qdrant_models.SparseVectorParams],
    ]:
        distance = self.config.distance
        dense_dim = self._dense_dimension()
        vectors: dict[str, qdrant_models.VectorParams] = {
            CORE_VECTOR_NAME: qdrant_models.VectorParams(
                size=dense_dim, distance=distance
            ),
            NEIGHBORHOOD_VECTOR_NAME: qdrant_models.VectorParams(
                size=dense_dim, distance=distance
            ),
        }
        # BM25 sparse scoring on plain dot product (no sparse modifier).
        sparse: dict[str, qdrant_models.SparseVectorParams] = {
            BM25_VECTOR_NAME: qdrant_models.SparseVectorParams(modifier=None)
        }
        return (vectors, sparse)

    def _validate_collection_vector_layout(
        self, collection: str, info: qdrant_models.CollectionInfo
    ) -> None:
        distance = self.config.distance
        dense_dim = self._dense_dimension()
        params = info.config.params
        raw_vectors = params.vectors
        vectors_map: dict[str, qdrant_models.VectorParams] = (
            dict(raw_vectors) if isinstance(raw_vectors, dict) else {}
        )
        raw_sparse = params.sparse_vectors
        sparse_map: dict[str, qdrant_models.SparseVectorParams] = (
            dict(raw_sparse) if isinstance(raw_sparse, dict) else {}
        )

        def _require_dense(name: str) -> None:
            if name not in vectors_map:
                raise ValueError(
                    f"Qdrant collection '{collection}' missing dense vector {name!r}; "
                    f"have dense keys {set(vectors_map.keys())}"
                )
            cfg = vectors_map[name]
            if cfg.size != dense_dim:
                raise EmbeddingContractMismatchError(
                    f"Qdrant collection '{collection}' vector {name!r} size "
                    f"{cfg.size} does not match configured dense size {dense_dim}. "
                    + _embedding_contract_help()
                )
            if cfg.distance != distance:
                raise ValueError(
                    f"Qdrant collection '{collection}' vector {name!r} "
                    f"uses distance {cfg.distance!r}; config expects {distance!r}."
                )

        _require_dense(CORE_VECTOR_NAME)
        _require_dense(NEIGHBORHOOD_VECTOR_NAME)

        bm25_cfg = sparse_map.get(BM25_VECTOR_NAME)
        if bm25_cfg is None:
            raise ValueError(
                f"Qdrant collection '{collection}' missing sparse vector "
                f"{BM25_VECTOR_NAME!r}; have sparse keys {set(sparse_map.keys())}"
            )
        if bm25_cfg.modifier is not None:
            raise ValueError(
                f"Qdrant collection '{collection}' sparse vector {BM25_VECTOR_NAME!r} "
                f"uses modifier {bm25_cfg.modifier!r}; expected no modifier "
                "(dot-product sparse scoring). Recreate the collection."
            )

    def _ensure_named_vector_collection(self, collection: str) -> None:
        metadata_dim = self._metadata_embedding_dimension()
        embedding_meta = self._collection_embedding_metadata(metadata_dim)
        vectors_cfg, sparse_cfg = self._vectors_and_sparse_for_create()
        if not self.client.collection_exists(collection_name=collection):
            self.client.create_collection(
                collection_name=collection,
                vectors_config=vectors_cfg,
                sparse_vectors_config=sparse_cfg,
                metadata=embedding_meta,
            )
            logger.info(
                "Created Qdrant collection '%s' metadata_dim=%s distance=%s model=%s",
                collection,
                metadata_dim,
                self.config.distance.value,
                embedding_meta[QDRANT_META_EMBEDDING_MODEL],
            )
        else:
            info = self.client.get_collection(collection_name=collection)
            self._validate_collection_vector_layout(collection, info)
            self._validate_existing_embedding_contract(collection, info)

    def index_ontology(self, ontology: Ontology) -> int:
        """Atomize + embed + upsert ontology neighborhoods."""
        atoms = self.atomizer.atomize(source=ontology, depth=1)
        if not atoms:
            return 0
        core_texts = [atom.core_representation for atom in atoms]
        neighborhood_texts = [atom.neighborhood_representation for atom in atoms]
        minimal_texts = [atom.minimal_representation for atom in atoms]

        core_vectors = self._embed_texts_batched(core_texts)
        neighborhood_vectors = self._embed_texts_batched(neighborhood_texts)
        bm25_vectors = self._embed_texts_batched_sparse(minimal_texts)

        if len(core_vectors) != len(atoms) or len(neighborhood_vectors) != len(atoms):
            raise ValueError(
                "Embedding provider returned mismatched vector counts for atoms"
            )
        if len(bm25_vectors) != len(atoms):
            raise ValueError(
                "BM25 embedder returned mismatched sparse vector counts for atoms"
            )

        points: list[qdrant_models.PointStruct] = []
        for i, atom in enumerate(atoms):
            vec_map: dict[str, Any] = {
                CORE_VECTOR_NAME: core_vectors[i],
                NEIGHBORHOOD_VECTOR_NAME: neighborhood_vectors[i],
                BM25_VECTOR_NAME: bm25_vectors[i],
            }
            points.append(
                qdrant_models.PointStruct(
                    id=self._point_id_for_atom(atom),
                    vector=vec_map,
                    payload=self._atom_payload(atom),
                )
            )
        collection = self._ontology_collection_name()
        for points_batch in self._iter_batches(points, self.config.upsert_batch_size):
            self.client.upsert(collection_name=collection, points=points_batch)
        return len(points)

    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 atoms by text query using weighted multi-vector fusion."""
        core_q, neigh_q, bm25_q = self._encode_single_query_vectors(query)
        return self.search_by_vector(
            core_vector=core_q,
            neighborhood_vector=neigh_q,
            bm25_query_vector=bm25_q,
            top_k=top_k,
            filter_iri=filter_iri,
            filter_version=filter_version,
            filter_hash=filter_hash,
        )

    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."""
        core_q, neigh_q, bm25_q = self._encode_single_query_vectors(query)
        channel_hits = self.search_hits_by_vector(
            core_vector=core_q,
            neighborhood_vector=neigh_q,
            bm25_query_vector=bm25_q,
            top_k=top_k,
            filter_iri=filter_iri,
            filter_version=filter_version,
            filter_hash=filter_hash,
        )
        eff_top_k = self._effective_top_k(top_k)
        return self._rank_fuse_channel_hits(
            channel_hits.core_hits,
            channel_hits.neighborhood_hits,
            channel_hits.bm25_hits,
            limit=eff_top_k,
        )

    def _search_patch_hits_for_query_triples(
        self,
        triples: list[tuple[list[float], list[float], qdrant_models.SparseVector]],
        top_k: int,
        filter_iri: str | None,
        filter_version: str | None,
        filter_hash: str | None,
    ) -> list[OntologySearchHitsByChannel]:
        """Run split-channel search per query (dense core/neighborhood + BM25)."""
        if not triples:
            return []

        def search_one(
            t: tuple[list[float], list[float], qdrant_models.SparseVector],
        ) -> OntologySearchHitsByChannel:
            core_v, neigh_v, bm25_v = t
            return self.search_hits_by_vector(
                core_vector=core_v,
                neighborhood_vector=neigh_v,
                bm25_query_vector=bm25_v,
                top_k=top_k,
                filter_iri=filter_iri,
                filter_version=filter_version,
                filter_hash=filter_hash,
            )

        workers = min(32, len(triples))
        with ThreadPoolExecutor(max_workers=workers) as pool:
            return list(pool.map(search_one, triples))

    def _search_patch_hits_many_impl(
        self,
        queries: list[str],
        top_k: int | None,
        filter_iri: str | None,
        filter_version: str | None,
        filter_hash: str | None,
    ) -> list[OntologySearchHitsByChannel]:
        if not queries:
            return []

        eff_top_k = self._effective_top_k(top_k)
        triples = self._encode_query_vectors_batch(queries)
        return self._search_patch_hits_for_query_triples(
            triples,
            eff_top_k,
            filter_iri,
            filter_version,
            filter_hash,
        )

    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."""
        return self._search_patch_hits_many_impl(
            queries,
            top_k,
            filter_iri,
            filter_version,
            filter_hash,
        )

    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: one batched embed, then parallel split-channel searches."""
        if not queries:
            return []
        eff_top_k = self._effective_top_k(top_k)
        triples = await asyncio.to_thread(self._encode_query_vectors_batch, queries)
        tasks = [
            asyncio.to_thread(
                self.search_hits_by_vector,
                core_v,
                neigh_v,
                bm25_v,
                eff_top_k,
                filter_iri,
                filter_version,
                filter_hash,
            )
            for core_v, neigh_v, bm25_v in triples
        ]
        return await asyncio.gather(*tasks)

    def _parse_dense_vector(self, raw: Any) -> list[float] | None:
        if isinstance(raw, list):
            if not raw or not all(isinstance(v, int | float) for v in raw):
                return None
            return [float(v) for v in cast(list[int | float], raw)]
        return None

    def fetch_vectors(
        self,
        atom_ids: list[str],
    ) -> dict[str, tuple[list[float], list[float]]]:
        """Batch-fetch dense core/neighborhood vectors for MMR (BM25 not used)."""
        if not atom_ids:
            return {}
        point_id_to_atom_id = {self._point_id(atom_id): atom_id for atom_id in atom_ids}
        points = self.client.retrieve(
            collection_name=self._ontology_collection_name(),
            ids=list(point_id_to_atom_id.keys()),
            with_vectors=True,
            with_payload=False,
        )
        out: dict[str, tuple[list[float], list[float]]] = {}
        for point in points:
            atom_id = point_id_to_atom_id.get(str(point.id))
            if atom_id is None:
                continue
            point_vector = point.vector
            if not isinstance(point_vector, dict):
                continue
            core_raw = point_vector.get(CORE_VECTOR_NAME)
            neighborhood_raw = point_vector.get(NEIGHBORHOOD_VECTOR_NAME)
            core = self._parse_dense_vector(core_raw)
            neighborhood = self._parse_dense_vector(neighborhood_raw)
            if core is None or neighborhood is None:
                continue
            out[atom_id] = (core, neighborhood)
        return out

    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 search_by_vector(
        self,
        core_vector: list[float],
        neighborhood_vector: list[float],
        bm25_query_vector: qdrant_models.SparseVector | None = None,
        top_k: int | None = None,
        filter_iri: str | None = None,
        filter_version: str | None = None,
        filter_hash: str | None = None,
    ) -> list[GraphAtom]:
        """Search ontology atoms with rank fusion over named vectors."""
        channel_hits = self.search_hits_by_vector(
            core_vector=core_vector,
            neighborhood_vector=neighborhood_vector,
            bm25_query_vector=bm25_query_vector,
            top_k=top_k,
            filter_iri=filter_iri,
            filter_version=filter_version,
            filter_hash=filter_hash,
        )
        eff_top_k = self._effective_top_k(top_k)
        fused_hits = self._rank_fuse_channel_hits(
            channel_hits.core_hits,
            channel_hits.neighborhood_hits,
            channel_hits.bm25_hits,
            limit=eff_top_k,
        )
        return [hit.atom for hit in fused_hits]

    def search_hits_by_vector(
        self,
        core_vector: list[float],
        neighborhood_vector: list[float],
        bm25_query_vector: qdrant_models.SparseVector | None = None,
        top_k: int | None = None,
        filter_iri: str | None = None,
        filter_version: str | None = None,
        filter_hash: str | None = None,
    ) -> OntologySearchHitsByChannel:
        """Search ontology atoms and return channel-separated scored hit objects."""
        eff_top_k = self._effective_top_k(top_k)
        self._require_embedding_vector_length(core_vector, role="Query core vector")
        self._require_embedding_vector_length(
            neighborhood_vector, role="Query neighborhood vector"
        )
        search_filter = self._build_filter(
            filter_iri=filter_iri,
            filter_version=filter_version,
            filter_hash=filter_hash,
        )
        core_hits = self._query_named_vector(
            vector_name=CORE_VECTOR_NAME,
            vector=core_vector,
            limit=eff_top_k,
            search_filter=search_filter,
        )
        neighborhood_hits = self._query_named_vector(
            vector_name=NEIGHBORHOOD_VECTOR_NAME,
            vector=neighborhood_vector,
            limit=eff_top_k,
            search_filter=search_filter,
        )
        bm25_hits_raw: list[Any] = []
        if bm25_query_vector is not None:
            bm25_hits_raw = self._query_named_vector(
                vector_name=BM25_VECTOR_NAME,
                vector=bm25_query_vector,
                limit=eff_top_k,
                search_filter=search_filter,
            )
        core_typed_hits = self._points_to_hits(core_hits)
        neighborhood_typed_hits = self._points_to_hits(
            neighborhood_hits, apply_neighborhood_empty_penalty=True
        )
        bm25_typed_hits = self._points_to_hits(bm25_hits_raw)
        if self.config.dedup_query_hits_by_iri:
            core_typed_hits = self._dedupe_hits_by_identity(core_typed_hits)
            neighborhood_typed_hits = self._dedupe_hits_by_identity(
                neighborhood_typed_hits
            )
            bm25_typed_hits = self._dedupe_hits_by_identity(bm25_typed_hits)
        return OntologySearchHitsByChannel(
            core_hits=core_typed_hits,
            neighborhood_hits=neighborhood_typed_hits,
            bm25_hits=bm25_typed_hits,
        )

    def _points_to_hits(
        self,
        points: list[Any],
        *,
        apply_neighborhood_empty_penalty: bool = False,
    ) -> list[OntologySearchHit]:
        hits: list[OntologySearchHit] = []
        for point in points:
            score = float(point.score) if point.score is not None else 0.0
            if apply_neighborhood_empty_penalty:
                payload = point.payload or {}
                neighborhood_text = str(payload.get("neighborhood_representation", ""))
                if (
                    neighborhood_text.strip().lower()
                    == "no neighborhood facts available"
                ):
                    score = 0.0
            atom = self._point_to_atom(point)
            atom.score = score
            hits.append(OntologySearchHit(atom=atom, score=score))
        return hits

    def _rank_fuse_channel_hits(
        self,
        core_hits: list[OntologySearchHit],
        neighborhood_hits: list[OntologySearchHit],
        bm25_hits: list[OntologySearchHit],
        *,
        limit: int,
    ) -> list[OntologySearchHit]:
        rank_scores: dict[str, float] = {}
        best_hit_by_id: dict[str, OntologySearchHit] = {}

        core_weight, neighborhood_weight, bm25_weight = (
            self._normalized_fusion_weights()
        )
        for rank, hit in enumerate(core_hits, start=1):
            atom_id = hit.atom.atom_id
            rank_scores[atom_id] = rank_scores.get(atom_id, 0.0) + (core_weight / rank)
            prev = best_hit_by_id.get(atom_id)
            if prev is None or hit.score > prev.score:
                best_hit_by_id[atom_id] = hit
        for rank, hit in enumerate(neighborhood_hits, start=1):
            atom_id = hit.atom.atom_id
            rank_scores[atom_id] = rank_scores.get(atom_id, 0.0) + (
                neighborhood_weight / rank
            )
            prev = best_hit_by_id.get(atom_id)
            if prev is None or hit.score > prev.score:
                best_hit_by_id[atom_id] = hit
        for rank, hit in enumerate(bm25_hits, start=1):
            atom_id = hit.atom.atom_id
            rank_scores[atom_id] = rank_scores.get(atom_id, 0.0) + (bm25_weight / rank)
            prev = best_hit_by_id.get(atom_id)
            if prev is None or hit.score > prev.score:
                best_hit_by_id[atom_id] = hit

        ranked_atom_ids = sorted(
            rank_scores.keys(),
            key=lambda atom_id: (
                rank_scores[atom_id],
                float(best_hit_by_id[atom_id].score),
                atom_id,
            ),
            reverse=True,
        )[:limit]
        out: list[OntologySearchHit] = []
        for atom_id in ranked_atom_ids:
            source_hit = best_hit_by_id[atom_id]
            atom = source_hit.atom.model_copy(update={"score": rank_scores[atom_id]})
            out.append(OntologySearchHit(atom=atom, score=rank_scores[atom_id]))
        return out

    def delete_ontology(
        self,
        iri: str,
        version: str | None = None,
        ontology_hash: str | None = None,
    ) -> None:
        """Delete atoms associated with one ontology IRI and optional version/hash."""
        delete_filter = self._build_filter(
            filter_iri=iri, filter_version=version, filter_hash=ontology_hash
        )
        if delete_filter is None:
            return
        self.client.delete(
            collection_name=self._ontology_collection_name(),
            points_selector=qdrant_models.FilterSelector(filter=delete_filter),
        )

    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 _build_filter(
        self,
        filter_iri: str | None = None,
        filter_version: str | None = None,
        filter_hash: str | None = None,
    ) -> qdrant_models.Filter | None:
        conditions: list[qdrant_models.Condition] = []
        if filter_iri is not None:
            conditions.append(
                qdrant_models.FieldCondition(
                    key="ontology_iri", match=qdrant_models.MatchValue(value=filter_iri)
                )
            )
        if filter_version is not None:
            conditions.append(
                qdrant_models.FieldCondition(
                    key="ontology_version",
                    match=qdrant_models.MatchValue(value=filter_version),
                )
            )
        if filter_hash is not None:
            conditions.append(
                qdrant_models.FieldCondition(
                    key="ontology_hash",
                    match=qdrant_models.MatchValue(value=filter_hash),
                )
            )
        if not conditions:
            return None
        return qdrant_models.Filter(must=conditions)

    def _point_to_atom(self, point: Any) -> GraphAtom:
        payload = point.payload or {}
        created_at_raw = payload.get("created_at")
        created_at = self._parse_created_at(created_at_raw)
        return GraphAtom(
            atom_id=str(payload.get("atom_id", point.id)),
            ontology_iri=str(payload.get("ontology_iri", "")),
            ontology_id=payload.get("ontology_id"),
            ontology_hash=payload.get("ontology_hash"),
            ontology_version=payload.get("ontology_version"),
            iri=str(payload.get("iri", "")),
            entity_role=canonicalize_entity_role(payload.get("entity_role")),
            core_representation=str(payload.get("core_representation", "")),
            minimal_representation=str(payload.get("minimal_representation", "")),
            neighborhood_representation=str(
                payload.get("neighborhood_representation", "")
            ),
            created_at=created_at,
            score=float(point.score) if point.score is not None else None,
        )

    def _atom_payload(self, atom: GraphAtom) -> dict[str, Any]:
        return {
            "atom_id": atom.atom_id,
            "ontology_iri": atom.ontology_iri,
            "ontology_id": atom.ontology_id,
            "ontology_hash": atom.ontology_hash,
            "ontology_version": atom.ontology_version,
            "iri": atom.iri,
            "entity_role": canonicalize_entity_role(atom.entity_role),
            "core_representation": atom.core_representation,
            "minimal_representation": atom.minimal_representation,
            "neighborhood_representation": atom.neighborhood_representation,
            "created_at": atom.created_at.isoformat(),
        }

    def _parse_created_at(self, value: Any) -> datetime:
        if isinstance(value, datetime):
            return value
        if isinstance(value, str):
            try:
                return datetime.fromisoformat(value.replace("Z", "+00:00"))
            except ValueError:
                pass
        return datetime.now(timezone.utc)

    def _effective_top_k(self, top_k: int | None) -> int:
        """Resolve retrieval depth: explicit ``top_k`` overrides :attr:`QdrantConfig.top_k`."""
        if top_k is not None:
            return top_k
        return self.config.top_k

    def _require_embedding_vector_length(
        self,
        vector: list[float],
        *,
        role: str,
    ) -> None:
        expected = self._dense_dimension()
        if len(vector) != expected:
            raise EmbeddingContractMismatchError(
                f"{role} vector length {len(vector)} does not match the configured "
                f"collection embedding dimension {expected}. "
                + _embedding_contract_help()
            )

    def _point_id(self, atom_id: str) -> str:
        """Return a Qdrant-compatible point id (UUID string)."""
        try:
            return str(uuid.UUID(atom_id))
        except ValueError:
            return str(uuid.uuid5(uuid.NAMESPACE_URL, atom_id))

    def _point_id_for_atom(self, atom: GraphAtom) -> str:
        if self.config.dedup_mode == QdrantDedupMode.ATOM_ID:
            return self._point_id(atom.atom_id)
        return self._point_id(self._identity_key_for_atom(atom))

    def _identity_key_for_atom(self, atom: GraphAtom) -> str:
        if self.config.dedup_mode == QdrantDedupMode.ATOM_ID:
            return atom.atom_id
        parts: list[str] = [
            atom.ontology_iri or "",
            atom.iri or "",
        ]
        if self.config.dedup_include_version:
            parts.append(atom.ontology_version or "")
        if self.config.dedup_include_hash:
            parts.append(atom.ontology_hash or "")
        return "|".join(parts)

    def _dedupe_hits_by_identity(
        self, hits: list[OntologySearchHit]
    ) -> list[OntologySearchHit]:
        if not hits:
            return []
        best_by_key: dict[str, OntologySearchHit] = {}
        order_index: dict[str, int] = {}
        for index, hit in enumerate(hits):
            key = self._identity_key_for_atom(hit.atom)
            previous = best_by_key.get(key)
            if previous is None:
                best_by_key[key] = hit
                order_index[key] = index
                continue
            if float(hit.score) > float(previous.score):
                best_by_key[key] = hit
        deduped = list(best_by_key.values())
        deduped.sort(
            key=lambda h: (
                -float(h.score),
                order_index[self._identity_key_for_atom(h.atom)],
            )
        )
        return deduped

    def delete_duplicate_iri_points(self, *, batch_size: int = 512) -> int:
        """Delete duplicate points sharing the same configured identity key.

        Keeps the first point for each key encountered in collection order.
        Intended as a one-off cleanup for collections created before strict dedup mode.
        """
        collection_name = self._ontology_collection_name()
        seen_by_key: dict[str, qdrant_models.ExtendedPointId] = {}
        duplicate_ids: list[qdrant_models.ExtendedPointId] = []
        offset: Any = None
        while True:
            points, next_offset = self.client.scroll(
                collection_name=collection_name,
                with_payload=True,
                with_vectors=False,
                offset=offset,
                limit=batch_size,
            )
            if not points:
                break
            for point in points:
                atom = self._point_to_atom(point)
                key = self._identity_key_for_atom(atom)
                if key in seen_by_key:
                    duplicate_ids.append(point.id)
                else:
                    seen_by_key[key] = point.id
            if next_offset is None:
                break
            offset = next_offset
        if not duplicate_ids:
            return 0
        self.client.delete(
            collection_name=collection_name,
            points_selector=qdrant_models.PointIdsList(points=duplicate_ids),
        )
        return len(duplicate_ids)

    def _query_named_vector(
        self,
        vector_name: str,
        vector: ChannelVector,
        limit: int,
        search_filter: qdrant_models.Filter | None,
    ) -> list[Any]:
        response = self.client.query_points(
            collection_name=self._ontology_collection_name(),
            query=vector,
            using=vector_name,
            query_filter=search_filter,
            with_payload=True,
            limit=limit,
        )
        return response.points

    def _embed_texts_batched(self, texts: list[str]) -> list[list[float]]:
        if not texts:
            return []
        vectors: list[list[float]] = []
        for batch in self._iter_batches(texts, self.config.embedding_batch_size):
            batch_vectors = self.embedding.embed(batch)
            if len(batch_vectors) != len(batch):
                raise ValueError(
                    "Embedding provider returned mismatched vectors for batch"
                )
            for j, vec in enumerate(batch_vectors):
                self._require_embedding_vector_length(
                    vec,
                    role=f"Index embedding batch offset {len(vectors) + j}",
                )
            vectors.extend(batch_vectors)
        return vectors

    def _embed_texts_batched_sparse(
        self, texts: list[str]
    ) -> list[qdrant_models.SparseVector]:
        if not texts:
            return []
        out: list[qdrant_models.SparseVector] = []
        sparse_tool = self._require_sparse_embedding_tool()
        for batch in self._iter_batches(texts, self.config.embedding_batch_size):
            batch_vectors = sparse_tool.embed_sparse(batch)
            if len(batch_vectors) != len(batch):
                raise ValueError(
                    "BM25 embedder returned mismatched sparse vectors for batch"
                )
            out.extend(batch_vectors)
        return out

    def _iter_batches(self, items: list[Any], batch_size: int) -> list[list[Any]]:
        batches: list[list[Any]] = []
        for index in range(0, len(items), batch_size):
            batches.append(items[index : index + batch_size])
        return batches

    def _ensure_payload_index(self, collection_name: str, field_name: str) -> None:
        try:
            self.client.create_payload_index(
                collection_name=collection_name,
                field_name=field_name,
                field_schema=qdrant_models.PayloadSchemaType.KEYWORD,
            )
        except Exception:
            logger.debug(
                "Qdrant payload index '%s' on '%s' already exists",
                field_name,
                collection_name,
            )

afetch_vectors(atom_ids) async

Async wrapper around :meth:fetch_vectors.

Source code in ontocast/tool/vector_store/qdrant.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)

Point config at collections for tenant / project.

Call :meth:initialize after.

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

def apply_tenancy(
    self,
    tenant: str,
    project: str,
    *,
    sep: str = TENANCY_SEP,
) -> None:
    """Point config at collections for ``tenant`` / ``project``.

    Call :meth:`initialize` after.
    """
    t, p = tenant.strip(), project.strip()
    self.config.ontology_collection = tenant_project_ontologies_name(t, p, sep=sep)
    self.config.facts_collection = tenant_project_facts_name(t, p, sep=sep)

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

Async variant: one batched embed, then parallel split-channel searches.

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

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: one batched embed, then parallel split-channel searches."""
    if not queries:
        return []
    eff_top_k = self._effective_top_k(top_k)
    triples = await asyncio.to_thread(self._encode_query_vectors_batch, queries)
    tasks = [
        asyncio.to_thread(
            self.search_hits_by_vector,
            core_v,
            neigh_v,
            bm25_v,
            eff_top_k,
            filter_iri,
            filter_version,
            filter_hash,
        )
        for core_v, neigh_v, bm25_v in triples
    ]
    return await asyncio.gather(*tasks)

clean_tenancy(tenant, project, *, sep=TENANCY_SEP) async

Delete Qdrant collections named for tenant / project.

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

async def clean_tenancy(
    self,
    tenant: str,
    project: str,
    *,
    sep: str = TENANCY_SEP,
) -> None:
    """Delete Qdrant collections named for ``tenant`` / ``project``."""
    t, p = tenant.strip(), project.strip()
    for name in (
        tenant_project_ontologies_name(t, p, sep=sep),
        tenant_project_facts_name(t, p, sep=sep),
    ):
        if self.client.collection_exists(collection_name=name):
            self.client.delete_collection(collection_name=name)
            logger.info("Deleted Qdrant collection %s", name)

delete_duplicate_iri_points(*, batch_size=512)

Delete duplicate points sharing the same configured identity key.

Keeps the first point for each key encountered in collection order. Intended as a one-off cleanup for collections created before strict dedup mode.

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

def delete_duplicate_iri_points(self, *, batch_size: int = 512) -> int:
    """Delete duplicate points sharing the same configured identity key.

    Keeps the first point for each key encountered in collection order.
    Intended as a one-off cleanup for collections created before strict dedup mode.
    """
    collection_name = self._ontology_collection_name()
    seen_by_key: dict[str, qdrant_models.ExtendedPointId] = {}
    duplicate_ids: list[qdrant_models.ExtendedPointId] = []
    offset: Any = None
    while True:
        points, next_offset = self.client.scroll(
            collection_name=collection_name,
            with_payload=True,
            with_vectors=False,
            offset=offset,
            limit=batch_size,
        )
        if not points:
            break
        for point in points:
            atom = self._point_to_atom(point)
            key = self._identity_key_for_atom(atom)
            if key in seen_by_key:
                duplicate_ids.append(point.id)
            else:
                seen_by_key[key] = point.id
        if next_offset is None:
            break
        offset = next_offset
    if not duplicate_ids:
        return 0
    self.client.delete(
        collection_name=collection_name,
        points_selector=qdrant_models.PointIdsList(points=duplicate_ids),
    )
    return len(duplicate_ids)

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

Delete atoms associated with one ontology IRI and optional version/hash.

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

def delete_ontology(
    self,
    iri: str,
    version: str | None = None,
    ontology_hash: str | None = None,
) -> None:
    """Delete atoms associated with one ontology IRI and optional version/hash."""
    delete_filter = self._build_filter(
        filter_iri=iri, filter_version=version, filter_hash=ontology_hash
    )
    if delete_filter is None:
        return
    self.client.delete(
        collection_name=self._ontology_collection_name(),
        points_selector=qdrant_models.FilterSelector(filter=delete_filter),
    )

fetch_vectors(atom_ids)

Batch-fetch dense core/neighborhood vectors for MMR (BM25 not used).

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

def fetch_vectors(
    self,
    atom_ids: list[str],
) -> dict[str, tuple[list[float], list[float]]]:
    """Batch-fetch dense core/neighborhood vectors for MMR (BM25 not used)."""
    if not atom_ids:
        return {}
    point_id_to_atom_id = {self._point_id(atom_id): atom_id for atom_id in atom_ids}
    points = self.client.retrieve(
        collection_name=self._ontology_collection_name(),
        ids=list(point_id_to_atom_id.keys()),
        with_vectors=True,
        with_payload=False,
    )
    out: dict[str, tuple[list[float], list[float]]] = {}
    for point in points:
        atom_id = point_id_to_atom_id.get(str(point.id))
        if atom_id is None:
            continue
        point_vector = point.vector
        if not isinstance(point_vector, dict):
            continue
        core_raw = point_vector.get(CORE_VECTOR_NAME)
        neighborhood_raw = point_vector.get(NEIGHBORHOOD_VECTOR_NAME)
        core = self._parse_dense_vector(core_raw)
        neighborhood = self._parse_dense_vector(neighborhood_raw)
        if core is None or neighborhood is None:
            continue
        out[atom_id] = (core, neighborhood)
    return out

index_ontology(ontology)

Atomize + embed + upsert ontology neighborhoods.

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

def index_ontology(self, ontology: Ontology) -> int:
    """Atomize + embed + upsert ontology neighborhoods."""
    atoms = self.atomizer.atomize(source=ontology, depth=1)
    if not atoms:
        return 0
    core_texts = [atom.core_representation for atom in atoms]
    neighborhood_texts = [atom.neighborhood_representation for atom in atoms]
    minimal_texts = [atom.minimal_representation for atom in atoms]

    core_vectors = self._embed_texts_batched(core_texts)
    neighborhood_vectors = self._embed_texts_batched(neighborhood_texts)
    bm25_vectors = self._embed_texts_batched_sparse(minimal_texts)

    if len(core_vectors) != len(atoms) or len(neighborhood_vectors) != len(atoms):
        raise ValueError(
            "Embedding provider returned mismatched vector counts for atoms"
        )
    if len(bm25_vectors) != len(atoms):
        raise ValueError(
            "BM25 embedder returned mismatched sparse vector counts for atoms"
        )

    points: list[qdrant_models.PointStruct] = []
    for i, atom in enumerate(atoms):
        vec_map: dict[str, Any] = {
            CORE_VECTOR_NAME: core_vectors[i],
            NEIGHBORHOOD_VECTOR_NAME: neighborhood_vectors[i],
            BM25_VECTOR_NAME: bm25_vectors[i],
        }
        points.append(
            qdrant_models.PointStruct(
                id=self._point_id_for_atom(atom),
                vector=vec_map,
                payload=self._atom_payload(atom),
            )
        )
    collection = self._ontology_collection_name()
    for points_batch in self._iter_batches(points, self.config.upsert_batch_size):
        self.client.upsert(collection_name=collection, points=points_batch)
    return len(points)

initialize() async

Create ontology/facts collections and payload indexes if missing.

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

async def initialize(self) -> None:
    """Create ontology/facts collections and payload indexes if missing."""
    ontology_col = self.config.ontology_collection
    facts_col = self.config.facts_collection
    assert ontology_col is not None
    assert facts_col is not None
    self._ensure_named_vector_collection(ontology_col)
    self._ensure_named_vector_collection(facts_col)

    self._ensure_payload_index(
        collection_name=ontology_col, field_name="ontology_iri"
    )
    self._ensure_payload_index(
        collection_name=ontology_col, field_name="ontology_version"
    )
    self._ensure_payload_index(
        collection_name=ontology_col, field_name="ontology_hash"
    )
    self._ensure_payload_index(collection_name=ontology_col, field_name="iri")

reindex_ontology(ontology)

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

Source code in ontocast/tool/vector_store/qdrant.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_by_vector(core_vector, neighborhood_vector, bm25_query_vector=None, top_k=None, filter_iri=None, filter_version=None, filter_hash=None)

Search ontology atoms with rank fusion over named vectors.

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

def search_by_vector(
    self,
    core_vector: list[float],
    neighborhood_vector: list[float],
    bm25_query_vector: qdrant_models.SparseVector | None = None,
    top_k: int | None = None,
    filter_iri: str | None = None,
    filter_version: str | None = None,
    filter_hash: str | None = None,
) -> list[GraphAtom]:
    """Search ontology atoms with rank fusion over named vectors."""
    channel_hits = self.search_hits_by_vector(
        core_vector=core_vector,
        neighborhood_vector=neighborhood_vector,
        bm25_query_vector=bm25_query_vector,
        top_k=top_k,
        filter_iri=filter_iri,
        filter_version=filter_version,
        filter_hash=filter_hash,
    )
    eff_top_k = self._effective_top_k(top_k)
    fused_hits = self._rank_fuse_channel_hits(
        channel_hits.core_hits,
        channel_hits.neighborhood_hits,
        channel_hits.bm25_hits,
        limit=eff_top_k,
    )
    return [hit.atom for hit in fused_hits]

search_hits_by_vector(core_vector, neighborhood_vector, bm25_query_vector=None, top_k=None, filter_iri=None, filter_version=None, filter_hash=None)

Search ontology atoms and return channel-separated scored hit objects.

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

def search_hits_by_vector(
    self,
    core_vector: list[float],
    neighborhood_vector: list[float],
    bm25_query_vector: qdrant_models.SparseVector | None = None,
    top_k: int | None = None,
    filter_iri: str | None = None,
    filter_version: str | None = None,
    filter_hash: str | None = None,
) -> OntologySearchHitsByChannel:
    """Search ontology atoms and return channel-separated scored hit objects."""
    eff_top_k = self._effective_top_k(top_k)
    self._require_embedding_vector_length(core_vector, role="Query core vector")
    self._require_embedding_vector_length(
        neighborhood_vector, role="Query neighborhood vector"
    )
    search_filter = self._build_filter(
        filter_iri=filter_iri,
        filter_version=filter_version,
        filter_hash=filter_hash,
    )
    core_hits = self._query_named_vector(
        vector_name=CORE_VECTOR_NAME,
        vector=core_vector,
        limit=eff_top_k,
        search_filter=search_filter,
    )
    neighborhood_hits = self._query_named_vector(
        vector_name=NEIGHBORHOOD_VECTOR_NAME,
        vector=neighborhood_vector,
        limit=eff_top_k,
        search_filter=search_filter,
    )
    bm25_hits_raw: list[Any] = []
    if bm25_query_vector is not None:
        bm25_hits_raw = self._query_named_vector(
            vector_name=BM25_VECTOR_NAME,
            vector=bm25_query_vector,
            limit=eff_top_k,
            search_filter=search_filter,
        )
    core_typed_hits = self._points_to_hits(core_hits)
    neighborhood_typed_hits = self._points_to_hits(
        neighborhood_hits, apply_neighborhood_empty_penalty=True
    )
    bm25_typed_hits = self._points_to_hits(bm25_hits_raw)
    if self.config.dedup_query_hits_by_iri:
        core_typed_hits = self._dedupe_hits_by_identity(core_typed_hits)
        neighborhood_typed_hits = self._dedupe_hits_by_identity(
            neighborhood_typed_hits
        )
        bm25_typed_hits = self._dedupe_hits_by_identity(bm25_typed_hits)
    return OntologySearchHitsByChannel(
        core_hits=core_typed_hits,
        neighborhood_hits=neighborhood_typed_hits,
        bm25_hits=bm25_typed_hits,
    )

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

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

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

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."""
    core_q, neigh_q, bm25_q = self._encode_single_query_vectors(query)
    channel_hits = self.search_hits_by_vector(
        core_vector=core_q,
        neighborhood_vector=neigh_q,
        bm25_query_vector=bm25_q,
        top_k=top_k,
        filter_iri=filter_iri,
        filter_version=filter_version,
        filter_hash=filter_hash,
    )
    eff_top_k = self._effective_top_k(top_k)
    return self._rank_fuse_channel_hits(
        channel_hits.core_hits,
        channel_hits.neighborhood_hits,
        channel_hits.bm25_hits,
        limit=eff_top_k,
    )

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

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

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

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."""
    return self._search_patch_hits_many_impl(
        queries,
        top_k,
        filter_iri,
        filter_version,
        filter_hash,
    )

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

Search ontology atoms by text query using weighted multi-vector fusion.

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

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 atoms by text query using weighted multi-vector fusion."""
    core_q, neigh_q, bm25_q = self._encode_single_query_vectors(query)
    return self.search_by_vector(
        core_vector=core_q,
        neighborhood_vector=neigh_q,
        bm25_query_vector=bm25_q,
        top_k=top_k,
        filter_iri=filter_iri,
        filter_version=filter_version,
        filter_hash=filter_hash,
    )

VectorStoreTool

Bases: Tool

Abstract interface for vector store implementations.

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

class VectorStoreTool(Tool):
    """Abstract interface for vector store implementations."""

    @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 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 supports_tenancy_partition(self) -> bool:
        """True if :meth:`clean_tenancy` clears isolated collections for (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"
        )

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

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

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

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 :meth:clean_tenancy clears isolated collections for (tenant, project).

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

def supports_tenancy_partition(self) -> bool:
    """True if :meth:`clean_tenancy` clears isolated collections for (tenant, project)."""
    return False