ontocast.tool.vector_store

Vector store package for ontology patch retrieval.

EmbeddingContractMismatchError

Bases: ValueError

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

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

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

EmbeddingTool

Bases: Tool

Base embedding tool with provider-specific implementations.

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

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

    config: EmbeddingConfig = Field(default_factory=EmbeddingConfig)

    @abc.abstractmethod
    def embed(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 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_verb = self._normalize_uri(prop)  # bare verb for SPO
            ranges = self._collect_range_labels(
                prop, graph, max_items=endpoint_label_cap
            )
            for r_label in ranges or ["something"]:
                clues.append(f"it {prop_verb} {r_label}")
            self._append_inverse_of_clues_for_property(prop, graph, clues)

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

    def _build_minimal_representation(self, entity: URIRef) -> 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):
                siblings = sorted(
                    (
                        sib
                        for sib in parent_to_children.get(par, set[URIRef]())
                        if sib != entity
                    ),
                    key=str,
                )
                for sib in siblings[:6]:
                    clues.append(
                        f"{self._normalize_uri(sib)} is also a kind of "
                        f"{self._parent_resource_phrase(graph, par)}"
                    )

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

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

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

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

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

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

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

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

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

            pred_phrase = self._normalize_uri(pred)

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

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

        return sorted(set(clues))

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

    def _collect_literals(
        self, graph: RDFGraph, subject: URIRef, predicates: list[URIRef], max_items: int
    ) -> list[str]:
        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]

LanceDBVectorStoreManager

Bases: VectorStoreManager

Stores ontology atoms in a single embedded LanceDB database directory.

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

