ontocast.tool.agg

Embedding-based aggregation pipeline for RDF content unit graphs.

Modules:

Name	Description
aggregate	Embedding-based RDF graph aggregator.
clustering	Embedding-based entity clustering for disambiguation.
entity_aligner	Global entity alignment across multiple RDF graphs.
match_common	Stateless helpers for RDF graph matching and evaluation.
match_derivation	Derive pairwise predicted↔gt entity matches from global clusters.
match_models	Shared models for RDF graph matching and evaluation.
normalizer	Entity normalization for disambiguation.
rewriter	Graph rewriting for entity disambiguation.
signatures	Literal/object signatures and schema harvesting for merge guards.
triple_evaluator	Evaluate predicted vs ground-truth RDF graphs given entity alignments.
unit_scope	Unit-scoped fact IRIs: cross-unit coreference as a merge decision.
uri_builder	URI construction with naming convention normalization.

Attributes

__all__ = ['EmbeddingBasedAggregator', 'EntityAligner', 'EntityAlignmentResult', 'EntityCluster', 'EntityMatch', 'EntityRole', 'GraphEntityMember', 'MatchMetrics', 'MatchRegime', 'TaggedGraph', 'TripleSetEvaluator', 'URIBuilder', 'apply_document_metadata_provenance', 'derive_pair_matches'] module-attribute

Classes

EmbeddingBasedAggregator

Main aggregator using embedding-based entity disambiguation.

Pipeline stages: 1. Entity normalisation (with semantic context) 2. Parallel embedding 3. Similarity-based clustering 4. Representative selection (prefer ontology, then simplicity) 5. URI normalisation (PascalCase/camelCase under DEFAULT_IRI) 6. Graph rewriting

ContentUnit types are handled as follows: - facts: entities under base_iri are normalised. - ontology: all other entities are considered ontology entities and preserved.

Source code in ontocast/tool/agg/aggregate.py

