Evaluation¶

Evaluation contracts, registries, orchestration, and retrieval metrics.

class pathforge.core.evaluation.base.TaskEvaluationAdapterBase[source]¶

Bases: ABC

Base class for task-specific evaluation adapters.

task_name: str = ''¶

__init__(experiment: Experiment) → None[source]¶

Parameters:: experiment (Experiment)
Return type:: None

abstractmethod classmethod get_discovery_keys() → list[str][source]¶

Return the benchmark keys used to rediscover task runs.

Return type:: list[str]

abstractmethod discover_runs() → list[EvaluationRunContext][source]¶

Discover all evaluable runs for the configured task.

Return type:: list[EvaluationRunContext]

abstractmethod load_run_data(run_context: EvaluationRunContext) → object[source]¶

Load and normalize one discovered run into metric-ready data.

Parameters:: run_context (EvaluationRunContext)
Return type:: object

class pathforge.core.evaluation.types.MetricRequest[source]¶

Bases: object

Parsed, immutable evaluation-metric request.

Example:

request = MetricRequest(
    raw_name="hit_at_5",
    canonical_name="hit_at_k",
    params={"k": 5},
)

raw_name: str¶

canonical_name: str¶

params: dict[str, Any]¶

__init__(raw_name: str, canonical_name: str, params: dict[str, ~typing.Any]=<factory>) → None¶

Parameters:

raw_name (str)
canonical_name (str)
params (dict[str, Any])

Return type:

None

class pathforge.core.evaluation.types.EvaluationRunContext[source]¶

Bases: object

Immutable evaluation context for one discovered run.

task_name: str¶

run_dir: Path¶

combo_cfg: ComboConfig¶

manifest: dict[str, Any]¶

label_column: str¶

aggregation_level: str¶

to_dict() → dict[str, Any][source]¶

Return a JSON-serializable representation of the run context.

Return type:: dict[str, Any]

__init__(task_name: str, run_dir: Path, combo_cfg: ComboConfig, manifest: dict[str, Any], label_column: str, aggregation_level: str) → None¶

Parameters:

task_name (str)
run_dir (Path)
combo_cfg (ComboConfig)
manifest (dict[str, Any])
label_column (str)
aggregation_level (str)

Return type:

None

class pathforge.core.evaluation.types.EvaluationSummary[source]¶

Bases: object

JSON-serializable evaluation summary for one run.

run_context: EvaluationRunContext¶

metrics: dict[str, Any]¶

to_dict() → dict[str, Any][source]¶

Return a JSON-serializable summary payload.

Return type:: dict[str, Any]

__init__(run_context: EvaluationRunContext, metrics: dict[str, Any]) → None¶

Parameters:

run_context (EvaluationRunContext)
metrics (dict[str, Any])

Return type:

None

class pathforge.core.evaluation.registry.EvaluationMetricSpec[source]¶

Bases: object

Registered evaluation-metric specification.

canonical_name: str¶

supported_tasks: frozenset[str]¶

parser: Callable[[str], MetricRequest | None]¶

compute_fn: Callable[[...], dict[str, Any]]¶

__init__(canonical_name: str, supported_tasks: frozenset[str], parser: Callable[[str], MetricRequest | None], compute_fn: Callable[[...], dict[str, Any]]) → None¶

Parameters:

canonical_name (str)
supported_tasks (frozenset[str])
parser (Callable[[str], MetricRequest | None])
compute_fn (Callable[[...], dict[str, Any]])

Return type:

None

pathforge.core.evaluation.registry.evaluation_task_adapter(task_name: str) → Callable[[type['TaskEvaluationAdapterBase']], type['TaskEvaluationAdapterBase']][source]¶

Parameters:: task_name (str)
Return type:: Callable[[type[‘TaskEvaluationAdapterBase’]], type[‘TaskEvaluationAdapterBase’]]

pathforge.core.evaluation.registry.get_task_evaluation_adapter(name: str) → type['TaskEvaluationAdapterBase'][source]¶

Return one registered task evaluation adapter.