class LanceDBVectorStoreManager(VectorStoreManager):
    """Stores ontology atoms in a single embedded LanceDB database directory."""

    store_config: VectorStoreConfig = Field(default_factory=VectorStoreConfig)
    lancedb_config: LanceDBConfig = Field(default_factory=LanceDBConfig)
    embedding: EmbeddingTool = Field(..., exclude=True)
    sparse_embedding: FastembedBm25SparseTool | None = Field(default=None, exclude=True)
    atomizer: GraphAtomizer = Field(default_factory=GraphAtomizer, exclude=True)

    @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 _data_dir(self) -> Path:
        return Path(self.lancedb_config.data_dir).expanduser().resolve()

    def _connect(self):
        _require_lancedb()
        import lancedb

        data_dir = self._data_dir()
        data_dir.mkdir(parents=True, exist_ok=True)
        return lancedb.connect(str(data_dir))

    def _ontology_table_name(self) -> str:
        name = self.lancedb_config.ontology_table
        if name is None:
            raise ValueError(
                "LanceDB ontology_table is unset; ensure LanceDBConfig validation"
                " ran or call apply_tenancy before vector operations"
            )
        return name

    def _facts_table_name(self) -> str:
        name = self.lancedb_config.facts_table
        if name is None:
            raise ValueError(
                "LanceDB facts_table is unset; ensure LanceDBConfig validation ran"
            )
        return name

    def _meta_path(self, table_name: str | None = None) -> Path:
        table = table_name or self._ontology_table_name()
        meta_dir = self._data_dir() / _META_DIRNAME
        meta_dir.mkdir(parents=True, exist_ok=True)
        return meta_dir / f"{table}.json"

    def supports_tenancy_partition(self) -> bool:
        return True

    def apply_tenancy(
        self,
        tenant: str,
        project: str,
        *,
        sep: str = TENANCY_SEP,
    ) -> None:
        """Switch active Lance tables for ``tenant`` / ``project``.

        Same naming as Qdrant collections; all tables live in one embedded DB.
        Call :meth:`initialize` after.
        """
        t, p = tenant.strip(), project.strip()
        ontology_name = tenant_project_ontologies_name(t, p, sep=sep)
        facts_name = tenant_project_facts_name(t, p, sep=sep)
        self.lancedb_config.ontology_table = ontology_name
        self.lancedb_config.facts_table = facts_name
        self.store_config.ontology_table = ontology_name
        self.store_config.facts_table = facts_name

    async def clean_tenancy(
        self,
        tenant: str,
        project: str,
        *,
        sep: str = TENANCY_SEP,
    ) -> None:
        """Drop Lance tables (and embedding metadata) for ``tenant`` / ``project``."""
        t, p = tenant.strip(), project.strip()
        table_names = (
            tenant_project_ontologies_name(t, p, sep=sep),
            tenant_project_facts_name(t, p, sep=sep),
        )
        db = self._connect()
        existing = self._list_tables(db)
        for name in table_names:
            if name in existing:
                db.drop_table(name)
                logger.info("Dropped LanceDB table %s", name)
            meta = self._meta_path(name)
            if meta.exists():
                meta.unlink()

    def _dense_dimension(self) -> int:
        return self.embedding_config.dimension

    def _write_embedding_meta(self) -> None:
        meta = collection_embedding_metadata(
            self.embedding_config,
            metadata_dim=self._dense_dimension(),
        )
        self._meta_path().write_text(json.dumps(meta), encoding="utf-8")

    def _read_embedding_meta(self) -> dict[str, Any]:
        path = self._meta_path()
        if not path.exists():
            return {}
        return json.loads(path.read_text(encoding="utf-8"))

    def _validate_embedding_meta(self) -> None:
        table = self._ontology_table_name()
        validate_embedding_contract_metadata(
            table,
            self._read_embedding_meta() or None,
            embedding_config=self.embedding_config,
            expected_meta_dim=self._dense_dimension(),
        )

    async def initialize(self) -> None:
        """Create ontology/facts tables and indexes if missing."""
        await asyncio.to_thread(self._initialize_sync)

    def _list_tables(self, db: Any) -> set[str]:
        list_tables = getattr(db, "list_tables", None)
        if callable(list_tables):
            response = list_tables()
            if hasattr(response, "tables"):
                raw = response.tables
                if raw and isinstance(raw[0], str):
                    return set(raw)
                return {str(item) for item in raw}
            if isinstance(response, list):
                return {str(item) for item in response}
            return set()
        return set(db.table_names())

    def _initialize_sync(self) -> None:
        db = self._connect()
        ontology_name = self._ontology_table_name()
        tables = self._list_tables(db)

        if ontology_name in tables:
            self._validate_embedding_meta()
            table = db.open_table(ontology_name)
            self._ensure_indexes(table)
        elif not self._meta_path().exists():
            self._write_embedding_meta()

    def _ensure_indexes(self, table: Any) -> None:
        try:
            table.create_index(metric="cosine", vector_column_name="core_vector")
        except Exception:
            logger.debug("LanceDB core_vector index already exists or skipped")
        try:
            table.create_index(
                metric="cosine", vector_column_name="neighborhood_vector"
            )
        except Exception:
            logger.debug("LanceDB neighborhood_vector index already exists or skipped")
        try:
            table.create_fts_index("minimal_representation")
        except Exception:
            logger.debug("LanceDB FTS index already exists or skipped")

    def _record_from_atom(
        self,
        atom: GraphAtom,
        core_vector: list[float],
        neighborhood_vector: list[float],
    ) -> dict[str, Any]:
        payload = atom_payload(atom)
        payload["point_id"] = point_id_for_atom(atom, store_config=self.store_config)
        payload["core_vector"] = core_vector
        payload["neighborhood_vector"] = neighborhood_vector
        return payload

    def index_ontology(self, ontology: Ontology) -> int:
        atoms = self.atomizer.atomize(source=ontology, depth=1)
        if not atoms:
            return 0

        core_vectors = self._embed_texts_batched(
            [atom.core_representation for atom in atoms]
        )
        neighborhood_vectors = self._embed_texts_batched(
            [atom.neighborhood_representation for atom in atoms]
        )

        records = [
            self._record_from_atom(atom, core_vectors[i], neighborhood_vectors[i])
            for i, atom in enumerate(atoms)
        ]

        db = self._connect()
        table_name = self._ontology_table_name()
        tables = self._list_tables(db)
        if table_name not in tables:
            db.create_table(table_name, data=records)
            self._write_embedding_meta()
            table = db.open_table(table_name)
            self._ensure_indexes(table)
            return len(records)

        table = db.open_table(table_name)
        table.merge_insert(
            "point_id"
        ).when_matched_update_all().when_not_matched_insert_all().execute(  # type: ignore[attr-defined]
            records
        )
        return len(records)

    def _encode_single_query_vectors(
        self, query: str
    ) -> tuple[list[float], list[float], str]:
        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], str]]:
        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):
            require_embedding_vector_length(
                vec,
                role=f"Query embedding[{i}]",
                expected=self._dense_dimension(),
            )
        return [(dense_vecs[i], dense_vecs[i], queries[i]) for i in range(n)]

    def _filter_clause(
        self,
        *,
        filter_iri: str | None,
        filter_version: str | None,
        filter_hash: str | None,
    ) -> str | None:
        parts: list[str] = []
        if filter_iri is not None:
            parts.append(
                f"ontology_iri = '{filter_iri.replace(chr(39), chr(39) + chr(39))}'"
            )
        if filter_version is not None:
            parts.append(
                f"ontology_version = '{filter_version.replace(chr(39), chr(39) + chr(39))}'"
            )
        if filter_hash is not None:
            parts.append(
                f"ontology_hash = '{filter_hash.replace(chr(39), chr(39) + chr(39))}'"
            )
        if not parts:
            return None
        return " AND ".join(parts)

    def _row_to_hit(
        self,
        row: dict[str, Any],
        *,
        score: float,
        apply_neighborhood_empty_penalty: bool = False,
    ) -> OntologySearchHit:
        if apply_neighborhood_empty_penalty:
            neighborhood_text = str(row.get("neighborhood_representation", ""))
            if neighborhood_text.strip().lower() == "no neighborhood facts available":
                score = 0.0
        atom = atom_from_payload(row, score=score)
        return OntologySearchHit(atom=atom, score=score)

    def _search_dense_channel(
        self,
        vector: list[float],
        *,
        vector_column: str,
        limit: int,
        where: str | None,
    ) -> list[OntologySearchHit]:
        db = self._connect()
        table_name = self._ontology_table_name()
        tables = self._list_tables(db)
        if table_name not in tables:
            return []
        table = db.open_table(table_name)
        builder = table.search(vector, vector_column_name=vector_column).limit(limit)
        if where:
            builder = builder.where(where)
        rows = builder.to_list()
        hits: list[OntologySearchHit] = []
        for row in rows:
            distance = float(row.get("_distance", 0.0))
            score = 1.0 - distance if distance <= 1.0 else 1.0 / (1.0 + distance)
            hits.append(
                self._row_to_hit(
                    row,
                    score=score,
                    apply_neighborhood_empty_penalty=(
                        vector_column == "neighborhood_vector"
                    ),
                )
            )
        return hits

    def _search_bm25_channel(
        self,
        query: str,
        *,
        limit: int,
        where: str | None,
    ) -> list[OntologySearchHit]:
        db = self._connect()
        table_name = self._ontology_table_name()
        tables = self._list_tables(db)
        if table_name not in tables:
            return []
        table = db.open_table(table_name)
        builder = table.search(query, query_type="fts").limit(limit)
        if where:
            builder = builder.where(where)
        rows = builder.to_list()
        hits: list[OntologySearchHit] = []
        for row in rows:
            score = float(row.get("_score", row.get("score", 0.0)))
            hits.append(self._row_to_hit(row, score=score))
        return hits

    def search_hits_by_vector(
        self,
        core_vector: list[float],
        neighborhood_vector: list[float],
        bm25_query: str | None = None,
        top_k: int | None = None,
        filter_iri: str | None = None,
        filter_version: str | None = None,
        filter_hash: str | None = None,
    ) -> OntologySearchHitsByChannel:
        eff_top_k = effective_top_k(self.store_config, top_k)
        where = self._filter_clause(
            filter_iri=filter_iri,
            filter_version=filter_version,
            filter_hash=filter_hash,
        )
        core_hits = self._search_dense_channel(
            core_vector,
            vector_column="core_vector",
            limit=eff_top_k,
            where=where,
        )
        neighborhood_hits = self._search_dense_channel(
            neighborhood_vector,
            vector_column="neighborhood_vector",
            limit=eff_top_k,
            where=where,
        )
        bm25_hits: list[OntologySearchHit] = []
        if bm25_query is not None:
            bm25_hits = self._search_bm25_channel(
                bm25_query, limit=eff_top_k, where=where
            )
        if self.store_config.dedup_query_hits_by_iri:
            core_hits = dedupe_hits_by_identity(
                core_hits, store_config=self.store_config
            )
            neighborhood_hits = dedupe_hits_by_identity(
                neighborhood_hits, store_config=self.store_config
            )
            bm25_hits = dedupe_hits_by_identity(
                bm25_hits, store_config=self.store_config
            )
        return OntologySearchHitsByChannel(
            core_hits=core_hits,
            neighborhood_hits=neighborhood_hits,
            bm25_hits=bm25_hits,
        )

    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]:
        hits = self.search_patch_hits(
            query=query,
            top_k=top_k,
            filter_iri=filter_iri,
            filter_version=filter_version,
            filter_hash=filter_hash,
        )
        return [hit.atom for hit in hits]

    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]:
        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=bm25_q,
            top_k=top_k,
            filter_iri=filter_iri,
            filter_version=filter_version,
            filter_hash=filter_hash,
        )
        eff_top_k = effective_top_k(self.store_config, top_k)
        cw, nw, bw = normalized_fusion_weights(self.store_config)
        return rank_fuse_channel_hits(
            channel_hits.core_hits,
            channel_hits.neighborhood_hits,
            channel_hits.bm25_hits,
            core_weight=cw,
            neighborhood_weight=nw,
            bm25_weight=bw,
            limit=eff_top_k,
        )

    def _search_patch_hits_for_query_triples(
        self,
        triples: list[tuple[list[float], list[float], str]],
        top_k: int,
        filter_iri: str | None,
        filter_version: str | None,
        filter_hash: str | None,
    ) -> list[OntologySearchHitsByChannel]:
        if not triples:
            return []

        def search_one(
            t: tuple[list[float], list[float], str],
        ) -> OntologySearchHitsByChannel:
            core_v, neigh_v, bm25_q = t
            return self.search_hits_by_vector(
                core_vector=core_v,
                neighborhood_vector=neigh_v,
                bm25_query=bm25_q,
                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(
        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]:
        if not queries:
            return []
        eff_top_k = effective_top_k(self.store_config, 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,
        )

    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]:
        if not queries:
            return []
        eff_top_k = effective_top_k(self.store_config, 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_q,
                eff_top_k,
                filter_iri,
                filter_version,
                filter_hash,
            )
            for core_v, neigh_v, bm25_q in triples
        ]
        return await asyncio.gather(*tasks)

    def fetch_vectors(
        self,
        atom_ids: list[str],
    ) -> dict[str, tuple[list[float], list[float]]]:
        if not atom_ids:
            return {}
        db = self._connect()
        table_name = self._ontology_table_name()
        tables = self._list_tables(db)
        if table_name not in tables:
            return {}
        table = db.open_table(table_name)
        out: dict[str, tuple[list[float], list[float]]] = {}
        for atom_id in atom_ids:
            escaped = atom_id.replace("'", "''")
            rows = table.search().where(f"atom_id = '{escaped}'").limit(1).to_list()
            if not rows:
                continue
            row = rows[0]
            core = row.get("core_vector")
            neighborhood = row.get("neighborhood_vector")
            if isinstance(core, list) and isinstance(neighborhood, list):
                out[atom_id] = (
                    [float(v) for v in core],
                    [float(v) for v in neighborhood],
                )
        return out

    def delete_ontology(
        self,
        iri: str,
        version: str | None = None,
        ontology_hash: str | None = None,
    ) -> None:
        where = self._filter_clause(
            filter_iri=iri,
            filter_version=version,
            filter_hash=ontology_hash,
        )
        if where is None:
            return
        db = self._connect()
        table_name = self._ontology_table_name()
        tables = self._list_tables(db)
        if table_name not in tables:
            return
        table = db.open_table(table_name)
        table.delete(where)

    def _embed_texts_batched(self, texts: list[str]) -> list[list[float]]:
        if not texts:
            return []
        vectors: list[list[float]] = []
        for batch in iter_batches(texts, self.store_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):
                require_embedding_vector_length(
                    vec,
                    role=f"Index embedding batch offset {len(vectors) + j}",
                    expected=self._dense_dimension(),
                )
            vectors.extend(batch_vectors)
        return vectors

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