class EmbeddingBasedAggregator:
    """Main aggregator using embedding-based entity disambiguation.

    Pipeline stages:
    1. Entity normalisation (with semantic context)
    2. Parallel embedding
    3. Similarity-based clustering
    4. Representative selection (prefer ontology, then simplicity)
    5. URI normalisation (PascalCase/camelCase under DEFAULT_IRI)
    6. Graph rewriting

    ContentUnit types are handled as follows:
    - ``facts``: entities under ``base_iri`` are normalised.
    - ``ontology``: all other entities are considered ontology entities and preserved.
    """

    def __init__(
        self,
        config: AggregationConfig | None = None,
        *,
        add_sameas_links: bool = True,
        base_iri: str = DEFAULT_IRI,
        candidate_similarity_threshold: float | None = None,
    ):
        """Initialise the embedding-based aggregator.

        Every tunable lives on :class:`AggregationConfig`, so ``settings.py``
        stays the single source of their defaults rather than restating them in
        this signature and again at the call site.

        Args:
            config: Aggregation tunables. Defaults to :class:`AggregationConfig`,
                i.e. the environment-resolved settings.
            add_sameas_links: Whether to add ``owl:sameAs`` for merged entities.
                Not config-driven: callers choose it per use, and the entity
                aligner wants different behaviour from the pipeline.
            base_iri: Base IRI for fact entity URIs. Entities under this
                namespace are facts; everything else is treated as an ontology
                entity and left unchanged.
            candidate_similarity_threshold: Overrides the configured permissive
                candidate threshold. The entity aligner pins it to its own
                similarity threshold rather than the pipeline's.
        """
        cfg = config or AggregationConfig()

        self.base_iri = base_iri
        self.candidate_similarity_threshold = (
            cfg.candidate_similarity_threshold
            if candidate_similarity_threshold is None
            else candidate_similarity_threshold
        )
        self.lexical_label_jaccard = cfg.lexical_label_jaccard
        self.lexical_sequence_ratio = cfg.lexical_sequence_ratio
        self.lexical_token_jaccard = cfg.lexical_token_jaccard
        self.functional_min_empirical_support = cfg.functional_min_empirical_support
        self.sibling_guard_scope = str(cfg.sibling_guard_scope)
        self.literal_conflict_guard = cfg.literal_conflict_guard
        self.initials_distinct_guard = cfg.initials_distinct_guard
        self.natural_key_merge = cfg.natural_key_merge
        self.type_guard_untyped = str(cfg.type_guard_untyped)
        self.unit_scoped_fact_iris = cfg.unit_scoped_fact_iris
        if candidate_similarity_threshold is None:
            self._warn_inert_similarity_threshold(cfg)

        # Pipeline components (EntityClusterer imports sklearn/ST lazily).
        # The clusterer runs at the permissive candidate threshold: candidates
        # are validated symbolically afterwards, so there is exactly one
        # clustering threshold on this path.
        from .clustering import EntityClusterer

        self.normalizer = EntityNormalizer(facts_iri=self.base_iri)
        self.clusterer = EntityClusterer(
            embedding_model=cfg.embedding_model,
            similarity_threshold=self.candidate_similarity_threshold,
        )
        self.selector = ClusterRepresentativeSelector()
        self.uri_builder = URIBuilder(base_iri=self.base_iri)
        self.rewriter = GraphRewriter(
            add_sameas_links=add_sameas_links,
            blocked_sameas_namespaces=(self.base_iri,),
        )

    @staticmethod
    def _warn_inert_similarity_threshold(cfg: AggregationConfig) -> None:
        """Warn when the aligner threshold is tuned but the pipeline one is not.

        ``similarity_threshold`` drives only the cross-graph
        :class:`~ontocast.tool.agg.entity_aligner.EntityAligner`; this
        aggregator clusters and gates at ``candidate_similarity_threshold``.
        Setting the former away from its default while the latter stays at its
        default is the signature of someone tuning the wrong knob.
        """
        fields = AggregationConfig.model_fields
        if (
            cfg.similarity_threshold != fields["similarity_threshold"].default
            and cfg.candidate_similarity_threshold
            == fields["candidate_similarity_threshold"].default
        ):
            logger.warning(
                "AGG_SIMILARITY_THRESHOLD=%s is set, but the in-pipeline "
                "aggregator does not read it: it clusters and gates at "
                "AGG_CANDIDATE_SIMILARITY_THRESHOLD (still at its default %s). "
                "AGG_SIMILARITY_THRESHOLD is only the cross-graph aligner's "
                "fallback threshold (POST /match/entities, match-graphs, the "
                "ontocast_align_entities tool).",
                cfg.similarity_threshold,
                cfg.candidate_similarity_threshold,
            )

    @staticmethod
    def _entity_in_namespace(entity: URIRef, namespace: URIRef | str | None) -> bool:
        """Return True when *entity* is under the provided namespace."""
        if namespace is None:
            return False
        return is_in_namespace(str(entity), str(namespace), context="auto")

    def _is_fact_entity_in_unit(self, entity: URIRef, unit: ContentUnit) -> bool:
        """Classify whether an entity should be treated as a fact in this unit.

        Facts are entities in either:
        - the configured base facts namespace (``base_iri``), or
        - the unit document namespace (``unit.doc_iri``).
        """
        return self._entity_in_namespace(
            entity, self.base_iri
        ) or self._entity_in_namespace(entity, unit.doc_iri)

    @staticmethod
    def _is_standard_ontology_entity(entity: URIRef) -> bool:
        """Return True for entities from built-in standard RDF vocabularies."""
        entity_str = str(entity)
        return any(entity_str.startswith(prefix) for prefix in _STANDARD_NAMESPACES)

    def _build_known_ontology_entities(
        self, ontology_graph: RDFGraph | None
    ) -> set[URIRef]:
        """Build a set of known ontology entities from ontology and std vocabularies."""
        known_entities: set[URIRef] = set()

        if ontology_graph is not None:
            for s, p, o in ontology_graph:
                if isinstance(s, URIRef):
                    known_entities.add(s)
                if isinstance(p, URIRef):
                    known_entities.add(p)
                if isinstance(o, URIRef):
                    known_entities.add(o)

        return known_entities

    @staticmethod
    def _tokenize(text: str) -> set[str]:
        # Short tokens stay: initials and single-letter identifiers
        # ("company S." vs "company T.") are often the only distinguishing
        # mark, and dropping them made such labels compare identical.
        return set(label_tokens(text))

    @staticmethod
    def _role_key(representation: EntityRepresentation) -> str:
        role = (
            representation.role
            if representation.role is not None
            else EntityRole.INSTANCE
        )
        return str(role)

    @staticmethod
    def _jaccard(left: set[str], right: set[str]) -> float:
        if not left and not right:
            return 1.0
        union = left | right
        return len(left & right) / len(union)

    @staticmethod
    def _instance_like_local_name(entity: URIRef) -> str | None:
        """Return normalized local name when URI ends with numeric suffix."""
        local_name = normalize_uri_local_name(unscoped_iri(entity)).replace(" ", "")
        if not local_name:
            return None
        match = _INSTANCE_LOCAL_NAME_RE.match(local_name)
        if match is None:
            return None
        if len(match.group("stem")) < 3:
            return None
        return local_name

    def _are_roles_compatible(
        self,
        left: URIRef,
        right: URIRef,
        representations: dict[URIRef, EntityRepresentation],
    ) -> bool:
        left_rep = representations.get(left)
        right_rep = representations.get(right)
        if left_rep is None or right_rep is None:
            return False
        return self._role_key(left_rep) == self._role_key(right_rep)

    def _are_types_compatible(
        self,
        left: URIRef,
        right: URIRef,
        representations: dict[URIRef, EntityRepresentation],
    ) -> bool:
        left_rep = representations.get(left)
        right_rep = representations.get(right)
        if left_rep is None or right_rep is None:
            return False
        left_types = set(left_rep.types)
        right_types = set(right_rep.types)
        if not left_types or not right_types:
            if self.type_guard_untyped == "strict":
                # Strict mode fails a typed-vs-untyped pair closed; two
                # untyped entities stay comparable — there is no type
                # evidence in either direction.
                return not left_types and not right_types
            return True
        return bool(left_types & right_types)

    def _entity_label_values(self, rep: EntityRepresentation) -> set[str]:
        """Normalized name strings an entity is identified by.

        ``alt_labels`` (string literals from arbitrary domain predicates)
        stand in only when the entity carries no ``rdfs:label``: for a
        labeled entity they are payload, not names — an honorific or role
        literal shared by several people must not read as label agreement.
        """
        source = rep.labels if rep.labels else rep.alt_labels
        return {
            self.normalizer.normalize_string(label) for label in source if label.strip()
        }

    def _are_lexical_aliases(
        self,
        left: URIRef,
        right: URIRef,
        representations: dict[URIRef, EntityRepresentation],
    ) -> bool:
        left_rep = representations.get(left)
        right_rep = representations.get(right)
        if left_rep is None or right_rep is None:
            return False
        if left_rep.normal_form == right_rep.normal_form:
            return True

        left_instance_name = self._instance_like_local_name(left)
        right_instance_name = self._instance_like_local_name(right)
        if (
            left_instance_name is not None
            and right_instance_name is not None
            and left_instance_name == right_instance_name
        ):
            return True

        left_label_tokens = self._entity_label_values(left_rep)
        right_label_tokens = self._entity_label_values(right_rep)
        if left_label_tokens & right_label_tokens:
            return True

        # Abbreviation-aware tier: "baranov d" vs "dmitry baranov" alias when
        # every token of one label matches a token of the other exactly or as
        # a single-character initial, with at least one shared full token.
        if self._labels_alias_with_initials(left_label_tokens, right_label_tokens):
            return True

        # Guard-literal-bearing entities (measurements, dated events) are
        # individuated by their payload, not their phrasing: "PL red shift of
        # SL1" vs "PL red shift of SL2" share most tokens yet denote distinct
        # values. Only the exact tiers above may merge them. String literals
        # (names, descriptions) do not raise this bar — disjoint identifier
        # strings are handled by _have_conflicting_literals instead.
        if left_rep.has_guard_literal and right_rep.has_guard_literal:
            return False

        if left_label_tokens and right_label_tokens:
            max_label_overlap = 0.0
            for left_label in left_label_tokens:
                left_tokens = self._tokenize(left_label)
                for right_label in right_label_tokens:
                    right_tokens = self._tokenize(right_label)
                    overlap = self._jaccard(left_tokens, right_tokens)
                    max_label_overlap = max(max_label_overlap, overlap)
            if max_label_overlap >= self.lexical_label_jaccard:
                return True

        left_normalized = left_rep.normal_form.strip()
        right_normalized = right_rep.normal_form.strip()
        if left_normalized and right_normalized:
            if left_normalized != right_normalized and (
                left_normalized.startswith(f"{right_normalized} ")
                or right_normalized.startswith(f"{left_normalized} ")
            ):
                return False

        ratio = SequenceMatcher(
            None, left_rep.normal_form, right_rep.normal_form
        ).ratio()
        if ratio >= self.lexical_sequence_ratio:
            return True

        left_tokens = self._tokenize(left_rep.normal_form)
        right_tokens = self._tokenize(right_rep.normal_form)
        if len(left_tokens) >= 2 and len(right_tokens) >= 2:
            if self._jaccard(left_tokens, right_tokens) >= self.lexical_token_jaccard:
                return True

        return False

    # Thin delegates: the shared implementations live in ``signatures`` so the
    # validation gate can consult the same string-compatibility notion without
    # importing the aggregator.
    _tokens_alias_compatible = staticmethod(tokens_alias_compatible)
    _labels_alias_with_initials = staticmethod(labels_alias_with_initials)
    _string_values_compatible = staticmethod(string_values_compatible)

    @classmethod
    def _have_conflicting_literals(
        cls,
        left_rep: EntityRepresentation,
        right_rep: EntityRepresentation,
    ) -> bool:
        """Return True when the entities assert disjoint values per predicate.

        A shared predicate with two non-empty, disjoint canonical value sets
        (numeric/temporal) marks the entities as distinct individuals; overlap
        or one-sided values read as re-mention/enrichment and stay mergeable.
        String payloads (identifiers, codes) conflict only when NO cross-pair
        is compatible (equality, prefix, or initial-abbreviation) — "d" vs
        "dmitry" is a re-mention, "S-2024-001" vs "S-2024-002" is a conflict.
        """
        for predicate, left_values in left_rep.predicate_literals.items():
            right_values = right_rep.predicate_literals.get(predicate)
            if not right_values or not left_values:
                continue
            if left_values.isdisjoint(right_values):
                return True
        for predicate, left_strings in left_rep.predicate_string_literals.items():
            right_strings = right_rep.predicate_string_literals.get(predicate)
            if not right_strings or not left_strings:
                continue
            if not any(
                cls._string_values_compatible(left_value, right_value)
                for left_value in left_strings
                for right_value in right_strings
            ):
                return True
        return False

    @staticmethod
    def _have_conflicting_functional_objects(
        left_rep: EntityRepresentation,
        right_rep: EntityRepresentation,
        functional_predicates: set[URIRef],
    ) -> bool:
        """Return True when a max-1 object predicate points at disjoint IRIs.

        Catches conflicts invisible to value comparison — e.g. two "10"
        quantities whose ``qudt:unit`` objects are ``DEG_C`` vs ``KiloHZ``.
        """
        if not functional_predicates:
            return False
        for predicate, left_objects in left_rep.predicate_iri_objects.items():
            if predicate not in functional_predicates:
                continue
            right_objects = right_rep.predicate_iri_objects.get(predicate)
            if not right_objects or not left_objects:
                continue
            if left_objects.isdisjoint(right_objects):
                return True
        return False

    def _labels_confirm_identity(
        self,
        left: URIRef,
        right: URIRef,
        representations: dict[URIRef, EntityRepresentation],
    ) -> bool:
        """Exact or initials-aware label agreement strong enough to skip cosine."""
        left_rep = representations.get(left)
        right_rep = representations.get(right)
        if left_rep is None or right_rep is None:
            return False
        left_labels = self._entity_label_values(left_rep)
        right_labels = self._entity_label_values(right_rep)
        if not left_labels or not right_labels:
            return False
        if left_labels & right_labels:
            return True
        return self._labels_alias_with_initials(left_labels, right_labels)

    def _labels_mark_distinct_entities(
        self,
        left_rep: EntityRepresentation,
        right_rep: EntityRepresentation,
    ) -> bool:
        """Label pairs identical except for conflicting initials mark distinctness."""
        if not self.initials_distinct_guard:
            return False
        return labels_differ_only_by_initials(
            self._entity_label_values(left_rep),
            self._entity_label_values(right_rep),
        )

    def _pair_distinctness_veto(
        self,
        left: URIRef,
        right: URIRef,
        representations: dict[URIRef, EntityRepresentation],
        direct_relation_pairs: set[frozenset[URIRef]] | None = None,
        guard_context: MergeGuardContext | None = None,
    ) -> bool:
        """Positive evidence that *left* and *right* denote distinct entities.

        Unlike the absence of a lexical alias — which merely fails to support
        a merge — a veto is grounds to keep the pair apart in *any* identity
        cluster, including transitively: two entities that a guard separates
        must not end up merged through a chain of intermediate aliases.
        """
        pair = frozenset((left, right))
        # Direct relations and sibling pairs are recorded by name; the gate's
        # merge vetoes name scoped source IRIs. Both spellings are checked.
        name_pair = frozenset((unscoped_iri(left), unscoped_iri(right)))
        if direct_relation_pairs is not None and (
            pair in direct_relation_pairs or name_pair in direct_relation_pairs
        ):
            return True
        left_rep = representations.get(left)
        right_rep = representations.get(right)
        if guard_context is not None:
            if name_pair in guard_context.sibling_pairs:
                return True
            if left_rep is not None and right_rep is not None:
                if self.literal_conflict_guard and self._have_conflicting_literals(
                    left_rep, right_rep
                ):
                    return True
                if self._have_conflicting_functional_objects(
                    left_rep, right_rep, guard_context.functional_predicates
                ):
                    return True
        if left_rep is not None and right_rep is not None:
            if self._labels_mark_distinct_entities(left_rep, right_rep):
                return True
        if not self._are_roles_compatible(left, right, representations):
            return True
        if not self._are_types_compatible(left, right, representations):
            return True
        return False

    def _can_merge_as_identity(
        self,
        left: URIRef,
        right: URIRef,
        representations: dict[URIRef, EntityRepresentation],
        direct_relation_pairs: set[frozenset[URIRef]] | None = None,
        guard_context: MergeGuardContext | None = None,
        key_pairs: set[frozenset[URIRef]] | None = None,
    ) -> bool:
        if self._pair_distinctness_veto(
            left,
            right,
            representations,
            direct_relation_pairs=direct_relation_pairs,
            guard_context=guard_context,
        ):
            return False
        if key_pairs is not None and frozenset((left, right)) in key_pairs:
            # A shared value on a single-valued identifier-like predicate is
            # positive identity evidence in its own right; the guards above
            # still had their say.
            return True
        return self._are_lexical_aliases(left, right, representations)

    def _collect_natural_key_pairs(
        self,
        representations: dict[URIRef, EntityRepresentation],
        schema_functional_predicates: set[URIRef],
    ) -> set[frozenset[URIRef]]:
        """Instance pairs sharing a value on a single-valued identifier predicate.

        Every guard in this module is a veto; this is the one source of
        *positive* symbolic identity evidence: two instances asserting the
        same value for a predicate that behaves like an identifier (declared
        max-1 by the schema, or observed single-valued on every subject) are
        candidate re-mentions of one entity — "Application no. 36760/06" is
        the same case wherever its number appears. Pairs found here are still
        subject to all distinctness vetoes; string values only (dates and
        numbers are coordinates, not identifiers), short values only (prose
        payloads such as notes and descriptions are not keys), and values
        shared too widely are treated as generic rather than identifying.
        """
        instance_role = str(EntityRole.INSTANCE)
        by_predicate: dict[URIRef, dict[URIRef, set[str]]] = {}
        for entity, rep in representations.items():
            if self._role_key(rep) != instance_role:
                continue
            for predicate, values in rep.predicate_string_literals.items():
                filtered = {
                    value
                    for value in values
                    if 0 < len(value) <= _NATURAL_KEY_MAX_VALUE_LENGTH
                }
                if filtered:
                    by_predicate.setdefault(predicate, {})[entity] = filtered

        pairs: set[frozenset[URIRef]] = set()
        for predicate, entity_values in by_predicate.items():
            if predicate not in schema_functional_predicates:
                if len(entity_values) < self.functional_min_empirical_support:
                    continue
                if any(len(values) != 1 for values in entity_values.values()):
                    continue
            value_index: dict[str, list[URIRef]] = {}
            for entity, values in entity_values.items():
                for value in values:
                    value_index.setdefault(value, []).append(entity)
            for value, entities in value_index.items():
                if not 2 <= len(entities) <= _NATURAL_KEY_MAX_VALUE_ENTITIES:
                    continue
                for left, right in combinations(sorted(entities, key=str), 2):
                    pairs.add(frozenset((left, right)))
        return pairs

    @staticmethod
    def _merge_candidate_clusters_by_key_pairs(
        candidate_clusters: list[list[URIRef]],
        key_pairs: set[frozenset[URIRef]],
    ) -> list[list[URIRef]]:
        """Join candidate clusters bridged by a natural-key pair.

        Embedding clustering only proposes pairs that read alike; two mentions
        of one entity under different surface forms ("Application no. X" vs
        "Case A v. B") never co-cluster, so a key pair spanning two candidate
        clusters must pull them into one before symbolic validation — which
        still adjudicates every pair inside the joined cluster.
        """
        if not key_pairs:
            return candidate_clusters
        cluster_of: dict[URIRef, int] = {}
        for index, cluster in enumerate(candidate_clusters):
            for entity in cluster:
                cluster_of[entity] = index

        parent = list(range(len(candidate_clusters)))

        def find(index: int) -> int:
            while parent[index] != index:
                parent[index] = parent[parent[index]]
                index = parent[index]
            return index

        for pair in key_pairs:
            left, right = tuple(pair)
            left_index = cluster_of.get(left)
            right_index = cluster_of.get(right)
            if left_index is None or right_index is None:
                continue
            left_root, right_root = find(left_index), find(right_index)
            if left_root != right_root:
                parent[max(left_root, right_root)] = min(left_root, right_root)

        grouped: dict[int, list[URIRef]] = {}
        for index, cluster in enumerate(candidate_clusters):
            grouped.setdefault(find(index), []).extend(cluster)
        return list(grouped.values())

    def _cluster_entities_by_role(
        self, representations: dict[URIRef, EntityRepresentation]
    ) -> tuple[list[list[URIRef]], dict[URIRef, np.ndarray]]:
        grouped_entities: dict[str, dict[URIRef, EntityRepresentation]] = {}
        for entity, representation in representations.items():
            grouped_entities.setdefault(self._role_key(representation), {})[entity] = (
                representation
            )

        all_clusters: list[list[URIRef]] = []
        all_embeddings: dict[URIRef, np.ndarray] = {}
        for role_representations in grouped_entities.values():
            role_clusters, role_embeddings = self.clusterer.cluster_entities(
                role_representations
            )
            all_clusters.extend(role_clusters)
            all_embeddings.update(role_embeddings)
        return all_clusters, all_embeddings

    @staticmethod
    def _candidate_similarity(
        left: URIRef,
        right: URIRef,
        embeddings: dict[URIRef, np.ndarray],
    ) -> float | None:
        left_embedding = embeddings.get(left)
        right_embedding = embeddings.get(right)
        if left_embedding is None or right_embedding is None:
            return None

        denominator = float(
            np.linalg.norm(left_embedding) * np.linalg.norm(right_embedding)
        )
        if denominator == 0:
            return None
        return float(np.dot(left_embedding, right_embedding) / denominator)

    def _merge_validation_failures(
        self,
        left: URIRef,
        right: URIRef,
        representations: dict[URIRef, EntityRepresentation],
        guard_context: MergeGuardContext | None = None,
    ) -> list[str]:
        failures: list[str] = []
        if guard_context is not None:
            name_pair = frozenset((unscoped_iri(left), unscoped_iri(right)))
            if name_pair in guard_context.sibling_pairs:
                failures.append("sibling")
            left_rep = representations.get(left)
            right_rep = representations.get(right)
            if left_rep is not None and right_rep is not None:
                if self.literal_conflict_guard and self._have_conflicting_literals(
                    left_rep, right_rep
                ):
                    failures.append("literal_conflict")
                if self._have_conflicting_functional_objects(
                    left_rep, right_rep, guard_context.functional_predicates
                ):
                    failures.append("functional_iri_conflict")
                if self._labels_mark_distinct_entities(left_rep, right_rep):
                    failures.append("initials_conflict")
        if not self._are_roles_compatible(left, right, representations):
            failures.append("role")
        if not self._are_types_compatible(left, right, representations):
            failures.append("type")
        if not self._are_lexical_aliases(left, right, representations):
            failures.append("lexical")
        return failures

    def _build_identity_clusters(
        self,
        candidate_clusters: list[list[URIRef]],
        representations: dict[URIRef, EntityRepresentation],
        embeddings: dict[URIRef, np.ndarray],
        direct_relation_pairs: set[frozenset[URIRef]] | None = None,
        guard_context: MergeGuardContext | None = None,
        key_pairs: set[frozenset[URIRef]] | None = None,
    ) -> tuple[
        list[list[URIRef]], list[tuple[URIRef, URIRef, float | None, tuple[str, ...]]]
    ]:
        validated_clusters: list[list[URIRef]] = []
        rejected_merges: list[tuple[URIRef, URIRef, float | None, tuple[str, ...]]] = []

        for candidate_cluster in candidate_clusters:
            if len(candidate_cluster) <= 1:
                validated_clusters.append(candidate_cluster)
                continue

            ordered_cluster = sorted(candidate_cluster, key=str)
            parents: dict[URIRef, URIRef] = {
                entity: entity for entity in ordered_cluster
            }
            members: dict[URIRef, set[URIRef]] = {
                entity: {entity} for entity in ordered_cluster
            }

            # Distinctness vetoes hold cluster-wide: an accepted A–B edge and
            # an accepted B–C edge must not merge a vetoed A–C pair through
            # transitive closure. Computed for every pair up front (the guards
            # are cheap symbolic checks) so unions can be checked against all
            # current members of both sides.
            vetoed_pairs: set[frozenset[URIRef]] = {
                frozenset((left, right))
                for left, right in combinations(ordered_cluster, 2)
                if self._pair_distinctness_veto(
                    left,
                    right,
                    representations,
                    direct_relation_pairs=direct_relation_pairs,
                    guard_context=guard_context,
                )
            }

            def find(entity: URIRef) -> URIRef:
                root = parents[entity]
                if root != entity:
                    parents[entity] = find(root)
                return parents[entity]

            def union_blocked(left_root: URIRef, right_root: URIRef) -> bool:
                left_members = members[left_root]
                right_members = members[right_root]
                return any(
                    frozenset((left_member, right_member)) in vetoed_pairs
                    for left_member in left_members
                    for right_member in right_members
                )

            def union(left: URIRef, right: URIRef) -> None:
                left_root = find(left)
                right_root = find(right)
                if left_root == right_root:
                    return
                if str(left_root) <= str(right_root):
                    parents[right_root] = left_root
                    members[left_root] |= members.pop(right_root)
                else:
                    parents[left_root] = right_root
                    members[right_root] |= members.pop(left_root)

            for left, right in combinations(ordered_cluster, 2):
                pair = frozenset((left, right))
                score = self._candidate_similarity(left, right, embeddings)
                if score is not None and score < self.candidate_similarity_threshold:
                    # Label-confirmed and key-confirmed pairs bypass the cosine
                    # gate (mirrors EntityAligner): short-string embeddings of
                    # aliases like "Baranov, D." vs "Dmitry Baranov" hover
                    # around the threshold, which made identity linking
                    # nondeterministic — and a shared identifier value needs no
                    # embedding agreement at all.
                    if not (
                        (key_pairs is not None and pair in key_pairs)
                        or self._labels_confirm_identity(left, right, representations)
                    ):
                        continue
                if self._can_merge_as_identity(
                    left,
                    right,
                    representations,
                    direct_relation_pairs=direct_relation_pairs,
                    guard_context=guard_context,
                    key_pairs=key_pairs,
                ):
                    left_root = find(left)
                    right_root = find(right)
                    if left_root == right_root:
                        continue
                    if union_blocked(left_root, right_root):
                        # The pair itself is mergeable, but somewhere in the
                        # two groups sits a vetoed pair — accepting the edge
                        # would chain around that guard.
                        rejected_merges.append((left, right, score, ("cluster_veto",)))
                        continue
                    union(left, right)
                    continue
                rejected_merges.append(
                    (
                        left,
                        right,
                        score,
                        tuple(
                            self._merge_validation_failures(
                                left,
                                right,
                                representations,
                                guard_context=guard_context,
                            )
                        ),
                    )
                )

            grouped: dict[URIRef, list[URIRef]] = {}
            for entity in ordered_cluster:
                grouped.setdefault(find(entity), []).append(entity)

            for group in grouped.values():
                sorted_group = sorted(group, key=str)
                validated_clusters.append(sorted_group)

        return validated_clusters, rejected_merges

    def _select_ontology_anchor_candidates(
        self,
        tentative_entities: list[URIRef],
        tentative_representations: dict[URIRef, EntityRepresentation],
        tentative_doc_iris: dict[URIRef, URIRef],
        ontology_graph: RDFGraph | None,
        known_ontology_entities: set[URIRef],
    ) -> dict[URIRef, URIRef]:
        """Pick ontology anchors and preserve the triggering document IRI."""
        if (
            ontology_graph is None
            or not tentative_entities
            or not known_ontology_entities
        ):
            return {}

        ontology_entities = [
            entity
            for entity in known_ontology_entities
            if not self._is_standard_ontology_entity(entity)
        ]
        if not ontology_entities:
            return {}

        ontology_graphs = {entity: ontology_graph for entity in ontology_entities}
        ontology_representations = self.normalizer.create_representations_batch(
            ontology_entities, ontology_graphs
        )

        token_index: dict[str, set[URIRef]] = {}
        for entity, representation in ontology_representations.items():
            for token in self._tokenize(representation.representation):
                token_index.setdefault(token, set()).add(entity)

        selected: dict[URIRef, URIRef] = {}
        for tentative_entity in tentative_entities:
            tentative_representation = tentative_representations.get(tentative_entity)
            if tentative_representation is None:
                continue
            tentative_doc_iri = tentative_doc_iris.get(tentative_entity)
            if tentative_doc_iri is None:
                continue
            tentative_tokens = self._tokenize(tentative_representation.representation)
            if not tentative_tokens:
                continue

            candidate_pool: set[URIRef] = set()
            for token in tentative_tokens:
                candidate_pool.update(token_index.get(token, set()))

            if not candidate_pool:
                continue

            scored: list[tuple[int, URIRef]] = []
            for candidate in candidate_pool:
                candidate_representation = ontology_representations.get(candidate)
                if candidate_representation is None:
                    continue
                candidate_tokens = self._tokenize(
                    candidate_representation.representation
                )
                overlap = len(tentative_tokens & candidate_tokens)
                if overlap >= 2:
                    scored.append((overlap, candidate))

            scored.sort(key=lambda item: (-item[0], str(item[1])))
            for _, candidate in scored[:3]:
                selected.setdefault(candidate, tentative_doc_iri)

        return selected

    def _classify_entity_for_unit(
        self,
        entity: URIRef,
        unit: ContentUnit,
        known_ontology_entities: set[URIRef],
    ) -> EntityClassification:
        """Classify an entity as fact, known ontology, or tentative ontology."""
        if unit.type == OutputType.ONTOLOGIES:
            return EntityClassification.KNOWN_ONTOLOGY

        if self._is_fact_entity_in_unit(entity, unit):
            return EntityClassification.FACT

        if entity in known_ontology_entities or self._is_standard_ontology_entity(
            entity
        ):
            return EntityClassification.KNOWN_ONTOLOGY

        return EntityClassification.TENTATIVE_ONTOLOGY

    @staticmethod
    def _classification_priority(classification: EntityClassification) -> int:
        """Return priority for multi-unit classification merging."""
        if classification == EntityClassification.KNOWN_ONTOLOGY:
            return 3
        if classification == EntityClassification.TENTATIVE_ONTOLOGY:
            return 2
        return 1

    def _register_entity(
        self,
        *,
        entity: URIRef,
        unit: ContentUnit,
        state: _EntityCollectionState,
    ) -> None:
        """Register one URI entity with its document and stable classification."""
        state.entities.add(entity)
        state.source_entities.add(entity)
        state.entity_doc_iris.setdefault(entity, unit.doc_iri)
        current = state.entity_classification.get(entity, EntityClassification.FACT)
        candidate = self._classify_entity_for_unit(entity, unit, state.known_entities)
        state.entity_classification[entity] = (
            candidate
            if self._classification_priority(candidate)
            >= self._classification_priority(current)
            else current
        )

    @staticmethod
    def _register_direct_relation(
        state: _EntityCollectionState,
        subject: URIRef,
        obj: URIRef,
    ) -> None:
        """Record direct subject-object URI relation pair in collection state."""
        if subject == obj:
            return
        state.direct_relation_pairs.add(frozenset((subject, obj)))

    def _collect_all_entities(
        self,
        units: list[ContentUnit],
        known_ontology_entities: set[URIRef] | None = None,
    ) -> tuple[
        list[URIRef],
        set[URIRef],
        dict[URIRef, RDFGraph],
        dict[URIRef, URIRef],
        dict[URIRef, EntityClassification],
        set[frozenset[URIRef]],
        dict[tuple[URIRef, URIRef], set[URIRef]],
    ]:
        """Collect all entities from all content unit graphs.

        Each entity is associated with a graph holding its triples and the
        ``doc_iri`` of the first :class:`ContentUnit` that mentions it (in
        practice most pipelines aggregate chunks of the same document, so all
        ``doc_iri`` values are identical).

        Args:
            units: List of content units to aggregate.
            known_ontology_entities: Entities of the selected ontology, used
                for classification.

        Returns:
            Tuple of (
                entities,
                source_entities,
                entity_to_graph,
                entity_to_doc_iri,
                entity_to_classification,
                direct_relation_pairs,
                object_groups,
            ).
        """
        state = _EntityCollectionState(known_entities=known_ontology_entities or set())

        for unit in units:
            if unit.graph is None:
                continue
            unit.graph.sanitize_prefixes_namespaces()
            # Keep collection in the same URI space that rewrite/merge consumes
            # (unit.graph). Using graph_absolute here causes mapping keys to miss
            # during rewrite, because unit.graph still contains the original terms.
            for triple in unit.graph:
                state.context.add(triple)
                s, p, o = triple
                if isinstance(s, URIRef) and isinstance(o, URIRef):
                    # Structural guards key on *names*, as they did before
                    # unit scoping: a subject mentioned in two units points at
                    # two scoped objects, and when those carry the same name
                    # they must not read as siblings, nor make the predicate
                    # look single-valued on twice the subjects.
                    name_s, name_o = unscoped_iri(s), unscoped_iri(o)
                    self._register_direct_relation(
                        state=state, subject=name_s, obj=name_o
                    )
                    if isinstance(p, URIRef) and p != RDF.type:
                        state.object_groups.setdefault((name_s, p), set()).add(name_o)
                for term in (s, p, o):
                    if isinstance(term, URIRef):
                        self._register_entity(entity=term, unit=unit, state=state)

        # One graph serves every entity: a representation reads only the
        # triples that mention its entity, through the graph's indexes.
        return (
            list(state.entities),
            state.source_entities,
            dict.fromkeys(state.entities, state.context),
            state.entity_doc_iris,
            state.entity_classification,
            state.direct_relation_pairs,
            state.object_groups,
        )

    def aggregate_graphs(
        self,
        units: list[ContentUnit],
        ontology_graph: RDFGraph,
        merge_vetoes: set[frozenset[URIRef]] | None = None,
    ) -> AggregationResult:
        """Aggregate multiple content unit graphs with embedding-based disambiguation.

        Args:
            units: List of ContentUnits to aggregate.
            ontology_graph: Selected ontology graph used to distinguish
                known ontology entities from tentative ontology-like aliases.
            merge_vetoes: Extra entity pairs that must never identity-merge —
                the targeted un-merge lever used by the post-aggregation
                validation gate. Unioned into the direct-relation veto set.

        Returns:
            :class:`AggregationResult` with the merged graph and merge
            bookkeeping (decisions, merged clusters, rejection count).
        """
        logger.info(f"Starting aggregation with metadata for {len(units)} units")
        if ontology_graph is None:
            raise ValueError("ontology_graph must not be None for facts aggregation")

        if not units:
            return AggregationResult(graph=RDFGraph())

        # Steps 1-3: Collect, normalise, candidate clustering
        known_ontology_entities = self._build_known_ontology_entities(ontology_graph)
        (
            entities,
            source_entities,
            entity_graphs,
            entity_doc_iris,
            entity_classification,
            direct_relation_pairs,
            object_groups,
        ) = self._collect_all_entities(units, known_ontology_entities)
        if merge_vetoes:
            direct_relation_pairs = direct_relation_pairs | merge_vetoes
        schema_functional_predicates = harvest_max_one_predicates(ontology_graph)
        guard_context = MergeGuardContext(
            sibling_pairs=build_sibling_pairs(
                object_groups, scope=self.sibling_guard_scope
            ),
            functional_predicates=schema_functional_predicates
            | empirically_functional_predicates(
                object_groups,
                min_support=self.functional_min_empirical_support,
            ),
        )
        representations = self.normalizer.create_representations_batch(
            entities, entity_graphs
        )
        decisions: dict[URIRef, EntityDecision] = {
            entity: EntityDecision(
                classification=classification,
                identity_target=entity,
            )
            for entity, classification in entity_classification.items()
        }
        tentative_entities = [
            entity
            for entity, decision in decisions.items()
            if decision.classification == EntityClassification.TENTATIVE_ONTOLOGY
        ]
        anchor_candidates = self._select_ontology_anchor_candidates(
            tentative_entities=tentative_entities,
            tentative_representations=representations,
            tentative_doc_iris=entity_doc_iris,
            ontology_graph=ontology_graph,
            known_ontology_entities=known_ontology_entities,
        )
        if anchor_candidates:
            for ontology_entity, anchor_doc_iri in anchor_candidates.items():
                if ontology_entity in entity_graphs:
                    continue
                entities.append(ontology_entity)
                entity_graphs[ontology_entity] = ontology_graph
                entity_doc_iris[ontology_entity] = anchor_doc_iri
                entity_classification[ontology_entity] = (
                    EntityClassification.KNOWN_ONTOLOGY
                )
                decisions[ontology_entity] = EntityDecision(
                    classification=EntityClassification.KNOWN_ONTOLOGY,
                    identity_target=ontology_entity,
                )
                representations[ontology_entity] = (
                    self.normalizer.create_representation(
                        ontology_entity, ontology_graph
                    )
                )
        entity_is_known_ontology = {
            entity: decision.classification == EntityClassification.KNOWN_ONTOLOGY
            for entity, decision in decisions.items()
        }
        if logger.isEnabledFor(logging.INFO):
            known_count = sum(
                1 for is_known in entity_is_known_ontology.values() if is_known
            )
            fact_count = sum(
                1
                for decision in decisions.values()
                if decision.classification == EntityClassification.FACT
            )
            logger.info(
                "Aggregation entity classification stats: fact=%d known_ontology=%d "
                "tentative_ontology=%d",
                fact_count,
                known_count,
                len(tentative_entities),
            )

        candidate_clusters, embeddings = self._cluster_entities_by_role(representations)
        key_pairs: set[frozenset[URIRef]] = set()
        if self.natural_key_merge:
            key_pairs = self._collect_natural_key_pairs(
                representations, schema_functional_predicates
            )
            if key_pairs:
                logger.info(
                    "Natural-key evidence proposed %d candidate pair(s)",
                    len(key_pairs),
                )
                candidate_clusters = self._merge_candidate_clusters_by_key_pairs(
                    candidate_clusters, key_pairs
                )
        clusters, rejected_merges = self._build_identity_clusters(
            candidate_clusters=candidate_clusters,
            representations=representations,
            embeddings=embeddings,
            direct_relation_pairs=direct_relation_pairs,
            guard_context=guard_context,
            key_pairs=key_pairs or None,
        )
        if rejected_merges:
            logger.info(
                "Rejected %d candidate merges after symbolic validation",
                len(rejected_merges),
            )
            for left, right, score, failed_checks in rejected_merges:
                logger.debug(
                    "Rejected candidate merge: %s <-> %s (score=%s, failed=%s)",
                    left,
                    right,
                    f"{score:.3f}" if score is not None else "n/a",
                    ",".join(failed_checks) if failed_checks else "unknown",
                )

        # Step 4: Canonical identity mapping (no URI policy yet)
        identity_mapping = self.selector.create_mapping(
            clusters,
            representations,
            entity_is_known_ontology=entity_is_known_ontology,
        )

        # Keep known ontology entities stable. Tentative ontology-like entities are:
        # - mapped to known ontology representatives when present in a mixed cluster
        # - preserved as-is when only tentative entities are present
        suppress_sameas_origins: set[URIRef] = set()
        suppress_fact_subject_sources: set[URIRef] = set()
        for cluster in clusters:
            known_ontology_entities_in_cluster = [
                entity
                for entity in cluster
                if decisions.get(entity) is not None
                and decisions[entity].classification
                == EntityClassification.KNOWN_ONTOLOGY
            ]
            tentative_entities_in_cluster = [
                entity
                for entity in cluster
                if decisions.get(entity) is not None
                and decisions[entity].classification
                == EntityClassification.TENTATIVE_ONTOLOGY
            ]
            fact_entities_in_cluster = [
                entity
                for entity in cluster
                if decisions.get(entity) is not None
                and decisions[entity].classification == EntityClassification.FACT
            ]

            for entity in known_ontology_entities_in_cluster:
                identity_mapping[entity] = entity

            if known_ontology_entities_in_cluster:
                canonical_known_ontology = self.selector.select_representative(
                    known_ontology_entities_in_cluster,
                    representations,
                    entity_is_known_ontology=entity_is_known_ontology,
                )
                for tentative_entity in tentative_entities_in_cluster:
                    if self._can_merge_as_identity(
                        tentative_entity,
                        canonical_known_ontology,
                        representations,
                        direct_relation_pairs=direct_relation_pairs,
                        guard_context=guard_context,
                    ):
                        identity_mapping[tentative_entity] = canonical_known_ontology
                        decisions[tentative_entity].suppress_sameas = True
                    else:
                        identity_mapping[tentative_entity] = tentative_entity
                for fact_entity in fact_entities_in_cluster:
                    if self._can_merge_as_identity(
                        fact_entity,
                        canonical_known_ontology,
                        representations,
                        direct_relation_pairs=direct_relation_pairs,
                        guard_context=guard_context,
                    ):
                        identity_mapping[fact_entity] = canonical_known_ontology
                        decisions[fact_entity].suppress_sameas = True
                        decisions[fact_entity].suppress_fact_subject_assertions = True
                    else:
                        identity_mapping[fact_entity] = fact_entity

            elif tentative_entities_in_cluster:
                # In mixed FACT + TENTATIVE clusters with no known ontology
                # entity, prefer the FACT side when symbolic identity checks
                # agree (e.g. hallucinated ontology prefix on an instance).
                if fact_entities_in_cluster:
                    canonical_fact = self.selector.select_representative(
                        fact_entities_in_cluster,
                        representations,
                        entity_is_known_ontology=entity_is_known_ontology,
                    )
                    for fact_entity in fact_entities_in_cluster:
                        identity_mapping[fact_entity] = canonical_fact
                    for tentative_entity in tentative_entities_in_cluster:
                        if self._can_merge_as_identity(
                            tentative_entity,
                            canonical_fact,
                            representations,
                            direct_relation_pairs=direct_relation_pairs,
                            guard_context=guard_context,
                        ):
                            identity_mapping[tentative_entity] = canonical_fact
                            decisions[tentative_entity].suppress_sameas = True
                        else:
                            identity_mapping[tentative_entity] = tentative_entity
                else:
                    for tentative_entity in tentative_entities_in_cluster:
                        identity_mapping[tentative_entity] = tentative_entity

        for entity, target in identity_mapping.items():
            if entity in decisions:
                decisions[entity].identity_target = target

        suppress_sameas_origins = {
            entity for entity, decision in decisions.items() if decision.suppress_sameas
        }
        suppress_fact_subject_sources = {
            entity
            for entity, decision in decisions.items()
            if decision.suppress_fact_subject_assertions
        }

        # Step 5: URI assignment from canonical identity + namespace policy
        final_mapping = self.uri_builder.create_entity_uri_mapping(
            identity_mapping=identity_mapping,
            representations=representations,
            entity_doc_iris=entity_doc_iris,
            entity_is_ontology={
                entity: (
                    decisions.get(entity) is not None
                    and decisions[entity].classification != EntityClassification.FACT
                )
                for entity in representations
            },
        )
        for entity, final_uri in final_mapping.items():
            if entity in decisions:
                decisions[entity].final_uri = final_uri
        known_ontology_entities_all = {
            entity
            for entity, decision in decisions.items()
            if decision.classification == EntityClassification.KNOWN_ONTOLOGY
        }
        assert all(
            identity_mapping.get(entity, entity) == entity
            for entity in known_ontology_entities_all
        ), "Known ontology entities must remain identity-mapped"
        assert not (known_ontology_entities_all & suppress_sameas_origins), (
            "Known ontology entities cannot be suppress_sameas origins"
        )
        assert not (known_ontology_entities_all & suppress_fact_subject_sources), (
            "Known ontology entities cannot be suppress_fact_subject origins"
        )
        assert all(entity in decisions for entity in source_entities), (
            "Every source entity must have a decision record"
        )
        final_mapping = {
            entity: mapped
            for entity, mapped in final_mapping.items()
            if entity in source_entities
        }

        # Step 7: Rewrite and merge with provenance
        active_units = [u for u in units if u.graph is not None and len(u.graph) > 0]
        merged_graph = self.rewriter.merge_graphs_with_provenance(
            active_units,
            final_mapping,
            suppress_sameas_origins=suppress_sameas_origins,
            suppress_fact_subject_sources=suppress_fact_subject_sources,
        )

        merged_clusters = build_merged_clusters(final_mapping, identity_mapping)
        key_supported_clusters = sorted(
            {
                str(final_mapping[left])
                for pair in key_pairs
                for left, right in [tuple(pair)]
                if left in final_mapping
                and final_mapping.get(right) == final_mapping[left]
            }
        )

        logger.info("Aggregation with metadata complete")
        return AggregationResult(
            graph=merged_graph,
            decisions=decisions,
            merged_clusters=merged_clusters,
            rejected_merge_count=len(rejected_merges),
            key_supported_clusters=key_supported_clusters,
            cross_unit_object_pairs=build_cross_unit_object_pairs(
                active_units, final_mapping
            ),
        )

    def postprocess_facts_units(
        self,
        units: list[ContentUnit],
        ontology_graph: RDFGraph,
        *,
        doc_iri: URIRef | None = None,
        document_metadata: dict[str, Any] | None = None,
        doc_namespace: str | None = None,
        merge_vetoes: set[frozenset[URIRef]] | None = None,
    ) -> AggregationResult:
        """Sanitize facts units, then run aggregation/normalization.

        This method is intentionally safe for both single-unit and multi-unit
        inputs so unit-pipeline and graph-pipeline paths share the same
        post-processing behavior.

        When ``doc_iri`` and non-empty ``document_metadata`` are provided,
        caller-asserted document identity triples are attached to the merged
        facts graph. Business-oriented keys mint typed entities under
        ``doc_namespace`` (defaults to the document facts namespace).

        With ``unit_scoped_fact_iris`` on, every minted fact IRI is suffixed
        with its unit index *in place* on the unit graph before aggregation
        (see :mod:`ontocast.tool.agg.unit_scope`), so a local name minted by
        two units is a merge decision rather than an accidental fusion. The
        rewrite is idempotent, which keeps the validation gate's
        re-aggregation of the same units on identical input.

        Args:
            units: Facts content units to aggregate.
            ontology_graph: Merged ontology context for classification/guards.
            doc_iri: Document IRI for metadata provenance attachment.
            document_metadata: Caller-asserted document identity metadata.
            doc_namespace: Namespace for metadata-minted entities.
            merge_vetoes: Entity pairs that must never identity-merge
                (validation-gate un-merge lever).

        Returns:
            :class:`AggregationResult`; its ``graph`` carries the merged facts
            plus any document-metadata provenance.
        """
        for unit in units:
            unit.sanitize()
            if self.unit_scoped_fact_iris and unit.type != OutputType.ONTOLOGIES:
                scope_fact_iris(
                    unit.graph, unit.index, (self.base_iri, str(unit.doc_iri))
                )
        result = self.aggregate_graphs(
            units=units, ontology_graph=ontology_graph, merge_vetoes=merge_vetoes
        )
        if doc_iri is not None and document_metadata:
            apply_document_metadata_provenance(
                doc_iri,
                document_metadata,
                result.graph,
                entity_namespace=doc_namespace,
            )
        # Cross-unit prefix conflicts surface only on the merged graph (e.g.
        # aliases of one namespace arriving from different units), so sanitize
        # once more after aggregation.
        result.graph.sanitize_prefixes_namespaces()
        return result

