pydantic-evals -- Custom Evaluation Aggregators

[baek54321]

Question

Hi all,

I would like to add custom aggregations for an evaluation report -- is this already supported? If yes, how can I do it? My use case is basically to compute various statistical KPIs based on the expected and actual outputs from multiple test cases.

Kind of like custom evaluators but for aggregated metrics.

Thanks!

Additional Context

No response

[DouweM]

@bkomboz The building blocks are there, but not yet exposed as an official feature. If you're up for trying it out and submitting a PR, I think it'd require these changes:

• Dataset will need a list of aggregators: functions that take list[ReportCase] and return an instance of ReportCaseAggregate
• Dataset.evaluate passes these on to EvaluationReport, which stores them like Dataset did
• EvaluationReport print and console_table get a new include_aggregates argument that's passed on to EvaluationRenderer
• EvaluationRenderer build_table and build_diff_table run the report.aggregators functions against report.cases and add the results to the table just like they currently do for ReportCaseAggregate.average

[DemirTonchev]

@DouweM looking at the required changes I think aggregators should be mapping between the evaluator name we want to apply this aggregation function to. If we have this evaluator

@dataclass
class ClassificationEvaluator(Evaluator):
    async def evaluate(self, ctx: EvaluatorContext[dict, TaskOutput, dict]) -> float:
        if ctx.expected_output is None:
            raise ValueError(f"Expected output cannot be None for {self.__qualname__}")
        return 1.0 if ctx.output.classification == ctx.expected_output.classification else 0.0

The key issue is that aggregation functions might be highly specific to the metrics they're computing. You can't apply a generic aggregation function (ei recall/precision) to arbitrary evaluators.

from sklearn.metrics import precision_score, recall_score

def precision(cases: list[ReportCase], evaluator_name: str) -> ReportCaseAggregate:
    # Precision MUST be computed from raw predictions, not from evaluator scores
    y_true = [{"negative": 0, "positive": 1}[case.expected_output.classification] for case in cases]
    y_pred = [{"negative": 0, "positive": 1}[case.output.classification] for case in cases]
    score = precision_score(y_true=y_true, y_pred=y_pred)
    
    return ReportCaseAggregate(name="precision", scores=score,
        metrics={},
        scores={evaluator_name: score},
        labels={},
        assertions=None,
        task_duration=0.0,
        total_duration=0.0,
    )

The precision aggregator cannot be applied to arbitrary evaluator scores like edit distance (doesnt make sense). It needs the actual classification outputs to compute true positives/false positives correctly.
Proposed solution:

aggregators = {
    "ClassificationEvaluator": [precision, ...],  # Only applies to ClassificationEvaluator,
    # ....
}

This mapping ensures each aggregation function only runs on evaluators it's designed for. Does that make sense?

[dmontagu]

I wish we could do something more type-safe, but I guess that's probably impractical.

I'm less convinced it makes sense for the aggregation configuration to include a reference to the evaluation name; the fact that the names can be dynamic means that agreement can't be checked even at runtime at the outset of the evaluation run. And given that, you can always just write a function like your precision example above and bind the evaluator name attribute. That said, I'm not necessarily opposed to this approach.

I think what would give me more confidence in the design would be if we implemented a version of custom aggregators that was more serialization-friendly, like our custom evaluators. In particular, I'm thinking we might be able to copy the approach for Evaluator but with a different signature for evaluation, so custom aggregators are dataclasses and serialization/deserialization "just works" but you can make use of logic that reflects whatever attributes you care about.

E.g., trying to adapt the logic you've shared, it might look something like:

@dataclass(repr=False)
class Precision(Aggregator[object, TaskOutput, object]):
    evaluator_name: str

    async def aggregate(
        self,
        cases: list[ReportCase[object, TaskOutput, object]],
    ) -> ReportCaseAggregate:
        y_true = [{"negative": 0, "positive": 1}[case.expected_output.classification] for case in cases]
        y_pred = [{"negative": 0, "positive": 1}[case.output.classification] for case in cases]
        score = precision_score(y_true=y_true, y_pred=y_pred)
        
        return ReportCaseAggregate(name="precision", scores=score,
            metrics={},
            scores={evaluator_name: score},
            labels={},
            assertions=None,
            task_duration=0.0,
            total_duration=0.0,
        )        

where of course we'd need to provide a base Aggregator class like we have for Evaluator. Though I think we'd probably want to rework ReportCaseAggregate a bit for this, it doesn't make sense for these duration fields to be required if we are having custom code produce outputs like this. I imagine the return value will actually be more similar in shape to the return values of Evaluators, I just wanted to adapt your code more directly for the sake of conveying the idea.