Evaluation

evaluator

Standardized evaluation metrics for EA, EL, WSD and WSA.

Evaluator

Computes standard linking metrics from predictions and ground truth.

evaluate staticmethod

evaluate(
    predictions: list[tuple[str, str]],
    ground_truth: list[tuple[str, str]],
) -> EvaluationReport

Compute micro precision@1, recall and F1 for unranked predictions.

Parameters:

Name	Type	Description	Default
predictions	list[tuple[str, str]]	List of (source_id, predicted_target_id) pairs.	required
ground_truth	list[tuple[str, str]]	List of (source_id, target_id) pairs. A source id may appear more than once, for tasks with more than one acceptable answer per source (e.g. some WSD gold-standard instances list several correct sense keys) -- a prediction matching any of them counts as correct, and the recall denominator is the number of distinct source ids, not the number of ground_truth rows.	required

Returns:

Type	Description
EvaluationReport	A report containing precision@1, recall and f1.

Source code in src/linkingtk/eval/evaluator.py

@staticmethod
def evaluate(
    predictions: list[tuple[str, str]],
    ground_truth: list[tuple[str, str]],
) -> EvaluationReport:
    """Compute micro precision@1, recall and F1 for unranked predictions.

    Args:
        predictions: List of ``(source_id, predicted_target_id)`` pairs.
        ground_truth: List of ``(source_id, target_id)`` pairs. A
            source id may appear more than once, for tasks with
            more than one acceptable answer per source (e.g. some
            WSD gold-standard instances list several correct sense
            keys) -- a prediction matching *any* of them counts as
            correct, and the recall denominator is the number of
            distinct source ids, not the number of ``ground_truth``
            rows.

    Returns:
        A report containing ``precision@1``, ``recall`` and ``f1``.
    """
    truth_by_source: dict[str, set[str]] = {}
    for source_id, target_id in ground_truth:
        truth_by_source.setdefault(source_id, set()).add(target_id)
    correct = sum(
        1
        for source_id, target_id in predictions
        if target_id in truth_by_source.get(source_id, set())
    )
    precision = correct / len(predictions) if predictions else 0.0
    recall = correct / len(truth_by_source) if truth_by_source else 0.0
    f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0.0
    return EvaluationReport(metrics={"precision@1": precision, "recall": recall, "f1": f1})

evaluate_blocking staticmethod

evaluate_blocking(
    candidate_pairs: list[tuple[str, str]],
    ground_truth: list[tuple[str, str]],
    dataset1_size: int,
    dataset2_size: int | None = None,
) -> EvaluationReport

Compute Pair Completeness and Reduction Ratio for a blocking pass.

These metrics assess a BlockingStrategy in isolation, before any linker scores the candidate pairs it produces.

Parameters:

Name	Type	Description	Default
candidate_pairs	list[tuple[str, str]]	(source_id, target_id) candidate pairs produced by a blocking strategy.	required
ground_truth	list[tuple[str, str]]	List of (source_id, target_id) true matching pairs.	required
dataset1_size	int	Number of entities in the first dataset.	required
dataset2_size	int | None	Number of entities in the second dataset, or None if unknown -- e.g. blocking targeted an EntitySource with no fixed size. reduction_ratio is then not computable.	None

Returns:

Type	Description
EvaluationReport	A report containing pair_completeness (the fraction of true
EvaluationReport	matches present among the candidate pairs, i.e. blocking recall)
EvaluationReport	and reduction_ratio (the fraction of the full ``dataset1_size
EvaluationReport	dataset2_size`` cross-product eliminated by blocking, or
EvaluationReport	None if dataset2_size wasn't given).

Source code in src/linkingtk/eval/evaluator.py

@staticmethod
def evaluate_blocking(
    candidate_pairs: list[tuple[str, str]],
    ground_truth: list[tuple[str, str]],
    dataset1_size: int,
    dataset2_size: int | None = None,
) -> EvaluationReport:
    """Compute Pair Completeness and Reduction Ratio for a blocking pass.

    These metrics assess a [BlockingStrategy][linkingtk.blocking.base.BlockingStrategy]
    in isolation, before any linker scores the candidate pairs it produces.

    Args:
        candidate_pairs: ``(source_id, target_id)`` candidate pairs
            produced by a blocking strategy.
        ground_truth: List of ``(source_id, target_id)`` true matching
            pairs.
        dataset1_size: Number of entities in the first dataset.
        dataset2_size: Number of entities in the second dataset, or
            ``None`` if unknown -- e.g. blocking targeted an
            [EntitySource][linkingtk.core.source.EntitySource] with no
            fixed size. ``reduction_ratio`` is then not computable.

    Returns:
        A report containing ``pair_completeness`` (the fraction of true
        matches present among the candidate pairs, i.e. blocking recall)
        and ``reduction_ratio`` (the fraction of the full ``dataset1_size
        * dataset2_size`` cross-product eliminated by blocking, or
        ``None`` if ``dataset2_size`` wasn't given).
    """
    candidate_set = set(candidate_pairs)
    true_positives = sum(1 for pair in ground_truth if pair in candidate_set)
    pair_completeness = true_positives / len(ground_truth) if ground_truth else 0.0

    reduction_ratio: float | None = None
    if dataset2_size is not None:
        total_possible = dataset1_size * dataset2_size
        reduction_ratio = 1 - len(candidate_pairs) / total_possible if total_possible else 0.0
    return EvaluationReport(
        metrics={"pair_completeness": pair_completeness, "reduction_ratio": reduction_ratio}
    )

evaluate_ranked staticmethod

evaluate_ranked(
    ranked_predictions: list[tuple[str, list[str]]],
    ground_truth: list[tuple[str, str]],
    top_k: list[int],
) -> EvaluationReport

Compute Hits@k and MRR for ranked predictions.

Iterates over ground_truth (not ranked_predictions), so the denominator is always exactly len(ground_truth) -- the number of queries actually being scored. This matters for callers (e.g. every EA-linker benchmark script under examples/) that call linker.link() over a superset of the evaluated split (train + val + test entities, since which entities need embeddings is determined before the split is known): a ground-truth source with no matching entry in ranked_predictions at all (e.g. blocking found no candidates for it) counts as a miss, exactly like a source whose ranked list doesn't contain the correct target -- neither case is silently excluded, and no non-ground-truth prediction can dilute the denominator.

Parameters:

Name	Type	Description	Default
ranked_predictions	list[tuple[str, list[str]]]	List of (source_id, ranked_target_ids) pairs, best candidate first. May contain entries for source_ids absent from ground_truth -- those are ignored rather than counted.	required
ground_truth	list[tuple[str, str]]	List of (source_id, target_id) pairs.	required
top_k	list[int]	The cut-offs to compute Hits@k for.	required

Returns:

Type	Description
EvaluationReport	A report containing Hits@k for each k in top_k and
EvaluationReport	MRR.

Source code in src/linkingtk/eval/evaluator.py

@staticmethod
def evaluate_ranked(
    ranked_predictions: list[tuple[str, list[str]]],
    ground_truth: list[tuple[str, str]],
    top_k: list[int],
) -> EvaluationReport:
    """Compute Hits@k and MRR for ranked predictions.

    Iterates over ``ground_truth`` (not ``ranked_predictions``), so the
    denominator is always exactly ``len(ground_truth)`` -- the number of
    queries actually being scored. This matters for callers (e.g. every
    EA-linker benchmark script under ``examples/``) that call
    ``linker.link()`` over a *superset* of the evaluated split (train +
    val + test entities, since which entities need embeddings is
    determined before the split is known): a ground-truth source with no
    matching entry in ``ranked_predictions`` at all (e.g. blocking found
    no candidates for it) counts as a miss, exactly like a source whose
    ranked list doesn't contain the correct target -- neither case is
    silently excluded, and no non-ground-truth prediction can dilute the
    denominator.

    Args:
        ranked_predictions: List of ``(source_id, ranked_target_ids)``
            pairs, best candidate first. May contain entries for
            ``source_id``s absent from ``ground_truth`` -- those are
            ignored rather than counted.
        ground_truth: List of ``(source_id, target_id)`` pairs.
        top_k: The cut-offs to compute Hits@k for.

    Returns:
        A report containing ``Hits@k`` for each ``k`` in ``top_k`` and
        ``MRR``.
    """
    ranked_ids_by_source = dict(ranked_predictions)
    hits = {k: 0 for k in top_k}
    reciprocal_ranks = []

    for source_id, target_id in ground_truth:
        ranked_ids = ranked_ids_by_source.get(source_id, [])
        if target_id in ranked_ids:
            rank = ranked_ids.index(target_id) + 1
            reciprocal_ranks.append(1.0 / rank)
            for k in top_k:
                if rank <= k:
                    hits[k] += 1
        else:
            reciprocal_ranks.append(0.0)

    total = len(ground_truth) or 1
    metrics: dict[str, float | None] = {f"Hits@{k}": hits[k] / total for k in top_k}
    metrics["MRR"] = sum(reciprocal_ranks) / total
    return EvaluationReport(metrics=metrics)

ranking

Exhaustive, non-blocking ranked evaluation for EA linkers.

rank_exhaustive

rank_exhaustive(
    linker: _ScoringLinker,
    source_entities: list[Entity],
    target_entities: list[Entity],
    metric: _Metric = "cosine",
    csls_k: int = 0,
) -> list[tuple[str, list[str]]]

Rank every source_entities entity against every target_entities entity.

No blocking/candidate-restriction step at all -- a single dense similarity matrix, matching OpenEA's own greedy_alignment methodology (modules/finding/alignment.py, itself a thin wrapper around modules/finding/similarity.py's sim/csls_sim, ported here). Intended for benchmark scripts that need numbers directly comparable to a published target: BlockingStrategy-restricted evaluation (the normal, production-scale path via BaseLinker.link) upper-bounds the reported metric by blocking's own recall, not purely by the linker's embedding/ranking quality -- see issue #37. Not intended for production-scale linking: this always computes a full len(source_entities) x len(target_entities) similarity matrix.