Attributes

base_iri = base_iri instance-attribute

candidate_similarity_threshold = cfg.candidate_similarity_threshold if candidate_similarity_threshold is None else candidate_similarity_threshold instance-attribute

clusterer = EntityClusterer(embedding_model=cfg.embedding_model, similarity_threshold=self.candidate_similarity_threshold) instance-attribute

functional_min_empirical_support = cfg.functional_min_empirical_support instance-attribute

initials_distinct_guard = cfg.initials_distinct_guard instance-attribute

lexical_label_jaccard = cfg.lexical_label_jaccard instance-attribute

lexical_sequence_ratio = cfg.lexical_sequence_ratio instance-attribute

lexical_token_jaccard = cfg.lexical_token_jaccard instance-attribute

literal_conflict_guard = cfg.literal_conflict_guard instance-attribute

natural_key_merge = cfg.natural_key_merge instance-attribute

normalizer = EntityNormalizer(facts_iri=self.base_iri) instance-attribute

rewriter = GraphRewriter(add_sameas_links=add_sameas_links, blocked_sameas_namespaces=(self.base_iri,)) instance-attribute

selector = ClusterRepresentativeSelector() instance-attribute

sibling_guard_scope = str(cfg.sibling_guard_scope) instance-attribute

type_guard_untyped = str(cfg.type_guard_untyped) instance-attribute