Switch active Lance tables for tenant / project.

Same naming as Qdrant collections; all tables live in one embedded DB. Call :meth:initialize after.

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

def apply_tenancy(
    self,
    tenant: str,
    project: str,
    *,
    sep: str = TENANCY_SEP,
) -> None:
    """Switch active Lance tables for ``tenant`` / ``project``.

    Same naming as Qdrant collections; all tables live in one embedded DB.
    Call :meth:`initialize` after.
    """
    t, p = tenant.strip(), project.strip()
    ontology_name = tenant_project_ontologies_name(t, p, sep=sep)
    facts_name = tenant_project_facts_name(t, p, sep=sep)
    self.lancedb_config.ontology_table = ontology_name
    self.lancedb_config.facts_table = facts_name
    self.store_config.ontology_table = ontology_name
    self.store_config.facts_table = facts_name

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

Drop Lance tables (and embedding metadata) for tenant / project.

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

async def clean_tenancy(
    self,
    tenant: str,
    project: str,
    *,
    sep: str = TENANCY_SEP,
) -> None:
    """Drop Lance tables (and embedding metadata) for ``tenant`` / ``project``."""
    t, p = tenant.strip(), project.strip()
    table_names = (
        tenant_project_ontologies_name(t, p, sep=sep),
        tenant_project_facts_name(t, p, sep=sep),
    )
    db = self._connect()
    existing = self._list_tables(db)
    for name in table_names:
        if name in existing:
            db.drop_table(name)
            logger.info("Dropped LanceDB table %s", name)
        meta = self._meta_path(name)
        if meta.exists():
            meta.unlink()