Parameters:: name (str)
Return type:: type[‘TaskEvaluationAdapterBase’]

pathforge.core.evaluation.registry.build_task_evaluation_adapter(name: str, experiment: Experiment) → TaskEvaluationAdapterBase[source]¶

Instantiate one registered task evaluation adapter.

Parameters:

name (str)
experiment (Experiment)

Return type:

TaskEvaluationAdapterBase

pathforge.core.evaluation.registry.list_task_evaluation_adapters() → list[str][source]¶

Return registered task evaluation adapters in sorted order.

Return type:: list[str]

pathforge.core.evaluation.registry.import_task_evaluation_adapter_modules(package_name: str = 'pathforge.core.evaluation') → None[source]¶

Import all task-evaluation adapter modules so registration side effects run.

Parameters:: package_name (str)
Return type:: None

pathforge.core.evaluation.registry.evaluation_metric(name: str, *, tasks: tuple[str, ...] | list[str], pattern: str | Pattern[str] | None = None, param_builder: Callable[[Match[str]], dict[str, Any]] | None = None) → Callable[[Callable[[...], dict[str, Any]]], Callable[[...], dict[str, Any]]][source]¶

Inputs: - name: canonical metric name or family name. - tasks: tasks supported by this metric. - pattern: optional regex used to parse parameterized metric requests. - param_builder: optional builder that converts a regex match into metric params.

Returns: - Decorator that registers the metric compute function.

Parameters:

name (str)
tasks (tuple[str, ...] | list[str])
pattern (str | Pattern[str] | None)
param_builder (Callable[[Match[str]], dict[str, Any]] | None)

Return type:

Callable[[Callable[[…], dict[str, Any]]], Callable[[…], dict[str, Any]]]

pathforge.core.evaluation.registry.import_evaluation_metric_modules(package_name: str = 'pathforge.core.evaluation') → None[source]¶

Import all evaluation-metric modules so registration side effects run.

Parameters:: package_name (str)
Return type:: None

pathforge.core.evaluation.registry.resolve_metric_request(*, task_name: str, raw_name: str) → tuple[EvaluationMetricSpec, MetricRequest][source]¶

Resolve one raw metric request to a registered metric specification.

Parameters:

task_name (str)
raw_name (str)

Return type:

tuple[EvaluationMetricSpec, MetricRequest]

pathforge.core.evaluation.registry.list_evaluation_metrics(task_name: str | None = None) → list[str][source]¶

Return canonical registered metric names, optionally filtered by task.

Parameters:: task_name (str | None)
Return type:: list[str]

class pathforge.core.evaluation.orchestrator.EvaluationOrchestrator[source]¶

Bases: object

Run config-driven evaluation for the configured task.

__init__(experiment: Experiment) → None[source]¶

Parameters:: experiment (Experiment)
Return type:: None

evaluate() → dict[str, Any][source]¶

Evaluate all discovered runs for the configured task.

Return type:: dict[str, Any]

Slide retrieval¶

class pathforge.core.evaluation.slide_retrieval.adapter.SlideRetrievalEvaluationAdapter[source]¶

Bases: TaskEvaluationAdapterBase

Evaluate saved slide-retrieval benchmark outputs.

task_name: str = 'slide_retrieval'¶

classmethod get_discovery_keys() → list[str][source]¶

Return the benchmark keys used to rediscover task runs.

Return type:: list[str]

discover_runs() → list[EvaluationRunContext][source]¶

Discover all evaluable runs for the configured task.

Return type:: list[EvaluationRunContext]

load_run_data(run_context: EvaluationRunContext) → SlideRetrievalEvaluationData[source]¶

Load and normalize one discovered run into metric-ready data.

Parameters:: run_context (EvaluationRunContext)
Return type:: SlideRetrievalEvaluationData

class pathforge.core.evaluation.slide_retrieval.data.SlideRetrievalEvaluationHit[source]¶

Bases: object

One normalized retrieval hit for evaluation.

sample_id: str¶

label: str¶

score: float¶

rank: int¶