unit_scoped_fact_iris = cfg.unit_scoped_fact_iris instance-attribute

uri_builder = URIBuilder(base_iri=self.base_iri) instance-attribute

Methods:

__init__(config=None, *, add_sameas_links=True, base_iri=DEFAULT_IRI, candidate_similarity_threshold=None)

Initialise the embedding-based aggregator.

Every tunable lives on :class:AggregationConfig, so settings.py stays the single source of their defaults rather than restating them in this signature and again at the call site.

Parameters:

Name	Type	Description	Default
config	AggregationConfig | None	Aggregation tunables. Defaults to :class:AggregationConfig, i.e. the environment-resolved settings.	None
add_sameas_links	bool	Whether to add owl:sameAs for merged entities. Not config-driven: callers choose it per use, and the entity aligner wants different behaviour from the pipeline.	True
base_iri	str	Base IRI for fact entity URIs. Entities under this namespace are facts; everything else is treated as an ontology entity and left unchanged.	DEFAULT_IRI
candidate_similarity_threshold	float | None	Overrides the configured permissive candidate threshold. The entity aligner pins it to its own similarity threshold rather than the pipeline's.	None

Source code in ontocast/tool/agg/aggregate.py

def __init__(
    self,
    config: AggregationConfig | None = None,
    *,
    add_sameas_links: bool = True,
    base_iri: str = DEFAULT_IRI,
    candidate_similarity_threshold: float | None = None,
):
    """Initialise the embedding-based aggregator.

    Every tunable lives on :class:`AggregationConfig`, so ``settings.py``
    stays the single source of their defaults rather than restating them in
    this signature and again at the call site.

    Args:
        config: Aggregation tunables. Defaults to :class:`AggregationConfig`,
            i.e. the environment-resolved settings.
        add_sameas_links: Whether to add ``owl:sameAs`` for merged entities.
            Not config-driven: callers choose it per use, and the entity
            aligner wants different behaviour from the pipeline.
        base_iri: Base IRI for fact entity URIs. Entities under this
            namespace are facts; everything else is treated as an ontology
            entity and left unchanged.
        candidate_similarity_threshold: Overrides the configured permissive
            candidate threshold. The entity aligner pins it to its own
            similarity threshold rather than the pipeline's.
    """
    cfg = config or AggregationConfig()

    self.base_iri = base_iri
    self.candidate_similarity_threshold = (
        cfg.candidate_similarity_threshold
        if candidate_similarity_threshold is None
        else candidate_similarity_threshold
    )
    self.lexical_label_jaccard = cfg.lexical_label_jaccard
    self.lexical_sequence_ratio = cfg.lexical_sequence_ratio
    self.lexical_token_jaccard = cfg.lexical_token_jaccard
    self.functional_min_empirical_support = cfg.functional_min_empirical_support
    self.sibling_guard_scope = str(cfg.sibling_guard_scope)
    self.literal_conflict_guard = cfg.literal_conflict_guard
    self.initials_distinct_guard = cfg.initials_distinct_guard
    self.natural_key_merge = cfg.natural_key_merge
    self.type_guard_untyped = str(cfg.type_guard_untyped)
    self.unit_scoped_fact_iris = cfg.unit_scoped_fact_iris
    if candidate_similarity_threshold is None:
        self._warn_inert_similarity_threshold(cfg)

    # Pipeline components (EntityClusterer imports sklearn/ST lazily).
    # The clusterer runs at the permissive candidate threshold: candidates
    # are validated symbolically afterwards, so there is exactly one
    # clustering threshold on this path.
    from .clustering import EntityClusterer

    self.normalizer = EntityNormalizer(facts_iri=self.base_iri)
    self.clusterer = EntityClusterer(
        embedding_model=cfg.embedding_model,
        similarity_threshold=self.candidate_similarity_threshold,
    )
    self.selector = ClusterRepresentativeSelector()
    self.uri_builder = URIBuilder(base_iri=self.base_iri)
    self.rewriter = GraphRewriter(
        add_sameas_links=add_sameas_links,
        blocked_sameas_namespaces=(self.base_iri,),
    )

aggregate_graphs(units, ontology_graph, merge_vetoes=None)

Aggregate multiple content unit graphs with embedding-based disambiguation.

Parameters:

Name	Type	Description	Default
units	list[ContentUnit]	List of ContentUnits to aggregate.	required
ontology_graph	RDFGraph	Selected ontology graph used to distinguish known ontology entities from tentative ontology-like aliases.	required
merge_vetoes	set[frozenset[URIRef]] | None	Extra entity pairs that must never identity-merge — the targeted un-merge lever used by the post-aggregation validation gate. Unioned into the direct-relation veto set.	None

Returns:

Type	Description
AggregationResult
AggregationResult	bookkeeping (decisions, merged clusters, rejection count).

Source code in ontocast/tool/agg/aggregate.py