initialize() async

Create ontology/facts tables and indexes if missing.

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

async def initialize(self) -> None:
    """Create ontology/facts tables and indexes if missing."""
    await asyncio.to_thread(self._initialize_sync)

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: VectorStoreManager = Field(exclude=True)
    sparql_tool: Any | None = Field(default=None, exclude=True)
    patch: PatchRetrievalConfig = Field(
        default_factory=PatchRetrievalConfig,
        exclude=True,
    )
    _last_retrieval_metrics: dict[str, Any] = PrivateAttr(default_factory=dict)

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

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

    def 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."""
        self._last_retrieval_metrics = {}
        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,
        )
        sc = self.vector_store.store_config
        pc = self.patch
        merged = _filter_and_merge_patch_hits(
            hits_by_query,
            store_config=sc,
            patch_config=pc,
            per_query_core_score_ratio=pc.per_query_core_score_ratio,
            per_query_neighborhood_score_ratio=pc.per_query_neighborhood_score_ratio,
            per_query_bm25_score_ratio=pc.per_query_bm25_score_ratio,
            min_core_query_best_score=pc.min_core_query_best_score,
            min_neighborhood_query_best_score=pc.min_neighborhood_query_best_score,
            min_bm25_query_best_score=pc.min_bm25_query_best_score,
            min_merged_max_score=pc.min_merged_max_score,
        )
        atoms_after_merge = len(merged)
        merged = [atom for atom in merged if not _is_ontology_declaration_atom(atom)]

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

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

        if not merged:
            self._last_retrieval_metrics = {
                "query_count": len(queries),
                "top_k": eff_top_k,
                "atoms_after_merge": atoms_after_merge,
                "atoms_final": 0,
            }
            return RDFGraph(), []

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

        self._last_retrieval_metrics = {
            "query_count": len(queries),
            "top_k": eff_top_k,
            "merge_mode": pc.cross_query_merge_mode.value,
            "atoms_after_merge": atoms_after_merge,
            "atoms_final": len(merged),
            "source_ontology_iris": source_iris,
            "seeds_by_ontology": dict(seeds_by_ontology),
        }

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

        entity_uris, entity_relevance, entity_roles = _ranked_entity_weights(merged)
        hit_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
                )

        ontology_iris = hit_ontology_iris
        if self.sparql_tool.triple_store_manager is not None:
            catalog = await self.sparql_tool.triple_store_manager.afetch_ontologies()
            ontology_iris = _expand_ontology_iris_by_reference(
                entity_uris,
                hit_ontology_iris,
                catalog,
            )
            expanded = sorted(set(ontology_iris) - set(hit_ontology_iris))
            if expanded:
                self._last_retrieval_metrics["expanded_ontology_iris"] = expanded

        hub_seed_count = sc.induced_subgraph_hub_seed_count
        ancestor_depth = sc.induced_subgraph_ancestor_closure_depth

        graph = await self.sparql_tool.aget_induced_subgraph(
            entity_uris=entity_uris,
            entity_relevance=entity_relevance,
            entity_roles=entity_roles,
            ontology_iris=ontology_iris,
            depth=subgraph_depth,
            max_total_triples=max_total_triples,
            estimated_triples_per_query=estimated_triples_per_query,
            ontology_version_filters=ontology_version_filters or None,
            ontology_hash_filters=ontology_hash_filters or None,
            hub_seed_count=hub_seed_count,
            ancestor_closure_depth=ancestor_depth,
        )
        self._last_retrieval_metrics["snapshot_triple_count"] = len(graph)
        self._last_retrieval_metrics["ontology_iris_for_expansion"] = ontology_iris
        self._last_retrieval_metrics.update(self.sparql_tool.last_finalize_metrics)

        _bind_common_vocab_prefixes(graph)
        return graph, source_iris

aretrieve(query, top_k=None, expand_sparql=True, subgraph_depth=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.

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."""
    self._last_retrieval_metrics = {}
    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,
    )
    sc = self.vector_store.store_config
    pc = self.patch
    merged = _filter_and_merge_patch_hits(
        hits_by_query,
        store_config=sc,
        patch_config=pc,
        per_query_core_score_ratio=pc.per_query_core_score_ratio,
        per_query_neighborhood_score_ratio=pc.per_query_neighborhood_score_ratio,
        per_query_bm25_score_ratio=pc.per_query_bm25_score_ratio,
        min_core_query_best_score=pc.min_core_query_best_score,
        min_neighborhood_query_best_score=pc.min_neighborhood_query_best_score,
        min_bm25_query_best_score=pc.min_bm25_query_best_score,
        min_merged_max_score=pc.min_merged_max_score,
    )
    atoms_after_merge = len(merged)
    merged = [atom for atom in merged if not _is_ontology_declaration_atom(atom)]

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

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

    if not merged:
        self._last_retrieval_metrics = {
            "query_count": len(queries),
            "top_k": eff_top_k,
            "atoms_after_merge": atoms_after_merge,
            "atoms_final": 0,
        }
        return RDFGraph(), []

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

    self._last_retrieval_metrics = {
        "query_count": len(queries),
        "top_k": eff_top_k,
        "merge_mode": pc.cross_query_merge_mode.value,
        "atoms_after_merge": atoms_after_merge,
        "atoms_final": len(merged),
        "source_ontology_iris": source_iris,
        "seeds_by_ontology": dict(seeds_by_ontology),
    }

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

    entity_uris, entity_relevance, entity_roles = _ranked_entity_weights(merged)
    hit_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
            )

    ontology_iris = hit_ontology_iris
    if self.sparql_tool.triple_store_manager is not None:
        catalog = await self.sparql_tool.triple_store_manager.afetch_ontologies()
        ontology_iris = _expand_ontology_iris_by_reference(
            entity_uris,
            hit_ontology_iris,
            catalog,
        )
        expanded = sorted(set(ontology_iris) - set(hit_ontology_iris))
        if expanded:
            self._last_retrieval_metrics["expanded_ontology_iris"] = expanded

    hub_seed_count = sc.induced_subgraph_hub_seed_count
    ancestor_depth = sc.induced_subgraph_ancestor_closure_depth

    graph = await self.sparql_tool.aget_induced_subgraph(
        entity_uris=entity_uris,
        entity_relevance=entity_relevance,
        entity_roles=entity_roles,
        ontology_iris=ontology_iris,
        depth=subgraph_depth,
        max_total_triples=max_total_triples,
        estimated_triples_per_query=estimated_triples_per_query,
        ontology_version_filters=ontology_version_filters or None,
        ontology_hash_filters=ontology_hash_filters or None,
        hub_seed_count=hub_seed_count,
        ancestor_closure_depth=ancestor_depth,
    )
    self._last_retrieval_metrics["snapshot_triple_count"] = len(graph)
    self._last_retrieval_metrics["ontology_iris_for_expansion"] = ontology_iris
    self._last_retrieval_metrics.update(self.sparql_tool.last_finalize_metrics)

    _bind_common_vocab_prefixes(graph)
    return graph, source_iris

