ontocast.tool.agg

Embedding-based aggregation pipeline for RDF content unit graphs.

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)

        # Pipeline components (EntityClusterer imports sklearn/ST lazily)
        from .clustering import EntityClusterer

        self.normalizer = EntityNormalizer(facts_iri=self.base_iri)
        self.clusterer = EntityClusterer(
            embedding_model=cfg.embedding_model,
            similarity_threshold=cfg.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 _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]:
        return {token for token in text.split() if len(token) > 2}

    @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(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:
            return True
        return bool(left_types & right_types)

    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.normalizer.normalize_string(label)
            for label in left_rep.labels + left_rep.alt_labels
            if label.strip()
        }
        right_label_tokens = {
            self.normalizer.normalize_string(label)
            for label in right_rep.labels + right_rep.alt_labels
            if label.strip()
        }
        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

    @staticmethod
    def _tokens_alias_compatible(left: str, right: str) -> bool:
        """Exact token match, or a (possibly dotted) single-char initial of it."""
        if left == right:
            return True
        shorter, longer = (left, right) if len(left) <= len(right) else (right, left)
        return len(shorter) == 1 and longer.startswith(shorter)

    @classmethod
    def _labels_alias_with_initials(
        cls,
        left_labels: set[str],
        right_labels: set[str],
    ) -> bool:
        """True when a label pair matches token-injectively allowing initials.

        Every token of the shorter label must match a distinct token of the
        longer one (exactly, or as a single-character initial), and at least
        one matched token must be a full word (len > 2). Generic abbreviation
        structure — nothing person-specific.
        """
        for left_label in left_labels:
            left_tokens = left_label.split()
            for right_label in right_labels:
                right_tokens = right_label.split()
                if not left_tokens or not right_tokens:
                    continue
                shorter, longer = (
                    (left_tokens, right_tokens)
                    if len(left_tokens) <= len(right_tokens)
                    else (right_tokens, left_tokens)
                )
                available = list(longer)
                shared_full_token = False
                matched_all = True
                for token in shorter:
                    match_index = next(
                        (
                            index
                            for index, candidate in enumerate(available)
                            if cls._tokens_alias_compatible(token, candidate)
                        ),
                        None,
                    )
                    if match_index is None:
                        matched_all = False
                        break
                    if token == available[match_index] and len(token) > 2:
                        shared_full_token = True
                    del available[match_index]
                if matched_all and shared_full_token:
                    return True
        return False

    @classmethod
    def _string_values_compatible(cls, left: str, right: str) -> bool:
        """Compatible when equal, prefix-related, or initial-abbreviations."""
        if left == right:
            return True
        if left.startswith(right) or right.startswith(left):
            return True
        return cls._labels_alias_with_initials({left}, {right})

    @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.normalizer.normalize_string(label)
            for label in left_rep.labels + left_rep.alt_labels
            if label.strip()
        }
        right_labels = {
            self.normalizer.normalize_string(label)
            for label in right_rep.labels + right_rep.alt_labels
            if label.strip()
        }
        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 _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,
    ) -> bool:
        pair = frozenset((left, right))
        if direct_relation_pairs is not None and pair in direct_relation_pairs:
            return False
        if guard_context is not None:
            if pair in guard_context.sibling_pairs:
                return False
            left_rep = representations.get(left)
            right_rep = representations.get(right)
            if left_rep is not None and right_rep is not None:
                if self._have_conflicting_literals(left_rep, right_rep):
                    return False
                if self._have_conflicting_functional_objects(
                    left_rep, right_rep, guard_context.functional_predicates
                ):
                    return False
        return (
            self._are_roles_compatible(left, right, representations)
            and self._are_types_compatible(left, right, representations)
            and self._are_lexical_aliases(left, right, representations)
        )

    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] = {}
        original_threshold = self.clusterer.similarity_threshold
        self.clusterer.similarity_threshold = self.candidate_similarity_threshold
        try:
            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)
        finally:
            self.clusterer.similarity_threshold = original_threshold
        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:
            if frozenset((left, right)) 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._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 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,
    ) -> 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

            parents: dict[URIRef, URIRef] = {
                entity: entity for entity in candidate_cluster
            }

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

            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
                else:
                    parents[left_root] = right_root

            for left, right in combinations(candidate_cluster, 2):
                score = self._candidate_similarity(left, right, embeddings)
                if score is not None and score < self.candidate_similarity_threshold:
                    # Label-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.
                    if not 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,
                ):
                    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 candidate_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

    @staticmethod
    def _merge_into_context_graph(target: RDFGraph, source: RDFGraph) -> None:
        """Merge source triples/namespaces into a per-entity context graph."""
        target += source

    def _register_entity(
        self,
        *,
        entity: URIRef,
        unit: ContentUnit,
        state: _EntityCollectionState,
    ) -> None:
        """Register one URI entity with merged context and stable classification."""
        state.entities.add(entity)
        state.source_entities.add(entity)
        if entity not in state.entity_graphs:
            state.entity_graphs[entity] = unit.graph.copy()
        else:
            self._merge_into_context_graph(state.entity_graphs[entity], unit.graph)
        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 the graph it was found in and the
        ``doc_iri`` of the :class:`ContentUnit` that produced it.  When an
        entity appears in several units the *last-seen* ``doc_iri`` wins (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.

        Returns:
            Tuple of (
                entities,
                entity_to_graph,
                entity_to_doc_iri,
                entity_to_is_ontology,
            ).
        """
        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 s, p, o in unit.graph:
                if isinstance(s, URIRef) and isinstance(o, URIRef):
                    self._register_direct_relation(state=state, subject=s, obj=o)
                    if isinstance(p, URIRef) and p != RDF.type:
                        state.object_groups.setdefault((s, p), set()).add(o)
                for term in (s, p, o):
                    if isinstance(term, URIRef):
                        self._register_entity(entity=term, unit=unit, state=state)

        return (
            list(state.entities),
            state.source_entities,
            state.entity_graphs,
            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
        guard_context = MergeGuardContext(
            sibling_pairs=build_sibling_pairs(
                object_groups, scope=self.sibling_guard_scope
            ),
            functional_predicates=harvest_max_one_predicates(ontology_graph)
            | 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)
        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,
        )
        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)

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

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

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

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

    # Pipeline components (EntityClusterer imports sklearn/ST lazily)
    from .clustering import EntityClusterer

    self.normalizer = EntityNormalizer(facts_iri=self.base_iri)
    self.clusterer = EntityClusterer(
        embedding_model=cfg.embedding_model,
        similarity_threshold=cfg.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
    guard_context = MergeGuardContext(
        sibling_pairs=build_sibling_pairs(
            object_groups, scope=self.sibling_guard_scope
        ),
        functional_predicates=harvest_max_one_predicates(ontology_graph)
        | 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)
    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,
    )
    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)

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

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

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

    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()
    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 = "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,
        )

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"

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_entities(predicted_graph))
        gt_entities = set(extract_entities(gt_graph))
        matched_predicted = {
            as_uri_ref(matched.predicted_entity) for matched in entity_matches
        }
        matched_gt = {as_uri_ref(matched.gt_entity) for matched in entity_matches}
        entity_true_positives = len(entity_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.

        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] = {}

        for entity, canonical in identity_mapping.items():
            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

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

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.

    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] = {}

    for entity, canonical in identity_mapping.items():
        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")

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