def aggregate_graphs(
    self,
    units: list[ContentUnit],
    ontology_graph: RDFGraph,
    merge_vetoes: set[frozenset[URIRef]] | None = None,
) -> AggregationResult:
    """Aggregate multiple content unit graphs with embedding-based disambiguation.

    Args:
        units: List of ContentUnits to aggregate.
        ontology_graph: Selected ontology graph used to distinguish
            known ontology entities from tentative ontology-like aliases.
        merge_vetoes: Extra entity pairs that must never identity-merge —
            the targeted un-merge lever used by the post-aggregation
            validation gate. Unioned into the direct-relation veto set.

    Returns:
        :class:`AggregationResult` with the merged graph and merge
        bookkeeping (decisions, merged clusters, rejection count).
    """
    logger.info(f"Starting aggregation with metadata for {len(units)} units")
    if ontology_graph is None:
        raise ValueError("ontology_graph must not be None for facts aggregation")

    if not units:
        return AggregationResult(graph=RDFGraph())

    # Steps 1-3: Collect, normalise, candidate clustering
    known_ontology_entities = self._build_known_ontology_entities(ontology_graph)
    (
        entities,
        source_entities,
        entity_graphs,
        entity_doc_iris,
        entity_classification,
        direct_relation_pairs,
        object_groups,
    ) = self._collect_all_entities(units, known_ontology_entities)
    if merge_vetoes:
        direct_relation_pairs = direct_relation_pairs | merge_vetoes
    schema_functional_predicates = harvest_max_one_predicates(ontology_graph)
    guard_context = MergeGuardContext(
        sibling_pairs=build_sibling_pairs(
            object_groups, scope=self.sibling_guard_scope
        ),
        functional_predicates=schema_functional_predicates
        | empirically_functional_predicates(
            object_groups,
            min_support=self.functional_min_empirical_support,
        ),
    )
    representations = self.normalizer.create_representations_batch(
        entities, entity_graphs
    )
    decisions: dict[URIRef, EntityDecision] = {
        entity: EntityDecision(
            classification=classification,
            identity_target=entity,
        )
        for entity, classification in entity_classification.items()
    }
    tentative_entities = [
        entity
        for entity, decision in decisions.items()
        if decision.classification == EntityClassification.TENTATIVE_ONTOLOGY
    ]
    anchor_candidates = self._select_ontology_anchor_candidates(
        tentative_entities=tentative_entities,
        tentative_representations=representations,
        tentative_doc_iris=entity_doc_iris,
        ontology_graph=ontology_graph,
        known_ontology_entities=known_ontology_entities,
    )
    if anchor_candidates:
        for ontology_entity, anchor_doc_iri in anchor_candidates.items():
            if ontology_entity in entity_graphs:
                continue
            entities.append(ontology_entity)
            entity_graphs[ontology_entity] = ontology_graph
            entity_doc_iris[ontology_entity] = anchor_doc_iri
            entity_classification[ontology_entity] = (
                EntityClassification.KNOWN_ONTOLOGY
            )
            decisions[ontology_entity] = EntityDecision(
                classification=EntityClassification.KNOWN_ONTOLOGY,
                identity_target=ontology_entity,
            )
            representations[ontology_entity] = (
                self.normalizer.create_representation(
                    ontology_entity, ontology_graph
                )
            )
    entity_is_known_ontology = {
        entity: decision.classification == EntityClassification.KNOWN_ONTOLOGY
        for entity, decision in decisions.items()
    }
    if logger.isEnabledFor(logging.INFO):
        known_count = sum(
            1 for is_known in entity_is_known_ontology.values() if is_known
        )
        fact_count = sum(
            1
            for decision in decisions.values()
            if decision.classification == EntityClassification.FACT
        )
        logger.info(
            "Aggregation entity classification stats: fact=%d known_ontology=%d "
            "tentative_ontology=%d",
            fact_count,
            known_count,
            len(tentative_entities),
        )

    candidate_clusters, embeddings = self._cluster_entities_by_role(representations)
    key_pairs: set[frozenset[URIRef]] = set()
    if self.natural_key_merge:
        key_pairs = self._collect_natural_key_pairs(
            representations, schema_functional_predicates
        )
        if key_pairs:
            logger.info(
                "Natural-key evidence proposed %d candidate pair(s)",
                len(key_pairs),
            )
            candidate_clusters = self._merge_candidate_clusters_by_key_pairs(
                candidate_clusters, key_pairs
            )
    clusters, rejected_merges = self._build_identity_clusters(
        candidate_clusters=candidate_clusters,
        representations=representations,
        embeddings=embeddings,
        direct_relation_pairs=direct_relation_pairs,
        guard_context=guard_context,
        key_pairs=key_pairs or None,
    )
    if rejected_merges:
        logger.info(
            "Rejected %d candidate merges after symbolic validation",
            len(rejected_merges),
        )
        for left, right, score, failed_checks in rejected_merges:
            logger.debug(
                "Rejected candidate merge: %s <-> %s (score=%s, failed=%s)",
                left,
                right,
                f"{score:.3f}" if score is not None else "n/a",
                ",".join(failed_checks) if failed_checks else "unknown",
            )

    # Step 4: Canonical identity mapping (no URI policy yet)
    identity_mapping = self.selector.create_mapping(
        clusters,
        representations,
        entity_is_known_ontology=entity_is_known_ontology,
    )

    # Keep known ontology entities stable. Tentative ontology-like entities are:
    # - mapped to known ontology representatives when present in a mixed cluster
    # - preserved as-is when only tentative entities are present
    suppress_sameas_origins: set[URIRef] = set()
    suppress_fact_subject_sources: set[URIRef] = set()
    for cluster in clusters:
        known_ontology_entities_in_cluster = [
            entity
            for entity in cluster
            if decisions.get(entity) is not None
            and decisions[entity].classification
            == EntityClassification.KNOWN_ONTOLOGY
        ]
        tentative_entities_in_cluster = [
            entity
            for entity in cluster
            if decisions.get(entity) is not None
            and decisions[entity].classification
            == EntityClassification.TENTATIVE_ONTOLOGY
        ]
        fact_entities_in_cluster = [
            entity
            for entity in cluster
            if decisions.get(entity) is not None
            and decisions[entity].classification == EntityClassification.FACT
        ]

        for entity in known_ontology_entities_in_cluster:
            identity_mapping[entity] = entity

        if known_ontology_entities_in_cluster:
            canonical_known_ontology = self.selector.select_representative(
                known_ontology_entities_in_cluster,
                representations,
                entity_is_known_ontology=entity_is_known_ontology,
            )
            for tentative_entity in tentative_entities_in_cluster:
                if self._can_merge_as_identity(
                    tentative_entity,
                    canonical_known_ontology,
                    representations,
                    direct_relation_pairs=direct_relation_pairs,
                    guard_context=guard_context,
                ):
                    identity_mapping[tentative_entity] = canonical_known_ontology
                    decisions[tentative_entity].suppress_sameas = True
                else:
                    identity_mapping[tentative_entity] = tentative_entity
            for fact_entity in fact_entities_in_cluster:
                if self._can_merge_as_identity(
                    fact_entity,
                    canonical_known_ontology,
                    representations,
                    direct_relation_pairs=direct_relation_pairs,
                    guard_context=guard_context,
                ):
                    identity_mapping[fact_entity] = canonical_known_ontology
                    decisions[fact_entity].suppress_sameas = True
                    decisions[fact_entity].suppress_fact_subject_assertions = True
                else:
                    identity_mapping[fact_entity] = fact_entity

        elif tentative_entities_in_cluster:
            # In mixed FACT + TENTATIVE clusters with no known ontology
            # entity, prefer the FACT side when symbolic identity checks
            # agree (e.g. hallucinated ontology prefix on an instance).
            if fact_entities_in_cluster:
                canonical_fact = self.selector.select_representative(
                    fact_entities_in_cluster,
                    representations,
                    entity_is_known_ontology=entity_is_known_ontology,
                )
                for fact_entity in fact_entities_in_cluster:
                    identity_mapping[fact_entity] = canonical_fact
                for tentative_entity in tentative_entities_in_cluster:
                    if self._can_merge_as_identity(
                        tentative_entity,
                        canonical_fact,
                        representations,
                        direct_relation_pairs=direct_relation_pairs,
                        guard_context=guard_context,
                    ):
                        identity_mapping[tentative_entity] = canonical_fact
                        decisions[tentative_entity].suppress_sameas = True
                    else:
                        identity_mapping[tentative_entity] = tentative_entity
            else:
                for tentative_entity in tentative_entities_in_cluster:
                    identity_mapping[tentative_entity] = tentative_entity

    for entity, target in identity_mapping.items():
        if entity in decisions:
            decisions[entity].identity_target = target

    suppress_sameas_origins = {
        entity for entity, decision in decisions.items() if decision.suppress_sameas
    }
    suppress_fact_subject_sources = {
        entity
        for entity, decision in decisions.items()
        if decision.suppress_fact_subject_assertions
    }

    # Step 5: URI assignment from canonical identity + namespace policy
    final_mapping = self.uri_builder.create_entity_uri_mapping(
        identity_mapping=identity_mapping,
        representations=representations,
        entity_doc_iris=entity_doc_iris,
        entity_is_ontology={
            entity: (
                decisions.get(entity) is not None
                and decisions[entity].classification != EntityClassification.FACT
            )
            for entity in representations
        },
    )
    for entity, final_uri in final_mapping.items():
        if entity in decisions:
            decisions[entity].final_uri = final_uri
    known_ontology_entities_all = {
        entity
        for entity, decision in decisions.items()
        if decision.classification == EntityClassification.KNOWN_ONTOLOGY
    }
    assert all(
        identity_mapping.get(entity, entity) == entity
        for entity in known_ontology_entities_all
    ), "Known ontology entities must remain identity-mapped"
    assert not (known_ontology_entities_all & suppress_sameas_origins), (
        "Known ontology entities cannot be suppress_sameas origins"
    )
    assert not (known_ontology_entities_all & suppress_fact_subject_sources), (
        "Known ontology entities cannot be suppress_fact_subject origins"
    )
    assert all(entity in decisions for entity in source_entities), (
        "Every source entity must have a decision record"
    )
    final_mapping = {
        entity: mapped
        for entity, mapped in final_mapping.items()
        if entity in source_entities
    }

    # Step 7: Rewrite and merge with provenance
    active_units = [u for u in units if u.graph is not None and len(u.graph) > 0]
    merged_graph = self.rewriter.merge_graphs_with_provenance(
        active_units,
        final_mapping,
        suppress_sameas_origins=suppress_sameas_origins,
        suppress_fact_subject_sources=suppress_fact_subject_sources,
    )

    merged_clusters = build_merged_clusters(final_mapping, identity_mapping)
    key_supported_clusters = sorted(
        {
            str(final_mapping[left])
            for pair in key_pairs
            for left, right in [tuple(pair)]
            if left in final_mapping
            and final_mapping.get(right) == final_mapping[left]
        }
    )

    logger.info("Aggregation with metadata complete")
    return AggregationResult(
        graph=merged_graph,
        decisions=decisions,
        merged_clusters=merged_clusters,
        rejected_merge_count=len(rejected_merges),
        key_supported_clusters=key_supported_clusters,
        cross_unit_object_pairs=build_cross_unit_object_pairs(
            active_units, final_mapping
        ),
    )

postprocess_facts_units(units, ontology_graph, *, doc_iri=None, document_metadata=None, doc_namespace=None, merge_vetoes=None)

Sanitize facts units, then run aggregation/normalization.

This method is intentionally safe for both single-unit and multi-unit inputs so unit-pipeline and graph-pipeline paths share the same post-processing behavior.

When doc_iri and non-empty document_metadata are provided, caller-asserted document identity triples are attached to the merged facts graph. Business-oriented keys mint typed entities under doc_namespace (defaults to the document facts namespace).

With unit_scoped_fact_iris on, every minted fact IRI is suffixed with its unit index in place on the unit graph before aggregation (see :mod:ontocast.tool.agg.unit_scope), so a local name minted by two units is a merge decision rather than an accidental fusion. The rewrite is idempotent, which keeps the validation gate's re-aggregation of the same units on identical input.

Parameters:

Name	Type	Description	Default
units	list[ContentUnit]	Facts content units to aggregate.	required
ontology_graph	RDFGraph	Merged ontology context for classification/guards.	required
doc_iri	URIRef | None	Document IRI for metadata provenance attachment.	None
document_metadata	dict[str, Any] | None	Caller-asserted document identity metadata.	None
doc_namespace	str | None	Namespace for metadata-minted entities.	None
merge_vetoes	set[frozenset[URIRef]] | None	Entity pairs that must never identity-merge (validation-gate un-merge lever).	None

Returns:

Type	Description
AggregationResult
AggregationResult	plus any document-metadata provenance.

Source code in ontocast/tool/agg/aggregate.py

def postprocess_facts_units(
    self,
    units: list[ContentUnit],
    ontology_graph: RDFGraph,
    *,
    doc_iri: URIRef | None = None,
    document_metadata: dict[str, Any] | None = None,
    doc_namespace: str | None = None,
    merge_vetoes: set[frozenset[URIRef]] | None = None,
) -> AggregationResult:
    """Sanitize facts units, then run aggregation/normalization.

    This method is intentionally safe for both single-unit and multi-unit
    inputs so unit-pipeline and graph-pipeline paths share the same
    post-processing behavior.

    When ``doc_iri`` and non-empty ``document_metadata`` are provided,
    caller-asserted document identity triples are attached to the merged
    facts graph. Business-oriented keys mint typed entities under
    ``doc_namespace`` (defaults to the document facts namespace).

    With ``unit_scoped_fact_iris`` on, every minted fact IRI is suffixed
    with its unit index *in place* on the unit graph before aggregation
    (see :mod:`ontocast.tool.agg.unit_scope`), so a local name minted by
    two units is a merge decision rather than an accidental fusion. The
    rewrite is idempotent, which keeps the validation gate's
    re-aggregation of the same units on identical input.

    Args:
        units: Facts content units to aggregate.
        ontology_graph: Merged ontology context for classification/guards.
        doc_iri: Document IRI for metadata provenance attachment.
        document_metadata: Caller-asserted document identity metadata.
        doc_namespace: Namespace for metadata-minted entities.
        merge_vetoes: Entity pairs that must never identity-merge
            (validation-gate un-merge lever).

    Returns:
        :class:`AggregationResult`; its ``graph`` carries the merged facts
        plus any document-metadata provenance.
    """
    for unit in units:
        unit.sanitize()
        if self.unit_scoped_fact_iris and unit.type != OutputType.ONTOLOGIES:
            scope_fact_iris(
                unit.graph, unit.index, (self.base_iri, str(unit.doc_iri))
            )
    result = self.aggregate_graphs(
        units=units, ontology_graph=ontology_graph, merge_vetoes=merge_vetoes
    )
    if doc_iri is not None and document_metadata:
        apply_document_metadata_provenance(
            doc_iri,
            document_metadata,
            result.graph,
            entity_namespace=doc_namespace,
        )
    # Cross-unit prefix conflicts surface only on the merged graph (e.g.
    # aliases of one namespace arriving from different units), so sanitize
    # once more after aggregation.
    result.graph.sanitize_prefixes_namespaces()
    return result

EntityAligner

Align entities globally across a list of tagged RDF graphs.

Source code in ontocast/tool/agg/entity_aligner.py

