Blocking

base

Abstract interface for blocking strategies.

BlockingStrategy

Bases: ABC

Generates candidate entity pairs prior to full linking.

Blocking reduces the O(n*m) comparison space between two datasets down to a smaller set of plausible candidate pairs.

ranked class-attribute

ranked: bool = False

Whether candidate_pairs() returns each source entity's own candidates best-first. Strategies with a graded notion of relevance built on rank_top_matches (LabelOverlap, EmbeddingSimilarityBlocker) set this True; consumers such as sample_hard_negatives check it to warn rather than silently mine the wrong candidates as "hard" negatives.

candidate_pairs abstractmethod

candidate_pairs(
    dataset1: list[Entity],
    dataset2: list[Entity] | EntitySource,
) -> Iterable[tuple[Entity, Entity]]

Return candidate entity pairs to be scored by a linker.

Parameters:

Name	Type	Description	Default
dataset1	list[Entity]	Entities from the first dataset.	required
dataset2	list[Entity] | EntitySource	Entities from the second dataset, or an EntitySource wrapping a target too large to materialize as a list[Entity]. Not every strategy supports an EntitySource -- one that needs to enumerate the whole target set to build an index (e.g. LabelOverlap, EmbeddingSimilarityBlocker) raises TypeError instead.	required

Returns:

Type	Description
Iterable[tuple[Entity, Entity]]	Candidate (entity1, entity2) pairs. Where a strategy has a
Iterable[tuple[Entity, Entity]]	graded notion of relevance, a given source entity's own
Iterable[tuple[Entity, Entity]]	candidates should appear best-first (see
Iterable[tuple[Entity, Entity]]	ranked) — consumers such as
Iterable[tuple[Entity, Entity]]	sample_hard_negatives
Iterable[tuple[Entity, Entity]]	rely on this ordering.

Source code in src/linkingtk/blocking/base.py

@abstractmethod
def candidate_pairs(
    self, dataset1: list[Entity], dataset2: list[Entity] | EntitySource
) -> Iterable[tuple[Entity, Entity]]:
    """Return candidate entity pairs to be scored by a linker.

    Args:
        dataset1: Entities from the first dataset.
        dataset2: Entities from the second dataset, or an
            [EntitySource][linkingtk.core.source.EntitySource] wrapping a
            target too large to materialize as a ``list[Entity]``. Not
            every strategy supports an ``EntitySource`` -- one that needs
            to enumerate the whole target set to build an index (e.g.
            [LabelOverlap][linkingtk.blocking.label_overlap.LabelOverlap],
            [EmbeddingSimilarityBlocker][linkingtk.blocking.embedding.EmbeddingSimilarityBlocker])
            raises ``TypeError`` instead.

    Returns:
        Candidate ``(entity1, entity2)`` pairs. Where a strategy has a
        graded notion of relevance, a given source entity's own
        candidates should appear best-first (see
        [ranked][linkingtk.blocking.base.BlockingStrategy.ranked]) — consumers such as
        [sample_hard_negatives][linkingtk.blocking.negative_sampling.sample_hard_negatives]
        rely on this ordering.
    """

rank_top_matches

rank_top_matches(
    scores: dict[str, float],
    entities_by_id: dict[str, Entity],
    k: int,
    *,
    descending: bool,
    keep: Callable[[float], bool] | None = None,
) -> list[Entity]

Rank scores (entity id -> score) and return the top k matching entities.

Shared by blocking strategies that score every candidate against a source entity and keep only the best few (LabelOverlap, EmbeddingSimilarityBlocker). Ties are broken by entity id for deterministic output.

Parameters:

Name	Type	Description	Default
scores	dict[str, float]	Candidate entity id -> score.	required
entities_by_id	dict[str, Entity]	Entity id -> Entity, used to resolve the kept ids back to entities.	required
k	int	Maximum number of entities to return.	required
descending	bool	Whether higher scores are better (e.g. similarity) or lower scores are better (e.g. distance).	required
keep	Callable[[float], bool] | None	Optional predicate; candidates for which it returns False are dropped before ranking.	None

Returns:

Type	Description
list[Entity]	Up to k entities, best-scoring first.

Source code in src/linkingtk/blocking/base.py