retrieve(query, top_k=None, expand_sparql=True, subgraph_depth=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,
            api_key=api_key,
            base_url=self.config.base_url,
        )

QdrantVectorStoreManager

Bases: VectorStoreManager

Stores ontology atoms in Qdrant and supports similarity lookup.

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

class QdrantVectorStoreManager(VectorStoreManager):
    """Stores ontology atoms in Qdrant and supports similarity lookup."""

    store_config: VectorStoreConfig = Field(default_factory=VectorStoreConfig)
    qdrant_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)]

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

    def _ontology_collection_name(self) -> str:
        name = self.qdrant_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.qdrant_config.ontology_collection
        facts_col = self.qdrant_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()
        ontology_name = tenant_project_ontologies_name(t, p, sep=sep)
        facts_name = tenant_project_facts_name(t, p, sep=sep)
        self.qdrant_config.ontology_collection = ontology_name
        self.qdrant_config.facts_collection = facts_name
        self.store_config.ontology_table = ontology_name
        self.store_config.facts_table = facts_name

    def _dense_dimension(self) -> int:
        return self.qdrant_config.vector_size or self.embedding_config.dimension

    def _metadata_embedding_dimension(self) -> int:
        return self._dense_dimension()

    def _validate_existing_embedding_contract(
        self, collection: str, info: qdrant_models.CollectionInfo
    ) -> None:
        raw = info.config.metadata
        if raw is None:
            meta: dict[str, Any] = {}
        elif isinstance(raw, dict):
            meta = dict(raw)
        else:
            raise ValueError(
                f"Qdrant collection '{collection}' has unsupported metadata type "
                f"{type(raw).__name__}"
            )
        validate_embedding_contract_metadata(
            collection,
            meta,
            embedding_config=self.embedding_config,
            expected_meta_dim=self._metadata_embedding_dimension(),
        )

    def _vectors_and_sparse_for_create(
        self,
    ) -> tuple[
        dict[str, qdrant_models.VectorParams],
        dict[str, qdrant_models.SparseVectorParams],
    ]:
        distance = self.qdrant_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
            ),
        }
        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.qdrant_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(backend="Qdrant collection")
                )
            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 = collection_embedding_metadata(
            self.embedding_config,
            metadata_dim=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.qdrant_config.distance.value,
                embedding_meta[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=point_id_for_atom(atom, store_config=self.store_config),
                    vector=vec_map,
                    payload=atom_payload(atom),
                )
            )
        collection = self._ontology_collection_name()
        for points_batch in iter_batches(points, self.qdrant_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 = effective_top_k(self.store_config, top_k)
        cw, nw, bw = normalized_fusion_weights(self.store_config)
        return rank_fuse_channel_hits(
            channel_hits.core_hits,
            channel_hits.neighborhood_hits,
            channel_hits.bm25_hits,
            core_weight=cw,
            neighborhood_weight=nw,
            bm25_weight=bw,
            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]:
        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 = effective_top_k(self.store_config, 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 = effective_top_k(self.store_config, 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 = {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

    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 = effective_top_k(self.store_config, top_k)
        cw, nw, bw = normalized_fusion_weights(self.store_config)
        fused_hits = rank_fuse_channel_hits(
            channel_hits.core_hits,
            channel_hits.neighborhood_hits,
            channel_hits.bm25_hits,
            core_weight=cw,
            neighborhood_weight=nw,
            bm25_weight=bw,
            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 = effective_top_k(self.store_config, 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.store_config.dedup_query_hits_by_iri:
            core_typed_hits = dedupe_hits_by_identity(
                core_typed_hits, store_config=self.store_config
            )
            neighborhood_typed_hits = dedupe_hits_by_identity(
                neighborhood_typed_hits, store_config=self.store_config
            )
            bm25_typed_hits = dedupe_hits_by_identity(
                bm25_typed_hits, store_config=self.store_config
            )
        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 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 _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 {}
        score = float(point.score) if point.score is not None else None
        return atom_from_payload(
            payload,
            score=score,
            default_id=str(point.id),
        )

    def _require_embedding_vector_length(
        self,
        vector: list[float],
        *,
        role: str,
    ) -> None:
        require_embedding_vector_length(
            vector,
            role=role,
            expected=self._dense_dimension(),
        )

    def delete_duplicate_iri_points(self, *, batch_size: int = 512) -> int:
        """Delete duplicate points sharing the same configured identity key."""
        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 = identity_key_for_atom(atom, store_config=self.store_config)
                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 count_points_by_ontology_iri(self, *, batch_size: int = 512) -> dict[str, int]:
        """Count indexed atoms grouped by ``ontology_iri`` payload (diagnostics)."""
        collection_name = self._ontology_collection_name()
        counts: dict[str, int] = defaultdict(int)
        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:
                payload = point.payload or {}
                onto_iri = str(payload.get("ontology_iri", ""))
                if onto_iri:
                    counts[onto_iri] += 1
            if next_offset is None:
                break
            offset = next_offset
        return dict(counts)

    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 iter_batches(texts, self.store_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 iter_batches(texts, self.store_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 _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,
            )

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()
    ontology_name = tenant_project_ontologies_name(t, p, sep=sep)
    facts_name = tenant_project_facts_name(t, p, sep=sep)
    self.qdrant_config.ontology_collection = ontology_name
    self.qdrant_config.facts_collection = facts_name
    self.store_config.ontology_table = ontology_name
    self.store_config.facts_table = facts_name

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 = effective_top_k(self.store_config, 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)

count_points_by_ontology_iri(*, batch_size=512)

Count indexed atoms grouped by ontology_iri payload (diagnostics).

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

def count_points_by_ontology_iri(self, *, batch_size: int = 512) -> dict[str, int]:
    """Count indexed atoms grouped by ``ontology_iri`` payload (diagnostics)."""
    collection_name = self._ontology_collection_name()
    counts: dict[str, int] = defaultdict(int)
    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:
            payload = point.payload or {}
            onto_iri = str(payload.get("ontology_iri", ""))
            if onto_iri:
                counts[onto_iri] += 1
        if next_offset is None:
            break
        offset = next_offset
    return dict(counts)

delete_duplicate_iri_points(*, batch_size=512)

Delete duplicate points sharing the same configured identity key.

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."""
    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 = identity_key_for_atom(atom, store_config=self.store_config)
            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 = {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=point_id_for_atom(atom, store_config=self.store_config),
                vector=vec_map,
                payload=atom_payload(atom),
            )
        )
    collection = self._ontology_collection_name()
    for points_batch in iter_batches(points, self.qdrant_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.qdrant_config.ontology_collection
    facts_col = self.qdrant_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")

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 = effective_top_k(self.store_config, top_k)
    cw, nw, bw = normalized_fusion_weights(self.store_config)
    fused_hits = rank_fuse_channel_hits(
        channel_hits.core_hits,
        channel_hits.neighborhood_hits,
        channel_hits.bm25_hits,
        core_weight=cw,
        neighborhood_weight=nw,
        bm25_weight=bw,
        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 = effective_top_k(self.store_config, 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.store_config.dedup_query_hits_by_iri:
        core_typed_hits = dedupe_hits_by_identity(
            core_typed_hits, store_config=self.store_config
        )
        neighborhood_typed_hits = dedupe_hits_by_identity(
            neighborhood_typed_hits, store_config=self.store_config
        )
        bm25_typed_hits = dedupe_hits_by_identity(
            bm25_typed_hits, store_config=self.store_config
        )
    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 = effective_top_k(self.store_config, top_k)
    cw, nw, bw = normalized_fusion_weights(self.store_config)
    return rank_fuse_channel_hits(
        channel_hits.core_hits,
        channel_hits.neighborhood_hits,
        channel_hits.bm25_hits,
        core_weight=cw,
        neighborhood_weight=nw,
        bm25_weight=bw,
        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,
    )

VectorStoreManager

Bases: Tool

Abstract interface for vector store implementations.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

afetch_vectors(atom_ids) async

Async wrapper around :meth:fetch_vectors.

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

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

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

Switch the active tenant/project partition when supported.

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

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

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

Async variant of :meth:search_patch_hits_many.

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

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

clean_tenancy(tenant, project) async

Drop or empty vector collections derived from tenant / project.

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

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

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

Delete all indexed atoms for a specific ontology IRI.

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

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

fetch_vectors(atom_ids) abstractmethod

Batch-fetch dense core/neighborhood vectors for MMR.

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

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

index_ontology(ontology) abstractmethod

Index an ontology and return number of indexed atoms.

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

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

initialize() abstractmethod async

Prepare schema/collections in the backing vector store.

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

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

reindex_ontology(ontology)

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

supports_tenancy_partition()

True if tenancy hooks isolate data by tenant/project.

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

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

create_vector_store_manager(tool_config, embedding, sparse_embedding)

Return a vector store manager when a backend URI is configured.

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

def create_vector_store_manager(
    tool_config: ToolConfig,
    embedding: EmbeddingTool,
    sparse_embedding: FastembedBm25SparseTool,
) -> VectorStoreManager | None:
    """Return a vector store manager when a backend URI is configured."""
    if tool_config.qdrant.uri:
        q_vs = tool_config.qdrant.vector_size
        emb_dim = tool_config.embedding.dimension
        if q_vs is not None and q_vs != emb_dim:
            raise ValueError(
                "QdrantConfig.vector_size must match "
                "EmbeddingConfig.dimension when set "
                f"(got vector_size={q_vs}, embedding.dimension={emb_dim})"
            )
        return QdrantVectorStoreManager(
            store_config=tool_config.vector_store,
            qdrant_config=tool_config.qdrant,
            embedding=embedding,
            sparse_embedding=sparse_embedding,
        )
    if tool_config.lancedb.enabled:
        return LanceDBVectorStoreManager(
            store_config=tool_config.vector_store,
            lancedb_config=tool_config.lancedb,
            embedding=embedding,
            sparse_embedding=sparse_embedding,
        )
    return None