class EntityAligner:
    """Align entities globally across a list of tagged RDF graphs."""

    def __init__(
        self,
        embedding_model: str = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2",
        similarity_threshold: float = 0.80,
    ) -> None:
        self.similarity_threshold = similarity_threshold
        self.normalizer: EntityNormalizer = EntityNormalizer()
        self.clusterer: EntityClusterer = EntityClusterer(
            embedding_model=embedding_model,
            similarity_threshold=similarity_threshold,
        )
        self._compat = EmbeddingBasedAggregator(
            AggregationConfig(
                embedding_model=embedding_model,
                similarity_threshold=similarity_threshold,
            ),
            candidate_similarity_threshold=similarity_threshold,
        )

    @staticmethod
    def _namespace_set(types: list[URIRef]) -> set[str]:
        namespaces: set[str] = set()
        for entity_type in types:
            namespace, _ = split_namespace_local(str(entity_type))
            if namespace is not None:
                namespaces.add(namespace)
        return namespaces

    def _strict_types_compatible(
        self,
        left: GraphEntityRef,
        right: GraphEntityRef,
        representations: dict[GraphEntityRef, EntityRepresentation],
    ) -> bool:
        left_rep = representations.get(left)
        right_rep = representations.get(right)
        if left_rep is None or right_rep is None:
            return False
        if not left_rep.types or not right_rep.types:
            return True
        left_namespaces = self._namespace_set(left_rep.types)
        right_namespaces = self._namespace_set(right_rep.types)
        if not left_namespaces or not right_namespaces:
            return False
        return bool(left_namespaces & right_namespaces)

    def _normalized_label_tokens(self, rep: EntityRepresentation) -> set[str]:
        return {
            self.normalizer.normalize_string(label)
            for label in rep.labels + rep.alt_labels
            if label.strip()
        }

    def _exact_label_match(
        self,
        left: GraphEntityRef,
        right: GraphEntityRef,
        representations: dict[GraphEntityRef, EntityRepresentation],
    ) -> bool:
        left_rep = representations.get(left)
        right_rep = representations.get(right)
        if left_rep is None or right_rep is None:
            return False
        left_tokens = self._normalized_label_tokens(left_rep)
        right_tokens = self._normalized_label_tokens(right_rep)
        if not left_tokens or not right_tokens:
            return False
        return bool(left_tokens & right_tokens)

    def _class_instance_compatible(
        self,
        left: GraphEntityRef,
        right: GraphEntityRef,
        representations: dict[GraphEntityRef, EntityRepresentation],
    ) -> bool:
        left_rep = representations.get(left)
        right_rep = representations.get(right)
        if left_rep is None or right_rep is None:
            return False

        left_is_type_of_right = left.entity in (right_rep.types or [])
        right_is_type_of_left = right.entity in (left_rep.types or [])
        return not (left_is_type_of_right or right_is_type_of_left)

    def _pair_compatible(
        self,
        left: GraphEntityRef,
        right: GraphEntityRef,
        representations: dict[GraphEntityRef, EntityRepresentation],
        regime: MatchRegime,
    ) -> bool:
        if left.entity == right.entity:
            return True

        if not self._class_instance_compatible(left, right, representations):
            return False

        pair_representations = {
            left.entity: representations[left],
            right.entity: representations[right],
        }
        if not self._compat._are_roles_compatible(
            left.entity, right.entity, pair_representations
        ):
            return False
        if not self._compat._are_lexical_aliases(
            left.entity, right.entity, pair_representations
        ):
            return False

        # Type check: always apply when both entities have types.
        # Regime only controls whether namespace must match (strict)
        # or just any shared type suffices (loose).
        left_rep = representations.get(left)
        right_rep = representations.get(right)
        both_have_types = (
            left_rep is not None
            and right_rep is not None
            and left_rep.types
            and right_rep.types
        )
        if both_have_types:
            if regime == MatchRegime.ONTOLOGY_STRICT:
                if not self._strict_types_compatible(left, right, representations):
                    return False
            else:
                if not self._any_type_overlap(left, right, representations):
                    return False

        return True

    def _any_type_overlap(
        self,
        left: GraphEntityRef,
        right: GraphEntityRef,
        representations: dict[GraphEntityRef, EntityRepresentation],
    ) -> bool:
        left_rep = representations.get(left)
        right_rep = representations.get(right)
        if left_rep is None or right_rep is None:
            return False
        left_types = set(left_rep.types)
        right_types = set(right_rep.types)
        return bool(left_types & right_types)

    def _connected_components(
        self,
        nodes: list[GraphEntityRef],
        edges: list[tuple[GraphEntityRef, GraphEntityRef]],
    ) -> list[list[GraphEntityRef]]:
        adjacency: dict[GraphEntityRef, set[GraphEntityRef]] = {
            node: set() for node in nodes
        }
        for left, right in edges:
            adjacency[left].add(right)
            adjacency[right].add(left)

        visited: set[GraphEntityRef] = set()
        components: list[list[GraphEntityRef]] = []
        for start in nodes:
            if start in visited:
                continue
            stack = [start]
            component: list[GraphEntityRef] = []
            while stack:
                node = stack.pop()
                if node in visited:
                    continue
                visited.add(node)
                component.append(node)
                stack.extend(
                    neighbor for neighbor in adjacency[node] if neighbor not in visited
                )
            component.sort(key=lambda ref: (ref.graph_id, str(ref.entity)))
            components.append(component)
        return components

    def align_graphs(
        self,
        graphs: list[TaggedGraph],
        *,
        regime: MatchRegime = MatchRegime.ONTOLOGY_LOOSE,
    ) -> EntityAlignmentResult:
        from .match_common import extract_entities

        refs: list[GraphEntityRef] = []
        representations: dict[GraphEntityRef, EntityRepresentation] = {}
        for tagged in graphs:
            for entity in extract_entities(tagged.graph):
                ref = GraphEntityRef(graph_id=tagged.id, entity=entity)
                refs.append(ref)
                representations[ref] = self.normalizer.create_representation(
                    entity, tagged.graph
                )

        if not refs:
            return EntityAlignmentResult(
                regime=regime,
                similarity_threshold=self.similarity_threshold,
                entity_count=0,
                cluster_count=0,
                clusters=[],
            )

        ordered_refs = list(representations.keys())
        texts = [representations[ref].representation for ref in ordered_refs]
        vectors = self.clusterer.embedder.encode(
            texts, convert_to_numpy=True, show_progress_bar=len(texts) > 100
        )
        embeddings: dict[GraphEntityRef, np.ndarray] = {
            ref: vector for ref, vector in zip(ordered_refs, vectors)
        }

        edges: list[tuple[GraphEntityRef, GraphEntityRef]] = []
        edge_scores: dict[tuple[GraphEntityRef, GraphEntityRef], float] = {}
        for left, right in combinations(refs, 2):
            if left.graph_id == right.graph_id:
                continue
            if left.entity == right.entity:
                edges.append((left, right))
                edge_scores[(left, right)] = 1.0
                edge_scores[(right, left)] = 1.0
                continue
            if not self._pair_compatible(left, right, representations, regime):
                continue
            left_embedding = embeddings[left]
            right_embedding = embeddings[right]
            score = cosine_similarity(left_embedding, right_embedding)
            label_confirmed = self._exact_label_match(left, right, representations)
            if score < self.similarity_threshold and not label_confirmed:
                continue
            edge_score = score if score >= self.similarity_threshold else 1.0
            edges.append((left, right))
            edge_scores[(left, right)] = edge_score
            edge_scores[(right, left)] = edge_score

        adjacency: dict[GraphEntityRef, set[GraphEntityRef]] = {
            node: set() for node in refs
        }
        for left, right in edges:
            adjacency[left].add(right)
            adjacency[right].add(left)

        components = self._connected_components(refs, edges)
        clusters: list[EntityCluster] = []
        for component in components:
            component_set = set(component)
            members: list[GraphEntityMember] = []
            for ref in component:
                best_score: float | None = None
                for neighbor in adjacency[ref]:
                    if neighbor not in component_set:
                        continue
                    score = edge_scores.get((ref, neighbor))
                    if score is None:
                        continue
                    if best_score is None or score > best_score:
                        best_score = score
                members.append(
                    GraphEntityMember(
                        graph_id=ref.graph_id,
                        entity=ref.entity,
                        similarity=best_score,
                    )
                )
            clusters.append(EntityCluster(members=members))

        logger.info(
            "Aligned %s entities into %s clusters across %s graphs",
            len(refs),
            len(clusters),
            len(graphs),
        )
        return EntityAlignmentResult(
            regime=regime,
            similarity_threshold=self.similarity_threshold,
            entity_count=len(refs),
            cluster_count=len(clusters),
            clusters=clusters,
        )

Attributes

clusterer = EntityClusterer(embedding_model=embedding_model, similarity_threshold=similarity_threshold) instance-attribute

normalizer = EntityNormalizer() instance-attribute

similarity_threshold = similarity_threshold instance-attribute

Methods:

__init__(embedding_model='sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2', similarity_threshold=0.8)

Source code in ontocast/tool/agg/entity_aligner.py

def __init__(
    self,
    embedding_model: str = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2",
    similarity_threshold: float = 0.80,
) -> None:
    self.similarity_threshold = similarity_threshold
    self.normalizer: EntityNormalizer = EntityNormalizer()
    self.clusterer: EntityClusterer = EntityClusterer(
        embedding_model=embedding_model,
        similarity_threshold=similarity_threshold,
    )
    self._compat = EmbeddingBasedAggregator(
        AggregationConfig(
            embedding_model=embedding_model,
            similarity_threshold=similarity_threshold,
        ),
        candidate_similarity_threshold=similarity_threshold,
    )

align_graphs(graphs, *, regime=MatchRegime.ONTOLOGY_LOOSE)

Source code in ontocast/tool/agg/entity_aligner.py

def align_graphs(
    self,
    graphs: list[TaggedGraph],
    *,
    regime: MatchRegime = MatchRegime.ONTOLOGY_LOOSE,
) -> EntityAlignmentResult:
    from .match_common import extract_entities

    refs: list[GraphEntityRef] = []
    representations: dict[GraphEntityRef, EntityRepresentation] = {}
    for tagged in graphs:
        for entity in extract_entities(tagged.graph):
            ref = GraphEntityRef(graph_id=tagged.id, entity=entity)
            refs.append(ref)
            representations[ref] = self.normalizer.create_representation(
                entity, tagged.graph
            )

    if not refs:
        return EntityAlignmentResult(
            regime=regime,
            similarity_threshold=self.similarity_threshold,
            entity_count=0,
            cluster_count=0,
            clusters=[],
        )

    ordered_refs = list(representations.keys())
    texts = [representations[ref].representation for ref in ordered_refs]
    vectors = self.clusterer.embedder.encode(
        texts, convert_to_numpy=True, show_progress_bar=len(texts) > 100
    )
    embeddings: dict[GraphEntityRef, np.ndarray] = {
        ref: vector for ref, vector in zip(ordered_refs, vectors)
    }

    edges: list[tuple[GraphEntityRef, GraphEntityRef]] = []
    edge_scores: dict[tuple[GraphEntityRef, GraphEntityRef], float] = {}
    for left, right in combinations(refs, 2):
        if left.graph_id == right.graph_id:
            continue
        if left.entity == right.entity:
            edges.append((left, right))
            edge_scores[(left, right)] = 1.0
            edge_scores[(right, left)] = 1.0
            continue
        if not self._pair_compatible(left, right, representations, regime):
            continue
        left_embedding = embeddings[left]
        right_embedding = embeddings[right]
        score = cosine_similarity(left_embedding, right_embedding)
        label_confirmed = self._exact_label_match(left, right, representations)
        if score < self.similarity_threshold and not label_confirmed:
            continue
        edge_score = score if score >= self.similarity_threshold else 1.0
        edges.append((left, right))
        edge_scores[(left, right)] = edge_score
        edge_scores[(right, left)] = edge_score

    adjacency: dict[GraphEntityRef, set[GraphEntityRef]] = {
        node: set() for node in refs
    }
    for left, right in edges:
        adjacency[left].add(right)
        adjacency[right].add(left)

    components = self._connected_components(refs, edges)
    clusters: list[EntityCluster] = []
    for component in components:
        component_set = set(component)
        members: list[GraphEntityMember] = []
        for ref in component:
            best_score: float | None = None
            for neighbor in adjacency[ref]:
                if neighbor not in component_set:
                    continue
                score = edge_scores.get((ref, neighbor))
                if score is None:
                    continue
                if best_score is None or score > best_score:
                    best_score = score
            members.append(
                GraphEntityMember(
                    graph_id=ref.graph_id,
                    entity=ref.entity,
                    similarity=best_score,
                )
            )
        clusters.append(EntityCluster(members=members))

    logger.info(
        "Aligned %s entities into %s clusters across %s graphs",
        len(refs),
        len(clusters),
        len(graphs),
    )
    return EntityAlignmentResult(
        regime=regime,
        similarity_threshold=self.similarity_threshold,
        entity_count=len(refs),
        cluster_count=len(clusters),
        clusters=clusters,
    )

EntityAlignmentResult

Bases: BaseModel

Source code in ontocast/tool/agg/match_models.py

class EntityAlignmentResult(BaseModel):
    regime: MatchRegime
    similarity_threshold: float
    entity_count: int
    cluster_count: int
    clusters: list[EntityCluster] = Field(default_factory=list)

    model_config = ConfigDict(arbitrary_types_allowed=True)

Attributes

cluster_count instance-attribute

clusters = Field(default_factory=list) class-attribute instance-attribute

entity_count instance-attribute

model_config = ConfigDict(arbitrary_types_allowed=True) class-attribute instance-attribute

regime instance-attribute

similarity_threshold instance-attribute

EntityCluster

Bases: BaseModel

Source code in ontocast/tool/agg/match_models.py

class EntityCluster(BaseModel):
    members: list[GraphEntityMember] = Field(default_factory=list)

    model_config = ConfigDict(arbitrary_types_allowed=True)

Attributes

members = Field(default_factory=list) class-attribute instance-attribute

model_config = ConfigDict(arbitrary_types_allowed=True) class-attribute instance-attribute

EntityMatch

Bases: BaseModel

Source code in ontocast/tool/agg/match_models.py

class EntityMatch(BaseModel):
    predicted_entity: RdfUriRef
    gt_entity: RdfUriRef
    similarity: float

    model_config = ConfigDict(arbitrary_types_allowed=True)

Attributes

gt_entity instance-attribute

model_config = ConfigDict(arbitrary_types_allowed=True) class-attribute instance-attribute

predicted_entity instance-attribute

similarity instance-attribute

EntityRole

Bases: StrEnum

Role of an entity in an RDF graph.

Source code in ontocast/tool/agg/uri_builder.py

class EntityRole(StrEnum):
    """Role of an entity in an RDF graph."""

    CLASS = "class"
    PROPERTY = "property"
    INSTANCE = "instance"

Attributes

CLASS = 'class' class-attribute instance-attribute

INSTANCE = 'instance' class-attribute instance-attribute

PROPERTY = 'property' class-attribute instance-attribute

GraphEntityMember

Bases: BaseModel

Source code in ontocast/tool/agg/match_models.py

class GraphEntityMember(BaseModel):
    graph_id: str
    entity: RdfUriRef
    similarity: float | None = None

    model_config = ConfigDict(arbitrary_types_allowed=True)

Attributes

entity instance-attribute

graph_id instance-attribute

model_config = ConfigDict(arbitrary_types_allowed=True) class-attribute instance-attribute

similarity = None class-attribute instance-attribute

MatchMetrics

Bases: BaseModel

Triple, fact and entity scores; a ratio is None where undefined.

Source code in ontocast/tool/agg/match_models.py

class MatchMetrics(BaseModel):
    """Triple, fact and entity scores; a ratio is ``None`` where undefined."""

    precision: float | None
    recall: float | None
    f1: float | None
    true_positives: int
    false_positives: int
    false_negatives: int
    predicted_count: int
    ground_truth_count: int
    entity_precision: float | None
    entity_recall: float | None
    entity_f1: float | None
    entity_true_positives: int
    entity_false_positives: int
    entity_false_negatives: int
    domain_entity_matches: int
    fact_precision: float | None
    fact_recall: float | None
    fact_f1: float | None
    fact_true_positives: int
    fact_false_positives: int
    fact_false_negatives: int
    fact_predicted_count: int
    fact_ground_truth_count: int

Attributes

domain_entity_matches instance-attribute

entity_f1 instance-attribute

entity_false_negatives instance-attribute

entity_false_positives instance-attribute

entity_precision instance-attribute

entity_recall instance-attribute

entity_true_positives instance-attribute

f1 instance-attribute

fact_f1 instance-attribute

fact_false_negatives instance-attribute

fact_false_positives instance-attribute

fact_ground_truth_count instance-attribute

fact_precision instance-attribute

fact_predicted_count instance-attribute

fact_recall instance-attribute

fact_true_positives instance-attribute

false_negatives instance-attribute

false_positives instance-attribute

ground_truth_count instance-attribute

precision instance-attribute

predicted_count instance-attribute

recall instance-attribute

true_positives instance-attribute

MatchRegime

Bases: StrEnum

Source code in ontocast/tool/agg/match_models.py

class MatchRegime(StrEnum):
    ONTOLOGY_LOOSE = "ontology_loose"
    ONTOLOGY_STRICT = "ontology_strict"

Attributes

ONTOLOGY_LOOSE = 'ontology_loose' class-attribute instance-attribute

ONTOLOGY_STRICT = 'ontology_strict' class-attribute instance-attribute

TaggedGraph

Bases: BaseModel

Source code in ontocast/tool/agg/match_models.py

class TaggedGraph(BaseModel):
    id: str
    graph: RDFGraph

    model_config = ConfigDict(arbitrary_types_allowed=True)

Attributes

graph instance-attribute

id instance-attribute

model_config = ConfigDict(arbitrary_types_allowed=True) class-attribute instance-attribute