def rank_top_matches(
    scores: dict[str, float],
    entities_by_id: dict[str, Entity],
    k: int,
    *,
    descending: bool,
    keep: Callable[[float], bool] | None = None,
) -> list[Entity]:
    """Rank ``scores`` (entity id -> score) and return the top ``k`` matching entities.

    Shared by blocking strategies that score every candidate against a
    source entity and keep only the best few
    ([LabelOverlap][linkingtk.blocking.label_overlap.LabelOverlap],
    [EmbeddingSimilarityBlocker][linkingtk.blocking.embedding.EmbeddingSimilarityBlocker]). Ties are
    broken by entity id for deterministic output.

    Args:
        scores: Candidate entity id -> score.
        entities_by_id: Entity id -> ``Entity``, used to resolve the kept
            ids back to entities.
        k: Maximum number of entities to return.
        descending: Whether higher scores are better (e.g. similarity) or
            lower scores are better (e.g. distance).
        keep: Optional predicate; candidates for which it returns
            ``False`` are dropped before ranking.

    Returns:
        Up to ``k`` entities, best-scoring first.
    """
    items: Iterable[tuple[str, float]] = scores.items()
    if keep is not None:
        items = [item for item in items if keep(item[1])]
    sign = -1 if descending else 1
    top = heapq.nsmallest(k, items, key=lambda item: (sign * item[1], item[0]))
    return [entities_by_id[entity_id] for entity_id, _ in top]

exact

Exact-label blocking strategy.

ExactMatch

ExactMatch(top_k: int = 10)

Bases: BlockingStrategy

Blocks entities that share at least one identical label.

This is the default blocking strategy: two entities are considered candidates if any of their labels are exactly equal (language tags are ignored for comparison purposes). Also supports an EntitySource for dataset2: it queries dataset2.search(label) per label instead of enumerating the whole target set, then keeps only the results that actually have a matching label -- same semantics, no materialization required.

Parameters:

Name	Type	Description	Default
top_k	int	Forwarded to dataset2.search(label, top_k=...) when dataset2 is an EntitySource -- ignored for a plain list[Entity] dataset2, which has no such cap. The default (EntitySource.search's own default) is too narrow for a highly polysemous word once every one of its senses is a valid candidate (e.g. WSD against a full WordNet) -- raise this in that case.	10

Source code in src/linkingtk/blocking/exact.py

def __init__(self, top_k: int = 10) -> None:
    self.top_k = top_k

label_overlap

Approximate string matching blocking strategy.

Ported from Naisc's ApproximateStringMatching: https://github.com/insight-centre/naisc/blob/master/naisc-core/src/main/java/org/insightcentre/uld/naisc/blocking/ApproximateStringMatching.java

Two interchangeable metrics are provided, matching the reference implementation:

• "ngram": candidates are ranked by a frequency-weighted count of shared character n-grams (fast, index-based, no edit-distance computation).
• "levenshtein": candidates are ranked by normalized Levenshtein distance, using the PatriciaTrie for an approximate nearest-neighbor search that stays tractable on larger datasets.

Unlike the original, both metrics support an optional threshold to drop weak candidates outright.

LabelOverlap

LabelOverlap(
    metric: Metric = "ngram",
    ngram_size: int = 3,
    max_matches: int = 5,
    lowercase: bool = True,
    threshold: float | None = None,
    queue_max: int | None = None,
    max_ngram_document_frequency: int = 1000,
)

Bases: BlockingStrategy

Blocks entities by approximate string similarity of their labels.

Parameters:

Name	Type	Description	Default
metric	Metric	"ngram" (default) ranks candidates by shared character n-grams; "levenshtein" ranks them by normalized edit distance.	'ngram'
ngram_size	int	Character n-gram length. Only used by the "ngram" metric.	3
max_matches	int	Maximum number of candidates to keep per source entity.	5
lowercase	bool	Whether to lowercase labels before comparing.	True
threshold	float | None	Optional cutoff used to drop weak candidates. For "ngram" this is a minimum score to keep (higher is better, unbounded above). For "levenshtein" this is a maximum normalized edit distance to keep (0 = identical, 1 = completely different).	None
queue_max	int | None	Maximum search-frontier size for the trie-based search. Only used by the "levenshtein" metric. Defaults to max_matches * 20, as in the original implementation.	None
max_ngram_document_frequency	int	N-grams shared by more than this many target entities are dropped as uninformative before matching. Only used by the "ngram" metric.	1000

Source code in src/linkingtk/blocking/label_overlap.py

def __init__(
    self,
    metric: Metric = "ngram",
    ngram_size: int = 3,
    max_matches: int = 5,
    lowercase: bool = True,
    threshold: float | None = None,
    queue_max: int | None = None,
    max_ngram_document_frequency: int = 1000,
) -> None:
    if max_matches < 1:
        raise ValueError("max_matches must be at least 1")
    if ngram_size < 1:
        raise ValueError("ngram_size must be at least 1")
    self.metric = metric
    self.ngram_size = ngram_size
    self.max_matches = max_matches
    self.lowercase = lowercase
    self.threshold = threshold
    self.queue_max = queue_max if queue_max is not None else max_matches * 20
    self.max_ngram_document_frequency = max_ngram_document_frequency

trie

A Patricia (radix) trie with approximate nearest-neighbor lookup.

Ported from Naisc's ApproximateStringMatching.PatriciaTrie: https://github.com/insight-centre/naisc/blob/master/naisc-core/src/main/java/org/insightcentre/uld/naisc/blocking/ApproximateStringMatching.java

The trie stores strings compressed along shared prefixes so that a best-first search, guided by a lower bound on normalized Levenshtein distance, can find the nearest keys to a query without scoring every entry in the trie.

PatriciaTrie

PatriciaTrie()

Bases: Generic[T]

Radix trie mapping strings to (possibly repeated) values.

Source code in src/linkingtk/blocking/trie.py

def __init__(self) -> None:
    self._root: _TrieNode[T] = _TrieNode()

insert

insert(key: str, value: T) -> None

Associate value with key. Empty keys are ignored.

Source code in src/linkingtk/blocking/trie.py

def insert(self, key: str, value: T) -> None:
    """Associate ``value`` with ``key``. Empty keys are ignored."""
    if key:
        self._insert(key, value, self._root)

nearest

nearest(
    query: str, n: int, queue_max: int
) -> list[tuple[T, float]]

Find up to n values whose keys best match query.

Uses best-first search over the trie, pruned by a lower bound on normalized edit distance, so that only a bounded, promising portion of the trie is actually scored.

This is an approximate search, as in the original Naisc implementation: the lower bound assumes the shortest possible completion of each trie prefix, so a true match reached through a much longer key can occasionally be pruned before it is scored. For blocking (where the goal is a good candidate set, not a provably-exact top-n) this is an acceptable and deliberate speed/recall trade-off.

Parameters:

Name	Type	Description	Default
query	str	The string to search for.	required
n	int	Maximum number of results to return.	required
queue_max	int	Maximum size of the search frontier; bounds worst-case search cost at some risk of missing a match.	required

Returns:

Type	Description
list[tuple[T, float]]	Up to n (value, normalized_edit_distance) pairs,
list[tuple[T, float]]	sorted by ascending distance (closest match first).

Source code in src/linkingtk/blocking/trie.py

def nearest(self, query: str, n: int, queue_max: int) -> list[tuple[T, float]]:
    """Find up to ``n`` values whose keys best match ``query``.

    Uses best-first search over the trie, pruned by a lower bound on
    normalized edit distance, so that only a bounded, promising
    portion of the trie is actually scored.

    This is an approximate search, as in the original Naisc
    implementation: the lower bound assumes the shortest possible
    completion of each trie prefix, so a true match reached through a
    much longer key can occasionally be pruned before it is scored.
    For blocking (where the goal is a good candidate set, not a
    provably-exact top-n) this is an acceptable and deliberate
    speed/recall trade-off.

    Args:
        query: The string to search for.
        n: Maximum number of results to return.
        queue_max: Maximum size of the search frontier; bounds
            worst-case search cost at some risk of missing a match.

    Returns:
        Up to ``n`` ``(value, normalized_edit_distance)`` pairs,
        sorted by ascending distance (closest match first).
    """
    if n < 1:
        return []
    counter = itertools.count()
    frontier: list[tuple[float, int, str, _TrieNode[T]]] = []
    for child in self._root.children:
        lb = _normalized_lower_bound(query, child.link)
        heapq.heappush(frontier, (lb, next(counter), child.link, child.node))

    beam: list[tuple[float, int, T]] = []

    def worst() -> float:
        return -beam[0][0] if beam else float("inf")

    while frontier:
        if len(beam) >= n and frontier[0][0] > worst():
            break
        _, _, key, node = heapq.heappop(frontier)
        for child in node.children:
            combined = key + child.link
            child_lb = _normalized_lower_bound(query, combined)
            if len(beam) < n or child_lb < worst():
                heapq.heappush(frontier, (child_lb, next(counter), combined, child.node))
        if len(frontier) > queue_max * 2:
            # nsmallest's output is already sorted, which trivially satisfies
            # the heap invariant, so no separate heapify is needed. Trimming
            # only once the frontier has grown well past queue_max (rather
            # than as soon as it's exceeded) amortizes this O(log) pass
            # across many iterations instead of paying it almost every time.
            frontier[:] = heapq.nsmallest(queue_max, frontier)

        denom = len(query) + len(key)
        score = edit_distance(query, key) / denom if denom else 0.0
        for value in node.values:
            if len(beam) < n or score < worst():
                heapq.heappush(beam, (-score, next(counter), value))
                if len(beam) > n:
                    heapq.heappop(beam)

    results = [(value, -neg_score) for neg_score, _, value in beam]
    results.sort(key=lambda item: item[1])
    return results

edit_distance

edit_distance(s: str, t: str) -> int

Levenshtein edit distance between two strings.

Source code in src/linkingtk/blocking/trie.py

def edit_distance(s: str, t: str) -> int:
    """Levenshtein edit distance between two strings."""
    if len(s) > len(t):
        s, t = t, s
    previous = list(range(len(s) + 1))
    for j in range(1, len(t) + 1):
        current = [previous[0] + 1] + [0] * len(s)
        for i in range(1, len(s) + 1):
            cost = 0 if s[i - 1] == t[j - 1] else 1
            current[i] = min(previous[i] + 1, current[i - 1] + 1, previous[i - 1] + cost)
        previous = current
    return previous[len(s)]

embedding

Fuzzy blocking strategy based on TF-IDF vector similarity.

A non-neural, classical-ML alternative to LabelOverlap: entities are compared by cosine similarity of vectors from a pluggable scikit-learn-style text vectorizer (TF-IDF by default) rather than surface- string overlap, so paraphrased or reordered text (e.g. a description and a differently-worded gloss of the same sense) can still block together.

EmbeddingSimilarityBlocker

EmbeddingSimilarityBlocker(
    field: Field = "label",
    top_k: int = 10,
    threshold: float | None = None,
    vectorizer: Vectorizer | None = None,
)

Bases: BlockingStrategy

Blocks entities by cosine similarity of vectorized text.

Fits vectorizer on dataset2 (treating it as the reference corpus) and projects dataset1 into that same vector space, then for each dataset1 entity retrieves the top_k closest dataset2 entities by cosine similarity via sklearn.neighbors.NearestNeighbors.

Parameters:

Name	Type	Description	Default
field	Field	Which text to compare entities on: "label" (all labels, space-joined), "description", or "context". A callable taking an Entity and returning str may be passed instead, for fields not covered above.	'label'
top_k	int	Maximum number of candidates to keep per source entity.	10
threshold	float | None	Optional minimum cosine similarity a candidate must reach to be kept. If not set, candidates with a similarity of exactly zero are still dropped, since top_k nearest- neighbor search otherwise always returns top_k results regardless of relevance. This default assumes a non-negative vector representation (true for TfidfVectorizer, CountVectorizer, and other bag-of-words-style vectorizers, where a zero similarity exactly means "shares nothing"); for a dense embedding vectorizer, where small positive or negative similarities can be meaningful, pass an explicit threshold instead of relying on this default.	None
vectorizer	Vectorizer | None	Any object exposing scikit-learn's fit_transform/transform interface (e.g. TfidfVectorizer, CountVectorizer, or a custom embedding wrapper). Defaults to TfidfVectorizer(stop_words="english"). Document-frequency pruning, n-gram ranges, and other vectorization choices are configured on the vectorizer itself rather than on this class.	None

Source code in src/linkingtk/blocking/embedding.py

def __init__(
    self,
    field: Field = "label",
    top_k: int = 10,
    threshold: float | None = None,
    vectorizer: Vectorizer | None = None,
) -> None:
    if top_k < 1:
        raise ValueError("top_k must be at least 1")
    self.field = field
    self.top_k = top_k
    self.threshold = threshold
    self.vectorizer: Vectorizer = (
        vectorizer if vectorizer is not None else TfidfVectorizer(stop_words="english")
    )

Vectorizer

Bases: Protocol

Structural type for a fit_transform/transform text vectorizer (e.g. scikit-learn's).

negative_sampling

Hard-negative sampling for training classical ML linkers.

Mines the most confusable non-matching candidates per source entity from a BlockingStrategy, rather than sampling negatives uniformly at random from the full candidate pool — a blocking strategy already considered these plausible enough to surface, so they're more informative training signal for a classifier (fit) than a random draw. Shaped as a plain function over Entity/ BlockingStrategy (no dependency on TrainingArguments or anything Trainer-specific) so it's equally usable by Trainer once that's implemented.

sample_hard_negatives

sample_hard_negatives(
    dataset1: list[Entity],
    dataset2: list[Entity] | EntitySource,
    ground_truth: list[tuple[str, str]],
    blocking: BlockingStrategy,
    top_k: int = 5,
) -> list[tuple[Entity, Entity]]

Sample the top-K most confusable non-matching candidates per source entity.

For each distinct source entity in ground_truth, takes that entity's candidates from blocking.candidate_pairs(dataset1, dataset2) in the best-first order documented by candidate_pairs, excluding any candidate that is itself a true match for that source entity, up to top_k.

A source entity with fewer than top_k candidates (or none at all) simply contributes fewer negatives — this is normal partial coverage, not an error.

Parameters:

Name	Type	Description	Default
dataset1	list[Entity]	Source entities.	required
dataset2	list[Entity] | EntitySource	Target entities.	required
ground_truth	list[tuple[str, str]]	List of (source_id, target_id) true pairs.	required
blocking	BlockingStrategy	Strategy used to generate candidates to mine negatives from.	required
top_k	int	Maximum number of hard negatives to keep per source entity.	5

Returns:

Type	Description
list[tuple[Entity, Entity]]	A flat list of (entity1, entity2) negative pairs, suitable as
list[tuple[Entity, Entity]]	fit's
list[tuple[Entity, Entity]]	negatives argument. If you're about to pass the result straight
list[tuple[Entity, Entity]]	to fit(..., blocking=blocking), note blocking.candidate_pairs()
list[tuple[Entity, Entity]]	runs once here and again inside fit() — fine at this class's
list[tuple[Entity, Entity]]	intended toy/small-dataset scale, but worth knowing if blocking
list[tuple[Entity, Entity]]	is itself expensive.

Source code in src/linkingtk/blocking/negative_sampling.py

def sample_hard_negatives(
    dataset1: list[Entity],
    dataset2: list[Entity] | EntitySource,
    ground_truth: list[tuple[str, str]],
    blocking: BlockingStrategy,
    top_k: int = 5,
) -> list[tuple[Entity, Entity]]:
    """Sample the top-K most confusable non-matching candidates per source entity.

    For each distinct source entity in ``ground_truth``, takes that
    entity's candidates from ``blocking.candidate_pairs(dataset1,
    dataset2)`` in the best-first order documented by
    [candidate_pairs][linkingtk.blocking.base.BlockingStrategy.candidate_pairs],
    excluding any candidate that is itself a true match for that source
    entity, up to ``top_k``.

    A source entity with fewer than ``top_k`` candidates (or none at all)
    simply contributes fewer negatives — this is normal partial coverage,
    not an error.

    Args:
        dataset1: Source entities.
        dataset2: Target entities.
        ground_truth: List of ``(source_id, target_id)`` true pairs.
        blocking: Strategy used to generate candidates to mine negatives from.
        top_k: Maximum number of hard negatives to keep per source entity.

    Returns:
        A flat list of ``(entity1, entity2)`` negative pairs, suitable as
        [fit][linkingtk.algorithms.feature_classifier.FeatureClassifierLinker.fit]'s
        ``negatives`` argument. If you're about to pass the result straight
        to ``fit(..., blocking=blocking)``, note ``blocking.candidate_pairs()``
        runs once here and again inside ``fit()`` — fine at this class's
        intended toy/small-dataset scale, but worth knowing if ``blocking``
        is itself expensive.
    """
    if not blocking.ranked:
        logger.warning(
            "sample_hard_negatives called with a blocking strategy that isn't "
            "marked as returning best-first candidates (ranked=False); negatives "
            "may not actually be the hardest available, just whichever candidates "
            "`blocking` happened to return first."
        )

    entities1_by_id = {entity.id: entity for entity in dataset1}
    ground_truth_set = set(ground_truth)

    candidates_by_source: dict[str, list[Entity]] = defaultdict(list)
    for entity1, entity2 in blocking.candidate_pairs(dataset1, dataset2):
        candidates_by_source[entity1.id].append(entity2)

    source_ids = dict.fromkeys(source_id for source_id, _ in ground_truth)
    negatives: list[tuple[Entity, Entity]] = []
    for source_id in source_ids:
        source_entity = entities1_by_id.get(source_id)
        if source_entity is None:
            continue
        hard = [
            entity2
            for entity2 in candidates_by_source.get(source_id, [])
            if (source_id, entity2.id) not in ground_truth_set
        ][:top_k]
        negatives.extend((source_entity, entity2) for entity2 in hard)

    if not negatives:
        logger.warning(
            "sample_hard_negatives found zero negative candidates across %d "
            "ground-truth source entities; blocking may be too narrow (e.g. "
            "top_k/max_matches too small, or threshold too strict) to have "
            "anything but true matches to mine.",
            len(source_ids),
        )
    return negatives