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)}