TripleSetEvaluator

Compute PR/F1 metrics for aligned predicted and ground-truth graphs.

Source code in ontocast/tool/agg/triple_evaluator.py

class TripleSetEvaluator:
    """Compute PR/F1 metrics for aligned predicted and ground-truth graphs."""

    def evaluate(
        self,
        predicted_graph: RDFGraph,
        gt_graph: RDFGraph,
        entity_matches: list[EntityMatch],
    ) -> MatchMetrics:
        predicted_to_gt = {
            as_uri_ref(matched.predicted_entity): as_uri_ref(matched.gt_entity)
            for matched in entity_matches
        }

        raw_predicted = project_triples(predicted_graph, predicted_to_gt)
        raw_ground_truth = set(gt_graph)

        predicted = prepare_metric_triples(raw_predicted)
        ground_truth = prepare_metric_triples(raw_ground_truth)

        true_positives = len(predicted & ground_truth)
        false_positives = len(predicted - ground_truth)
        false_negatives = len(ground_truth - predicted)
        precision, recall, f1 = compute_prf(
            true_positives,
            len(predicted),
            len(ground_truth),
        )

        ontology_entities = collect_ontology_entities(predicted | ground_truth)
        predicted_facts = prepare_fact_triples(predicted, ontology_entities)
        ground_truth_facts = prepare_fact_triples(ground_truth, ontology_entities)
        fact_true_positives = len(predicted_facts & ground_truth_facts)
        fact_false_positives = len(predicted_facts - ground_truth_facts)
        fact_false_negatives = len(ground_truth_facts - predicted_facts)
        fact_precision, fact_recall, fact_f1 = compute_prf(
            fact_true_positives,
            len(predicted_facts),
            len(ground_truth_facts),
        )

        predicted_entities = set(extract_instance_entities(predicted_graph))
        gt_entities = set(extract_instance_entities(gt_graph))
        instance_matches = [
            matched
            for matched in entity_matches
            if as_uri_ref(matched.predicted_entity) in predicted_entities
            and as_uri_ref(matched.gt_entity) in gt_entities
        ]
        matched_predicted = {
            as_uri_ref(matched.predicted_entity) for matched in instance_matches
        }
        matched_gt = {as_uri_ref(matched.gt_entity) for matched in instance_matches}
        entity_true_positives = len(instance_matches)
        entity_false_positives = len(predicted_entities - matched_predicted)
        entity_false_negatives = len(gt_entities - matched_gt)
        entity_precision, entity_recall, entity_f1 = compute_prf(
            entity_true_positives,
            len(predicted_entities),
            len(gt_entities),
        )
        domain_entity_matches = count_domain_entity_matches(entity_matches)

        return MatchMetrics(
            precision=precision,
            recall=recall,
            f1=f1,
            true_positives=true_positives,
            false_positives=false_positives,
            false_negatives=false_negatives,
            predicted_count=len(predicted),
            ground_truth_count=len(ground_truth),
            entity_precision=entity_precision,
            entity_recall=entity_recall,
            entity_f1=entity_f1,
            entity_true_positives=entity_true_positives,
            entity_false_positives=entity_false_positives,
            entity_false_negatives=entity_false_negatives,
            domain_entity_matches=domain_entity_matches,
            fact_precision=fact_precision,
            fact_recall=fact_recall,
            fact_f1=fact_f1,
            fact_true_positives=fact_true_positives,
            fact_false_positives=fact_false_positives,
            fact_false_negatives=fact_false_negatives,
            fact_predicted_count=len(predicted_facts),
            fact_ground_truth_count=len(ground_truth_facts),
        )

Methods:

evaluate(predicted_graph, gt_graph, entity_matches)

Source code in ontocast/tool/agg/triple_evaluator.py

def evaluate(
    self,
    predicted_graph: RDFGraph,
    gt_graph: RDFGraph,
    entity_matches: list[EntityMatch],
) -> MatchMetrics:
    predicted_to_gt = {
        as_uri_ref(matched.predicted_entity): as_uri_ref(matched.gt_entity)
        for matched in entity_matches
    }

    raw_predicted = project_triples(predicted_graph, predicted_to_gt)
    raw_ground_truth = set(gt_graph)

    predicted = prepare_metric_triples(raw_predicted)
    ground_truth = prepare_metric_triples(raw_ground_truth)

    true_positives = len(predicted & ground_truth)
    false_positives = len(predicted - ground_truth)
    false_negatives = len(ground_truth - predicted)
    precision, recall, f1 = compute_prf(
        true_positives,
        len(predicted),
        len(ground_truth),
    )

    ontology_entities = collect_ontology_entities(predicted | ground_truth)
    predicted_facts = prepare_fact_triples(predicted, ontology_entities)
    ground_truth_facts = prepare_fact_triples(ground_truth, ontology_entities)
    fact_true_positives = len(predicted_facts & ground_truth_facts)
    fact_false_positives = len(predicted_facts - ground_truth_facts)
    fact_false_negatives = len(ground_truth_facts - predicted_facts)
    fact_precision, fact_recall, fact_f1 = compute_prf(
        fact_true_positives,
        len(predicted_facts),
        len(ground_truth_facts),
    )

    predicted_entities = set(extract_instance_entities(predicted_graph))
    gt_entities = set(extract_instance_entities(gt_graph))
    instance_matches = [
        matched
        for matched in entity_matches
        if as_uri_ref(matched.predicted_entity) in predicted_entities
        and as_uri_ref(matched.gt_entity) in gt_entities
    ]
    matched_predicted = {
        as_uri_ref(matched.predicted_entity) for matched in instance_matches
    }
    matched_gt = {as_uri_ref(matched.gt_entity) for matched in instance_matches}
    entity_true_positives = len(instance_matches)
    entity_false_positives = len(predicted_entities - matched_predicted)
    entity_false_negatives = len(gt_entities - matched_gt)
    entity_precision, entity_recall, entity_f1 = compute_prf(
        entity_true_positives,
        len(predicted_entities),
        len(gt_entities),
    )
    domain_entity_matches = count_domain_entity_matches(entity_matches)

    return MatchMetrics(
        precision=precision,
        recall=recall,
        f1=f1,
        true_positives=true_positives,
        false_positives=false_positives,
        false_negatives=false_negatives,
        predicted_count=len(predicted),
        ground_truth_count=len(ground_truth),
        entity_precision=entity_precision,
        entity_recall=entity_recall,
        entity_f1=entity_f1,
        entity_true_positives=entity_true_positives,
        entity_false_positives=entity_false_positives,
        entity_false_negatives=entity_false_negatives,
        domain_entity_matches=domain_entity_matches,
        fact_precision=fact_precision,
        fact_recall=fact_recall,
        fact_f1=fact_f1,
        fact_true_positives=fact_true_positives,
        fact_false_positives=fact_false_positives,
        fact_false_negatives=fact_false_negatives,
        fact_predicted_count=len(predicted_facts),
        fact_ground_truth_count=len(ground_truth_facts),
    )

URIBuilder

Build normalized URIs for all entities following RDF naming conventions.

• Fact entities (under base_iri) get new URIs under base_iri.
• Ontology entities (everything else) are preserved as-is.

Source code in ontocast/tool/agg/uri_builder.py

class URIBuilder:
    """Build normalized URIs for all entities following RDF naming conventions.

    - **Fact entities** (under *base_iri*) get new URIs under *base_iri*.
    - **Ontology entities** (everything else) are preserved as-is.
    """

    def __init__(
        self,
        base_iri: str = DEFAULT_IRI,
    ):
        """Initialise the builder.

        Args:
            base_iri: Base IRI for fact entities (default ``DEFAULT_IRI``).
                Entities under this namespace are facts; everything else is
                treated as an ontology entity.
        """
        self.base_iri = normalize_namespace_iri(base_iri, context="facts")
        self._used_uris: set[URIRef] = set()

    # ------------------------------------------------------------------
    # helpers
    # ------------------------------------------------------------------

    def is_ontology_entity(self, entity: URIRef) -> bool:
        """Return True if *entity* does **not** belong to the facts namespace."""
        return not is_in_namespace(str(entity), self.base_iri, context="facts")

    @staticmethod
    def _extract_namespace(entity: URIRef) -> str:
        """Extract the namespace part of a URI (everything before the local name).

        For ``http://example.org/ns#Foo`` returns ``http://example.org/ns#``.
        For ``http://example.org/ns/Foo`` returns ``http://example.org/ns/``.
        """
        namespace, _ = split_namespace_local(str(entity))
        return namespace or str(entity)

    def _ensure_unique_uri(self, base: str, local_name: str) -> URIRef:
        """Return a unique URI under *base* for *local_name*."""
        candidate = URIRef(join_namespace_local(base, local_name, context="auto"))
        if candidate not in self._used_uris:
            self._used_uris.add(candidate)
            return candidate

        counter = 1
        while True:
            candidate = URIRef(
                join_namespace_local(base, f"{local_name}_{counter}", context="auto")
            )
            if candidate not in self._used_uris:
                self._used_uris.add(candidate)
                return candidate
            counter += 1

    # ------------------------------------------------------------------
    # public API
    # ------------------------------------------------------------------

    def build_uri(
        self,
        entity: URIRef,
        representation: EntityRepresentation,
        role: EntityRole | str,
        target_iri: URIRef | str | None = None,
        is_ontology_entity: bool | None = None,
    ) -> URIRef:
        """Build a normalised URI for a single entity.

        Fact entities are normalised and placed under *target_iri* (falling
        back to *base_iri*). Ontology entities are preserved as-is.

        Args:
            entity: Original entity URI.
            representation: Entity representation with metadata.
            role: Entity role (an :class:`EntityRole` value).
            target_iri: Optional document IRI to use as namespace for fact
                entities instead of the default *base_iri*.  When chunks carry
                different ``doc_iri`` values the caller passes the appropriate
                one here so that each fact is placed under its document
                namespace.
            is_ontology_entity: Explicit ontology/fact classification.  When
                provided this takes precedence over namespace-based inference.

        Returns:
            Normalised URI.
        """
        is_ontology = (
            self.is_ontology_entity(entity)
            if is_ontology_entity is None
            else is_ontology_entity
        )

        if is_ontology:
            return entity

        local_name = normalize_local_name(representation, role)
        base = (
            normalize_namespace_iri(str(target_iri), context="facts")
            if target_iri
            else self.base_iri
        )
        return self._ensure_unique_uri(base=base, local_name=local_name)

    def create_entity_uri_mapping(
        self,
        identity_mapping: dict[URIRef, URIRef],
        representations: dict[URIRef, EntityRepresentation],
        entity_doc_iris: dict[URIRef, URIRef],
        entity_is_ontology: dict[URIRef, bool],
    ) -> dict[URIRef, URIRef]:
        """Create final URI mapping from identity mapping + namespace policy.

        This method decouples canonical identity choice from URI surface choice:
        identity mapping decides *what* is the same entity, while this method
        decides *how* each source entity should be rendered as a final URI.
        Fact entities are always rendered in their source ``doc_iri`` namespace.
        Ontology entities are preserved as their canonical URI.

        Entities are minted in a fixed order -- by unscoped IRI, then by unit
        index -- so when two unmerged entities compete for one local name the
        counter suffix lands on the same one every run, rather than on
        whichever a set iteration happened to yield first.

        Args:
            identity_mapping: Mapping ``entity -> canonical_entity``.
            representations: All entity representations.
            entity_doc_iris: Mapping from source entity to source ``doc_iri``.
            entity_is_ontology: Classification map where ``True`` means the
                canonical entity should stay in ontology space.

        Returns:
            Mapping ``entity -> final_uri``.
        """
        self._used_uris.clear()
        mapping: dict[URIRef, URIRef] = {}
        canonical_cache: dict[tuple[URIRef, str], URIRef] = {}

        def mint_order(item: tuple[URIRef, URIRef]) -> tuple[str, int, str]:
            text = str(item[0])
            index = unit_scope_index(text)
            return (strip_unit_scope(text), -1 if index is None else index, text)

        for entity, canonical in sorted(identity_mapping.items(), key=mint_order):
            rep = representations.get(canonical)
            if rep is None:
                mapping[entity] = entity
                continue

            role = rep.role if rep.role is not None else EntityRole.INSTANCE
            is_ontology = entity_is_ontology.get(
                canonical, self.is_ontology_entity(canonical)
            )
            if is_ontology:
                mapping[entity] = canonical
                continue

            doc_iri = entity_doc_iris.get(entity)
            base = (
                normalize_namespace_iri(str(doc_iri), context="facts")
                if doc_iri
                else self.base_iri
            )
            cache_key = (canonical, base)
            if cache_key in canonical_cache:
                mapping[entity] = canonical_cache[cache_key]
                continue

            canonical_uri = self.build_uri(
                canonical,
                rep,
                role,
                target_iri=doc_iri,
                is_ontology_entity=False,
            )
            canonical_cache[cache_key] = canonical_uri
            mapping[entity] = canonical_uri

        normalised = sum(1 for e, u in mapping.items() if e != u)
        logger.info(
            f"Built URI mapping: {len(mapping)} entities, {normalised} normalised"
        )
        return mapping

    @staticmethod
    def compose_mappings(
        clustering_mapping: dict[URIRef, URIRef],
        uri_mapping: dict[URIRef, URIRef],
    ) -> dict[URIRef, URIRef]:
        """Compose clustering and URI mappings.

        ``e → representative(e) → normalised_uri(representative(e))``

        Args:
            clustering_mapping: ``e → e_rep``.
            uri_mapping: ``e_rep → final_uri``.

        Returns:
            Composed mapping ``e → final_uri``.
        """
        composed = {
            original: uri_mapping.get(representative, representative)
            for original, representative in clustering_mapping.items()
        }
        logger.info(
            f"Composed mapping: {len(composed)} entities → "
            f"{len(set(composed.values()))} final URIs"
        )
        return composed

Attributes

base_iri = normalize_namespace_iri(base_iri, context='facts') instance-attribute

Methods:

__init__(base_iri=DEFAULT_IRI)

Initialise the builder.

Parameters:

Name	Type	Description	Default
base_iri	str	Base IRI for fact entities (default DEFAULT_IRI). Entities under this namespace are facts; everything else is treated as an ontology entity.	DEFAULT_IRI

Source code in ontocast/tool/agg/uri_builder.py

def __init__(
    self,
    base_iri: str = DEFAULT_IRI,
):
    """Initialise the builder.

    Args:
        base_iri: Base IRI for fact entities (default ``DEFAULT_IRI``).
            Entities under this namespace are facts; everything else is
            treated as an ontology entity.
    """
    self.base_iri = normalize_namespace_iri(base_iri, context="facts")
    self._used_uris: set[URIRef] = set()

build_uri(entity, representation, role, target_iri=None, is_ontology_entity=None)

Build a normalised URI for a single entity.