__init__(sample_id: str, label: str, score: float, rank: int) → None¶

Parameters:

sample_id (str)
label (str)
score (float)
rank (int)

Return type:

None

class pathforge.core.evaluation.slide_retrieval.data.SlideRetrievalEvaluationQuery[source]¶

Bases: object

One normalized retrieval query with resolved labels.

query_id: str¶

query_label: str¶

hits: list[SlideRetrievalEvaluationHit]¶

__init__(query_id: str, query_label: str, hits: list[SlideRetrievalEvaluationHit] = <factory>) → None¶

Parameters:

query_id (str)
query_label (str)
hits (list[SlideRetrievalEvaluationHit])

Return type:

None

class pathforge.core.evaluation.slide_retrieval.data.SlideRetrievalEvaluationData[source]¶

Bases: object

Task-specific evaluation payload for slide retrieval.

queries: list[SlideRetrievalEvaluationQuery]¶

__init__(queries: list[SlideRetrievalEvaluationQuery]) → None¶

Parameters:: queries (list[SlideRetrievalEvaluationQuery])
Return type:: None

pathforge.core.evaluation.slide_retrieval.pool.normalize_text_id(value: object) → str[source]¶

Normalize annotation and retrieval identifiers for evaluation joins.

Inputs:: value: Raw identifier from a CSV, Excel file, manifest, or result row.

Returns:: A whitespace-trimmed identifier, or "" for missing values.
Parameters:: value (object)
Return type:: str

Example

normalize_text_id(" S1 ") == "S1".

pathforge.core.evaluation.slide_retrieval.pool.compute_dcg(relevance_values: list[int]) → float[source]¶

Compute DCG for one ranked relevance list.

Parameters:: relevance_values (list[int])
Return type:: float

pathforge.core.evaluation.slide_retrieval.pool.resolve_project_root_from_run_context(run_context: object) → Path[source]¶

Find the project root by walking up from the run directory.

Parameters:: run_context (object)
Return type:: Path

pathforge.core.evaluation.slide_retrieval.pool.resolve_id_column_for_aggregation(aggregation_level: str) → str[source]¶

Map aggregation level to its corresponding annotation identifier column.

Parameters:: aggregation_level (str)
Return type:: str

pathforge.core.evaluation.slide_retrieval.pool.build_exclusion_key_from_row(row: Series, *, aggregation_level: str, exclusion_level: str) → str | None[source]¶

Recreate the retrieval exclusion key from aggregated annotation metadata.

Parameters:

row (Series)
aggregation_level (str)
exclusion_level (str)

Return type:

str | None

pathforge.core.evaluation.slide_retrieval.pool.build_aggregated_reference_pool(*, run_context: object) → tuple[DataFrame, DataFrame][source]¶

Build in-memory aggregated query/reference metadata from annotations.csv.

Returns: - all_items_df: one row per aggregated sample across all datasets. - reference_pool_df: subset limited to datasets allowed as references.

Parameters:: run_context (object)
Return type:: tuple[DataFrame, DataFrame]

pathforge.core.evaluation.slide_retrieval.pool.count_relevant_reference_items_for_query(*, query: SlideRetrievalEvaluationQuery, all_items_df: DataFrame, reference_pool_df: DataFrame) → int[source]¶

Count findable same-label reference items for one query.

Parameters:

query (SlideRetrievalEvaluationQuery)
all_items_df (DataFrame)
reference_pool_df (DataFrame)

Return type:

int

pathforge.core.evaluation.slide_retrieval.pool.count_relevant_reference_items_by_query(*, evaluation_data: SlideRetrievalEvaluationData, run_context: object | None) → dict[str, int] | None[source]¶

Count findable references for each query in the persisted reference pool.

Inputs:: evaluation_data: Evaluated query rows and their resolved labels. run_context: Persisted run metadata, or None for direct metric calls.

Returns:

Per-query counts, or None when no run context is available.

Parameters:

evaluation_data (SlideRetrievalEvaluationData)
run_context (object | None)

Return type:

dict[str, int] | None

Example

