solar_eval/evaluators/composite.py

from typing import Any

from solar_eval.evaluators.base import BaseEvaluator
from solar_eval.models.sample import EvalSample
from solar_eval.providers.base import BaseProvider


class CompositeEvaluator(BaseEvaluator):
    """Combines multiple evaluators with weights.

    Example: 50% LLM Judge + 50% Rule-based for title evaluation.
    """

    def __init__(self, components: list[dict[str, Any]]) -> None:
        """Components: [{evaluator: BaseEvaluator, weight: float}]"""
        self.components = components

    @property
    def required_fields(self) -> frozenset[str]:
        """구성된 하위 평가기 요구 필드의 합집합.

        `components` 는 인스턴스마다 다르게 구성되므로(생성자 인자) 고정된
        클래스 속성으로는 표현할 수 없다 -- `BaseEvaluator.required_fields`
        를 `@property` 로 오버라이드해 매번 하위 평가기에서 합쳐낸다.
        """
        fields: frozenset[str] = frozenset()
        for comp in self.components:
            fields |= comp["evaluator"].required_fields
        return fields

    async def evaluate(
        self,
        sample: EvalSample,
        provider: BaseProvider | None = None,
        judge_model: str = "gpt-4o",
    ) -> dict[str, Any]:
        component_results = []
        weighted_score = 0.0

        for comp in self.components:
            evaluator = comp["evaluator"]
            weight = comp["weight"]
            result = await evaluator.evaluate(
                sample=sample,
                provider=provider,
                judge_model=judge_model,
            )
            weighted_score += result["score"] * weight
            component_results.append(
                {
                    "weight": weight,
                    "score": result["score"],
                    "details": result.get("details", {}),
                }
            )

        return {
            "score": weighted_score,
            "components": component_results,
            "details": {"components": component_results},
        }

    def aggregate(self, results: list[dict[str, Any]]) -> dict[str, Any]:
        if not results:
            return {"overall_score": 0.0, "scores": {}}

        overall = sum(r["score"] for r in results) / len(results)
        return {"overall_score": overall, "num_samples": len(results)}