Fact entities are normalised and placed under target_iri (falling back to base_iri). Ontology entities are preserved as-is.

Parameters:

Name	Type	Description	Default
entity	URIRef	Original entity URI.	required
representation	EntityRepresentation	Entity representation with metadata.	required
role	EntityRole | str	Entity role (an :class:EntityRole value).	required
target_iri	URIRef | str | None	Optional document IRI to use as namespace for fact entities instead of the default base_iri. When chunks carry different doc_iri values the caller passes the appropriate one here so that each fact is placed under its document namespace.	None
is_ontology_entity	bool | None	Explicit ontology/fact classification. When provided this takes precedence over namespace-based inference.	None

Returns:

Type	Description
URIRef	Normalised URI.

Source code in ontocast/tool/agg/uri_builder.py

def build_uri(
    self,
    entity: URIRef,
    representation: EntityRepresentation,
    role: EntityRole | str,
    target_iri: URIRef | str | None = None,
    is_ontology_entity: bool | None = None,
) -> URIRef:
    """Build a normalised URI for a single entity.

    Fact entities are normalised and placed under *target_iri* (falling
    back to *base_iri*). Ontology entities are preserved as-is.

    Args:
        entity: Original entity URI.
        representation: Entity representation with metadata.
        role: Entity role (an :class:`EntityRole` value).
        target_iri: Optional document IRI to use as namespace for fact
            entities instead of the default *base_iri*.  When chunks carry
            different ``doc_iri`` values the caller passes the appropriate
            one here so that each fact is placed under its document
            namespace.
        is_ontology_entity: Explicit ontology/fact classification.  When
            provided this takes precedence over namespace-based inference.

    Returns:
        Normalised URI.
    """
    is_ontology = (
        self.is_ontology_entity(entity)
        if is_ontology_entity is None
        else is_ontology_entity
    )

    if is_ontology:
        return entity

    local_name = normalize_local_name(representation, role)
    base = (
        normalize_namespace_iri(str(target_iri), context="facts")
        if target_iri
        else self.base_iri
    )
    return self._ensure_unique_uri(base=base, local_name=local_name)

compose_mappings(clustering_mapping, uri_mapping) staticmethod

Compose clustering and URI mappings.

e → representative(e) → normalised_uri(representative(e))

Parameters:

Name	Type	Description	Default
clustering_mapping	dict[URIRef, URIRef]	e → e_rep.	required
uri_mapping	dict[URIRef, URIRef]	e_rep → final_uri.	required

Returns:

Type	Description
dict[URIRef, URIRef]	Composed mapping e → final_uri.

Source code in ontocast/tool/agg/uri_builder.py

@staticmethod
def compose_mappings(
    clustering_mapping: dict[URIRef, URIRef],
    uri_mapping: dict[URIRef, URIRef],
) -> dict[URIRef, URIRef]:
    """Compose clustering and URI mappings.

    ``e → representative(e) → normalised_uri(representative(e))``

    Args:
        clustering_mapping: ``e → e_rep``.
        uri_mapping: ``e_rep → final_uri``.

    Returns:
        Composed mapping ``e → final_uri``.
    """
    composed = {
        original: uri_mapping.get(representative, representative)
        for original, representative in clustering_mapping.items()
    }
    logger.info(
        f"Composed mapping: {len(composed)} entities → "
        f"{len(set(composed.values()))} final URIs"
    )
    return composed

create_entity_uri_mapping(identity_mapping, representations, entity_doc_iris, entity_is_ontology)

Create final URI mapping from identity mapping + namespace policy.

This method decouples canonical identity choice from URI surface choice: identity mapping decides what is the same entity, while this method decides how each source entity should be rendered as a final URI. Fact entities are always rendered in their source doc_iri namespace. Ontology entities are preserved as their canonical URI.

Entities are minted in a fixed order -- by unscoped IRI, then by unit index -- so when two unmerged entities compete for one local name the counter suffix lands on the same one every run, rather than on whichever a set iteration happened to yield first.

Parameters:

Name	Type	Description	Default
identity_mapping	dict[URIRef, URIRef]	Mapping entity -> canonical_entity.	required
representations	dict[URIRef, EntityRepresentation]	All entity representations.	required
entity_doc_iris	dict[URIRef, URIRef]	Mapping from source entity to source doc_iri.	required
entity_is_ontology	dict[URIRef, bool]	Classification map where True means the canonical entity should stay in ontology space.	required

Returns:

Type	Description
dict[URIRef, URIRef]	Mapping entity -> final_uri.

Source code in ontocast/tool/agg/uri_builder.py

def create_entity_uri_mapping(
    self,
    identity_mapping: dict[URIRef, URIRef],
    representations: dict[URIRef, EntityRepresentation],
    entity_doc_iris: dict[URIRef, URIRef],
    entity_is_ontology: dict[URIRef, bool],
) -> dict[URIRef, URIRef]:
    """Create final URI mapping from identity mapping + namespace policy.

    This method decouples canonical identity choice from URI surface choice:
    identity mapping decides *what* is the same entity, while this method
    decides *how* each source entity should be rendered as a final URI.
    Fact entities are always rendered in their source ``doc_iri`` namespace.
    Ontology entities are preserved as their canonical URI.

    Entities are minted in a fixed order -- by unscoped IRI, then by unit
    index -- so when two unmerged entities compete for one local name the
    counter suffix lands on the same one every run, rather than on
    whichever a set iteration happened to yield first.

    Args:
        identity_mapping: Mapping ``entity -> canonical_entity``.
        representations: All entity representations.
        entity_doc_iris: Mapping from source entity to source ``doc_iri``.
        entity_is_ontology: Classification map where ``True`` means the
            canonical entity should stay in ontology space.

    Returns:
        Mapping ``entity -> final_uri``.
    """
    self._used_uris.clear()
    mapping: dict[URIRef, URIRef] = {}
    canonical_cache: dict[tuple[URIRef, str], URIRef] = {}

    def mint_order(item: tuple[URIRef, URIRef]) -> tuple[str, int, str]:
        text = str(item[0])
        index = unit_scope_index(text)
        return (strip_unit_scope(text), -1 if index is None else index, text)

    for entity, canonical in sorted(identity_mapping.items(), key=mint_order):
        rep = representations.get(canonical)
        if rep is None:
            mapping[entity] = entity
            continue

        role = rep.role if rep.role is not None else EntityRole.INSTANCE
        is_ontology = entity_is_ontology.get(
            canonical, self.is_ontology_entity(canonical)
        )
        if is_ontology:
            mapping[entity] = canonical
            continue

        doc_iri = entity_doc_iris.get(entity)
        base = (
            normalize_namespace_iri(str(doc_iri), context="facts")
            if doc_iri
            else self.base_iri
        )
        cache_key = (canonical, base)
        if cache_key in canonical_cache:
            mapping[entity] = canonical_cache[cache_key]
            continue

        canonical_uri = self.build_uri(
            canonical,
            rep,
            role,
            target_iri=doc_iri,
            is_ontology_entity=False,
        )
        canonical_cache[cache_key] = canonical_uri
        mapping[entity] = canonical_uri

    normalised = sum(1 for e, u in mapping.items() if e != u)
    logger.info(
        f"Built URI mapping: {len(mapping)} entities, {normalised} normalised"
    )
    return mapping

is_ontology_entity(entity)

Return True if entity does not belong to the facts namespace.

Source code in ontocast/tool/agg/uri_builder.py

def is_ontology_entity(self, entity: URIRef) -> bool:
    """Return True if *entity* does **not** belong to the facts namespace."""
    return not is_in_namespace(str(entity), self.base_iri, context="facts")

Functions:

apply_document_metadata_provenance(doc_iri, metadata, graph, *, entity_namespace=None)

Emit caller-asserted document identity triples on doc_iri.

Document-level identity is provenance-adjacent but intentionally kept on the facts graph (survives chunk-level strip_provenance) so query/RAG clients can filter by DOI, business id, filename, etc.

Business-oriented keys (author, project, and any non-reserved key) mint typed RDF entities under entity_namespace (defaults to the document facts namespace) so they are SPARQL-discoverable via rdf:type.

Registry keys are matched via :func:_resolve_metadata_key (case / separator / optional id affix for identifier and source keys). Keys with an identifier-shaped affix (id, ref, no, key, …) that do not resolve to a registry entry become structured dcterms:identifier blank nodes, or attach to a companion entity-link stem when one was minted in the same payload (e.g. project + project_id).

Source code in ontocast/tool/agg/aggregate.py

def apply_document_metadata_provenance(
    doc_iri: URIRef,
    metadata: dict[str, Any],
    graph: RDFGraph,
    *,
    entity_namespace: str | None = None,
) -> None:
    """Emit caller-asserted document identity triples on ``doc_iri``.

    Document-level identity is provenance-adjacent but intentionally kept on the
    facts graph (survives chunk-level ``strip_provenance``) so query/RAG clients
    can filter by DOI, business id, filename, etc.

    Business-oriented keys (``author``, ``project``, and any non-reserved key)
    mint typed RDF entities under ``entity_namespace`` (defaults to the document
    facts namespace) so they are SPARQL-discoverable via ``rdf:type``.

    Registry keys are matched via :func:`_resolve_metadata_key` (case /
    separator / optional ``id`` affix for identifier and source keys). Keys with
    an identifier-shaped affix (``id``, ``ref``, ``no``, ``key``, …) that do not
    resolve to a registry entry become structured ``dcterms:identifier`` blank
    nodes, or attach to a companion entity-link stem when one was minted in the
    same payload (e.g. ``project`` + ``project_id``).
    """
    if not metadata:
        return

    graph.bind("prov", str(PROV))
    graph.bind("foaf", str(FOAF))
    graph.bind("dcterms", str(DCTERMS))
    graph.bind("owl", str(OWL))
    graph.bind("rdfs", str(RDFS))
    graph.bind("schema", str(SCHEMA))

    graph.add((doc_iri, RDF.type, PROV.Entity))
    graph.add((doc_iri, RDF.type, FOAF.Document))

    ns = entity_namespace or normalize_namespace_iri(str(doc_iri), context="facts")
    entity_iri_by_stem: dict[str, URIRef] = {}
    deferred_affix: list[tuple[str, object, tuple[str, str]]] = []

    for raw_key, value in metadata.items():
        if value is None or value == "":
            continue
        key = _resolve_metadata_key(raw_key)
        if key == "stable_source_iri":
            graph.add((doc_iri, OWL.sameAs, _as_iri_or_literal(value)))
            continue
        if key in _DOC_METADATA_SOURCE_KEYS:
            graph.add((doc_iri, DCTERMS.source, _as_iri_or_literal(value)))
            continue
        if key in _DOC_METADATA_IDENTIFIER_KEYS:
            graph.add((doc_iri, DCTERMS.identifier, Literal(str(value))))
            continue
        if key == "identifiers":
            items = value if isinstance(value, list) else [value]
            for item in items:
                if not isinstance(item, dict):
                    continue
                scheme = item.get("scheme") or item.get("type")
                val = item.get("value")
                if scheme and val is not None and val != "":
                    _add_structured_identifier(
                        graph, doc_iri, scheme=str(scheme), value=val
                    )
            continue
        if key in _DOC_METADATA_FIRST_CLASS:
            predicate = _DOC_METADATA_FIRST_CLASS[key]
            if isinstance(value, list):
                for item in value:
                    if item is not None and item != "":
                        graph.add((doc_iri, predicate, Literal(str(item))))
            else:
                graph.add((doc_iri, predicate, Literal(str(value))))
            continue

        if key in _DOC_METADATA_ENTITY_LINKS:
            link, default_type = _DOC_METADATA_ENTITY_LINKS[key]
            minted = _emit_metadata_entities(
                graph,
                doc_iri,
                ns,
                link=link,
                default_type=default_type,
                value=value,
            )
            # Companion ``*_id`` attachment only for a singular non-list entity.
            if not isinstance(value, list) and len(minted) == 1:
                entity_iri_by_stem[key] = minted[0]
            continue

        split = _split_identifier_affix(key)
        if split is not None:
            deferred_affix.append((key, value, split))
            continue

        _emit_metadata_entities(
            graph,
            doc_iri,
            ns,
            link=_DEFAULT_ENTITY_LINK_PREDICATE,
            default_type=_DEFAULT_ENTITY_TYPE,
            value=value,
        )

    for _key, value, (stem, _affix) in deferred_affix:
        entity_iri = entity_iri_by_stem.get(stem)
        if entity_iri is not None:
            if isinstance(value, list):
                for item in value:
                    if item is not None and item != "":
                        graph.add((entity_iri, DCTERMS.identifier, Literal(str(item))))
            else:
                graph.add((entity_iri, DCTERMS.identifier, Literal(str(value))))
            continue
        if isinstance(value, list):
            for item in value:
                if item is not None and item != "":
                    _add_structured_identifier(graph, doc_iri, scheme=stem, value=item)
        else:
            _add_structured_identifier(graph, doc_iri, scheme=stem, value=value)

derive_pair_matches(clusters, predicted_graph_id, gt_graph_id, *, similarity_threshold=0.0)

Map global clusters to 1:1 predicted↔gt entity matches for one graph pair.

Source code in ontocast/tool/agg/match_derivation.py

def derive_pair_matches(
    clusters: list[EntityCluster],
    predicted_graph_id: str,
    gt_graph_id: str,
    *,
    similarity_threshold: float = 0.0,
) -> list[EntityMatch]:
    """Map global clusters to 1:1 predicted↔gt entity matches for one graph pair."""
    matches: list[EntityMatch] = []
    for cluster in clusters:
        predicted_members = _members_for_graph(cluster, predicted_graph_id)
        gt_members = _members_for_graph(cluster, gt_graph_id)
        if not predicted_members or not gt_members:
            continue

        if len(predicted_members) == 1 and len(gt_members) == 1:
            predicted_entity, _ = predicted_members[0]
            gt_entity, predicted_similarity = gt_members[0]
            gt_similarity = gt_members[0][1]
            score = predicted_similarity or gt_similarity or 1.0
            matches.append(
                EntityMatch(
                    predicted_entity=predicted_entity,
                    gt_entity=gt_entity,
                    similarity=score,
                )
            )
            continue

        candidates: list[EntityMatch] = []
        for (predicted_entity, predicted_score), (gt_entity, gt_score) in product(
            predicted_members, gt_members
        ):
            score = predicted_score or gt_score or 1.0
            if score < similarity_threshold:
                continue
            candidates.append(
                EntityMatch(
                    predicted_entity=predicted_entity,
                    gt_entity=gt_entity,
                    similarity=score,
                )
            )
        candidates.sort(
            key=lambda item: (
                -item.similarity,
                str(item.predicted_entity),
                str(item.gt_entity),
            )
        )
        matches.extend(greedy_one_to_one(candidates))

    matches.sort(
        key=lambda item: (
            -item.similarity,
            str(item.predicted_entity),
            str(item.gt_entity),
        )
    )
    return matches