count_relevant_reference_items_by_query(evaluation_data=data, run_context=context)["S1"] == 2.

pathforge.core.evaluation.slide_retrieval.voting.filter_query_self_hits(query: SlideRetrievalEvaluationQuery) → list[SlideRetrievalEvaluationHit][source]¶

Drop defensive self-hits so all metrics evaluate external neighbors only.

Parameters:: query (SlideRetrievalEvaluationQuery)
Return type:: list[SlideRetrievalEvaluationHit]

pathforge.core.evaluation.slide_retrieval.voting.get_top_k_hits(query: SlideRetrievalEvaluationQuery, *, k: int) → list[SlideRetrievalEvaluationHit][source]¶

Return the top-k ranked hits after defensive self-hit filtering.

Parameters:

query (SlideRetrievalEvaluationQuery)
k (int)

Return type:

list[SlideRetrievalEvaluationHit]

pathforge.core.evaluation.slide_retrieval.voting.predict_label_from_top_k(query: SlideRetrievalEvaluationQuery, *, k: int) → str | None[source]¶

Convert retrieval hits into one predicted label using strict-majority voting.

For k=1, return the top hit label after self-hit filtering. For k>1, vote over the available top min(k, num_hits) hits and return a label only when it has a strict majority within that available neighborhood. Ties return None.

Parameters:

query (SlideRetrievalEvaluationQuery)
k (int)

Return type:

str | None

pathforge.core.evaluation.slide_retrieval.metrics.hit.compute_hit_at_k(evaluation_data: SlideRetrievalEvaluationData, *, request: MetricRequest, run_context: object | None = None) → dict[str, object][source]¶

Compute slide-retrieval hit-rate at the requested k.

Parameters:

evaluation_data (SlideRetrievalEvaluationData)
request (MetricRequest)
run_context (object | None)

Return type:

dict[str, object]

pathforge.core.evaluation.slide_retrieval.metrics.precision.compute_precision_at_k(evaluation_data: SlideRetrievalEvaluationData, *, request: MetricRequest, run_context: object | None = None) → dict[str, object][source]¶

Compute retrieval precision at the requested k.

Parameters:

evaluation_data (SlideRetrievalEvaluationData)
request (MetricRequest)
run_context (object | None)

Return type:

dict[str, object]

pathforge.core.evaluation.slide_retrieval.metrics.f1.compute_macro_f1_at_k(evaluation_data: SlideRetrievalEvaluationData, *, request: MetricRequest, run_context: object | None = None) → dict[str, object][source]¶

Compute strict-majority multiclass macro-F1 at the requested k.

Parameters:

evaluation_data (SlideRetrievalEvaluationData)
request (MetricRequest)
run_context (object | None)

Return type:

dict[str, object]

pathforge.core.evaluation.slide_retrieval.metrics.map.compute_map_at_k(evaluation_data: SlideRetrievalEvaluationData, *, request: MetricRequest, run_context: object | None = None) → dict[str, object][source]¶

Compute mean average precision at the requested k.

Parameters:

evaluation_data (SlideRetrievalEvaluationData)
request (MetricRequest)
run_context (object | None)

Return type:

dict[str, object]

pathforge.core.evaluation.slide_retrieval.metrics.mmv.compute_mmv_at_k(evaluation_data: SlideRetrievalEvaluationData, *, request: MetricRequest, run_context: object | None = None) → dict[str, object][source]¶

Compute majority-vote accuracy at the requested k.

Parameters:

evaluation_data (SlideRetrievalEvaluationData)
request (MetricRequest)
run_context (object | None)

Return type:

dict[str, object]

pathforge.core.evaluation.slide_retrieval.metrics.ndcg.compute_ndcg_at_k(evaluation_data: SlideRetrievalEvaluationData, *, request: MetricRequest, run_context: object | None = None) → dict[str, object][source]¶

Compute binary NDCG at the requested k.

Parameters:

evaluation_data (SlideRetrievalEvaluationData)
request (MetricRequest)
run_context (object | None)

Return type:

dict[str, object]