Parameters:

Name	Type	Description	Default
linker	_ScoringLinker	A fitted EA linker exposing source_embedding/ target_embedding -- every linker in linkingtk.algorithms.ea does.	required
source_entities	list[Entity]	Entities to rank from.	required
target_entities	list[Entity]	Entities to rank against.	required
metric	_Metric	Similarity metric -- "cosine" (default, this function's long-standing behavior), "manhattan" (1 - cityblock_distance, unnormalized), or "inner" (raw dot product, unnormalized). Different EA methods' own OpenEA configs use different metrics for their published numbers -- e.g. GCN-Align's and RDGCN's own eval_metric: "manhattan", eval_norm: false (confirmed by reading run/args/{gcnalign,rdgcn}_args_15K.json directly) -- so a benchmark script targeting a specific published number should pass the metric that method's own config specifies, not assume cosine. Confirmed empirically on RDGCN (#43): switching from cosine to manhattan+csls_k=10 closed a published-number gap from ~88% down to ~97% relative Hits@1.	'cosine'
csls_k	int	If > 0, applies Cross-domain Similarity Local Scaling (Lample et al. 2018) with this many nearest neighbors -- ports OpenEA's own csls_sim, which several EA methods' configs enable (typically 10) for their published numbers. 0 (default) disables it, this function's long-standing behavior.	0

Returns:

Type	Description
list[tuple[str, list[str]]]	(source_id, ranked_target_ids) pairs, best match first, one
list[tuple[str, list[str]]]	entry per source_entities. Each list contains every
list[tuple[str, list[str]]]	target_entities id -- never truncated, since
list[tuple[str, list[str]]]	Evaluator.evaluate_ranked
list[tuple[str, list[str]]]	needs the true target's real rank for MRR even when it falls
list[tuple[str, list[str]]]	outside any Hits@k cutoff; truncating here would silently
list[tuple[str, list[str]]]	reintroduce a version of the denominator/miss bug that function's
list[tuple[str, list[str]]]	own docstring documents fixing. top_k cutoffs are
list[tuple[str, list[str]]]	evaluate_ranked's job, not this function's.

Source code in src/linkingtk/eval/ranking.py

def rank_exhaustive(
    linker: _ScoringLinker,
    source_entities: list[Entity],
    target_entities: list[Entity],
    metric: _Metric = "cosine",
    csls_k: int = 0,
) -> list[tuple[str, list[str]]]:
    """Rank every ``source_entities`` entity against every ``target_entities`` entity.

    No blocking/candidate-restriction step at all -- a single dense
    similarity matrix, matching OpenEA's own ``greedy_alignment``
    methodology (``modules/finding/alignment.py``, itself a thin wrapper
    around ``modules/finding/similarity.py``'s ``sim``/``csls_sim``, ported
    here). Intended for benchmark scripts that need numbers directly
    comparable to a published target:
    [BlockingStrategy][linkingtk.blocking.base.BlockingStrategy]-restricted
    evaluation (the normal, production-scale path via
    [BaseLinker.link][linkingtk.algorithms.base.BaseLinker.link]) upper-bounds
    the reported metric by blocking's own recall, not purely by the
    linker's embedding/ranking quality -- see issue #37. Not intended for
    production-scale linking: this always computes a full
    ``len(source_entities) x len(target_entities)`` similarity matrix.

    Args:
        linker: A fitted EA linker exposing ``source_embedding``/
            ``target_embedding`` -- every linker in ``linkingtk.algorithms.ea``
            does.
        source_entities: Entities to rank from.
        target_entities: Entities to rank against.
        metric: Similarity metric -- ``"cosine"`` (default, this
            function's long-standing behavior), ``"manhattan"``
            (``1 - cityblock_distance``, unnormalized), or ``"inner"``
            (raw dot product, unnormalized). **Different EA methods'
            own OpenEA configs use different metrics for their published
            numbers** -- e.g. GCN-Align's and RDGCN's own
            ``eval_metric: "manhattan"``, ``eval_norm: false`` (confirmed
            by reading ``run/args/{gcnalign,rdgcn}_args_15K.json``
            directly) -- so a benchmark script targeting a specific
            published number should pass the metric *that method's own
            config* specifies, not assume cosine. Confirmed empirically
            on RDGCN (#43): switching from cosine to manhattan+csls_k=10
            closed a published-number gap from ~88% down to ~97% relative
            Hits@1.
        csls_k: If ``> 0``, applies Cross-domain Similarity Local Scaling
            (Lample et al. 2018) with this many nearest neighbors --
            ports OpenEA's own ``csls_sim``, which several EA methods'
            configs enable (typically ``10``) for their published numbers.
            ``0`` (default) disables it, this function's long-standing
            behavior.

    Returns:
        ``(source_id, ranked_target_ids)`` pairs, best match first, one
        entry per ``source_entities``. Each list contains **every**
        ``target_entities`` id -- never truncated, since
        [Evaluator.evaluate_ranked][linkingtk.eval.evaluator.Evaluator.evaluate_ranked]
        needs the true target's real rank for MRR even when it falls
        outside any Hits@k cutoff; truncating here would silently
        reintroduce a version of the denominator/miss bug that function's
        own docstring documents fixing. ``top_k`` cutoffs are
        ``evaluate_ranked``'s job, not this function's.
    """
    if not source_entities or not target_entities:
        return [(entity.id, []) for entity in source_entities]

    source_matrix = np.stack([linker.source_embedding(entity.id) for entity in source_entities])
    target_matrix = np.stack([linker.target_embedding(entity.id) for entity in target_entities])
    similarities = _similarity_matrix(source_matrix, target_matrix, metric)
    if csls_k > 0:
        similarities = _csls(similarities, csls_k)
    target_ids = [entity.id for entity in target_entities]

    order = np.argsort(-similarities, axis=1)
    return [
        (entity.id, [target_ids[i] for i in order[row]])
        for row, entity in enumerate(source_entities)
    ]

report

Result container returned by linkingtk.eval.evaluator.Evaluator.

EvaluationReport dataclass

EvaluationReport(metrics: dict[str, float | None])

Holds computed metrics for a single evaluation run.

Attributes:

Name	Type	Description
metrics	dict[str, float | None]	Mapping of metric name (e.g. "precision@1", "f1", "Hits@1", "MRR") to its computed value. A metric is None when it isn't computable for the given inputs (e.g. "reduction_ratio" when evaluate_blocking is called without a dataset2_size).

harness

A small comparison harness for running many linkers/datasets side by side.

Different tasks (EA/EL/WSD/WSA) and tiers (string-similarity, classical ML, KGE, deep learning, LLM-oriented) need very different fit/link/evaluate wiring -- ranked vs. unranked evaluation, exhaustive vs. blocking-restricted ranking, .fit() or not. Rather than have the harness guess at that, each BenchmarkRun carries a zero-argument callable that already does exactly what a normal examples/*_benchmark.py script does, and returns an EvaluationReport. run_benchmarks only adds timing and, for runs that make real LLM calls via a CountingLlmClient, a call count -- see examples/comparative_benchmark.py for real usage.

BenchmarkResult dataclass

BenchmarkResult(
    run: BenchmarkRun,
    metrics: dict[str, float | None],
    seconds: float,
    llm_calls: int | None,
)

The outcome of executing a single BenchmarkRun.

BenchmarkRun dataclass

BenchmarkRun(
    task: str,
    tier: str,
    linker: str,
    dataset: str,
    run: Callable[[], EvaluationReport],
    client: CountingLlmClient | None = None,
)

One (task, tier, linker, dataset) comparison cell.

Attributes:

Name	Type	Description
task	str	Display label, e.g. "EA", "EL", "WSD". Results are grouped by this in format_table.
tier	str	Display label, e.g. "MVP", "Classical ML", "KGE", "Deep Learning", "LLM-Oriented".
linker	str	Display label, e.g. "ReFinEDLinker".
dataset	str	Display label, e.g. "AIDA-CoNLL".
run	Callable[[], EvaluationReport]	Zero-argument callable that fits/links/evaluates and returns an EvaluationReport. Owns all task-specific wiring (blocking, ranked vs. unranked evaluation, training) -- the harness only times it.
client	CountingLlmClient | None	A CountingLlmClient used inside run, if any -- set this so the resulting BenchmarkResult reports an LLM-call count. None for runs that make no LLM calls.

CountingLlmClient

CountingLlmClient(client: LlmClient)

Bases: LlmClient

Wraps a real LlmClient, counting its calls.

Purely a benchmarking aid -- delegates every call unchanged and doesn't modify linkingtk.llm.client at all. Attach the instance to a BenchmarkRun's client field so run_benchmarks can report an LLM-call count alongside wall-clock time.

Parameters:

Name	Type	Description	Default
client	LlmClient	The real client to wrap.	required

Source code in src/linkingtk/eval/harness.py

def __init__(self, client: LlmClient) -> None:
    self._client = client
    self.call_count = 0

format_table

format_table(results: list[BenchmarkResult]) -> str

Renders results as a plain-text table, grouped by run.task.

Each task gets its own section with its own metric columns (the union of metric keys across that task's results, in first-seen order), since different tasks report different metrics (e.g. EA's Hits@k/MRR vs. EL/WSD's precision@1/recall/f1).

Parameters:

Name	Type	Description	Default
results	list[BenchmarkResult]	Results to render, in the order they should appear within each task's section.	required

Returns:

Type	Description
str	The rendered table.

Source code in src/linkingtk/eval/harness.py

def format_table(results: list[BenchmarkResult]) -> str:
    """Renders `results` as a plain-text table, grouped by `run.task`.

    Each task gets its own section with its own metric columns (the union
    of metric keys across that task's results, in first-seen order), since
    different tasks report different metrics (e.g. EA's `Hits@k`/`MRR` vs.
    EL/WSD's `precision@1`/`recall`/`f1`).

    Args:
        results: Results to render, in the order they should appear within
            each task's section.

    Returns:
        The rendered table.
    """
    tasks: dict[str, list[BenchmarkResult]] = {}
    for result in results:
        tasks.setdefault(result.run.task, []).append(result)

    sections = []
    for task, task_results in tasks.items():
        metric_keys: list[str] = []
        for result in task_results:
            for key in result.metrics:
                if key not in metric_keys:
                    metric_keys.append(key)

        headers = ["Tier", "Linker", "Dataset", *metric_keys, "Seconds", "LLM calls"]
        rows = [
            [
                result.run.tier,
                result.run.linker,
                result.run.dataset,
                *[_format_metric(result.metrics.get(key)) for key in metric_keys],
                f"{result.seconds:.1f}",
                "-" if result.llm_calls is None else str(result.llm_calls),
            ]
            for result in task_results
        ]
        widths = [
            max(len(headers[i]), *(len(row[i]) for row in rows)) if rows else len(headers[i])
            for i in range(len(headers))
        ]
        lines = [
            f"=== {task} ===",
            "  ".join(header.ljust(width) for header, width in zip(headers, widths, strict=True)),
        ]
        lines.extend(
            "  ".join(cell.ljust(width) for cell, width in zip(row, widths, strict=True))
            for row in rows
        )
        sections.append("\n".join(lines))
    return "\n\n".join(sections)

run_benchmarks

run_benchmarks(
    runs: list[BenchmarkRun],
) -> list[BenchmarkResult]

Executes each of runs in order, timing it and collecting its metrics.

Parameters:

Name	Type	Description	Default
runs	list[BenchmarkRun]	The comparison cells to execute, in order.	required

Returns:

Type	Description
list[BenchmarkResult]	One BenchmarkResult per run.

Source code in src/linkingtk/eval/harness.py

def run_benchmarks(runs: list[BenchmarkRun]) -> list[BenchmarkResult]:
    """Executes each of `runs` in order, timing it and collecting its metrics.

    Args:
        runs: The comparison cells to execute, in order.

    Returns:
        One [BenchmarkResult][linkingtk.eval.harness.BenchmarkResult] per run.
    """
    results = []
    for benchmark_run in runs:
        start = time.monotonic()
        report = benchmark_run.run()
        elapsed = time.monotonic() - start
        llm_calls = benchmark_run.client.call_count if benchmark_run.client is not None else None
        results.append(
            BenchmarkResult(
                run=benchmark_run, metrics=report.metrics, seconds=elapsed, llm_calls=llm_calls
            )
        )
    return results