Skip to content

Evaluation and Objectives

Evaluators

evaluators

Built-in evaluator helpers and optional third-party integrations.

MetricScorer module-attribute 

MetricScorer = Callable[[Any, Any], float | None | Awaitable[float | None]]

RagasSampleMapper module-attribute 

RagasSampleMapper = Callable[[Any, Any], Mapping[str, Any]]

MetricEvaluator 

MetricEvaluator(name: str, scorer: MetricScorer, *, status: ValueStatus = EXACT)

Adapt one synchronous or asynchronous numeric scorer to an evaluator.

Source code in src/tunarag/evaluators.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def __init__(
    self,
    name: str,
    scorer: MetricScorer,
    *,
    status: ValueStatus = ValueStatus.EXACT,
) -> None:
    if not name.strip():
        raise ValueError("metric name must not be empty")
    if name == "tunarag.objective":
        raise ValueError("metric name is reserved: tunarag.objective")
    if not callable(scorer):
        raise TypeError("metric scorer must be callable")
    self._name = name
    self._scorer = scorer
    self._status = status

evaluate async 

evaluate(output: Any, example: Any) -> Sequence[MetricValue]

Score one adapter output and evaluation example.

Source code in src/tunarag/evaluators.py 
40
41
42
43
44
45
46
47
48
async def evaluate(self, output: Any, example: Any) -> Sequence[MetricValue]:
    """Score one adapter output and evaluation example."""

    result = self._scorer(output, example)
    value = await result if inspect.isawaitable(result) else result
    normalized = _numeric_score(value, metric=self._name)
    if normalized is None:
        return (MetricValue(self._name, None, ValueStatus.UNAVAILABLE, 0.0),)
    return (MetricValue(self._name, normalized, self._status),)

fingerprint 

fingerprint() -> Any

Return stable configuration used by study and cache identity.

Source code in src/tunarag/evaluators.py 
50
51
52
53
54
55
56
57
def fingerprint(self) -> Any:
    """Return stable configuration used by study and cache identity."""

    return {
        "name": self._name,
        "status": self._status.value,
        "scorer": _callable_identity(self._scorer),
    }

RagasEvaluator 

RagasEvaluator(metrics: Sequence[Any], sample_mapper: RagasSampleMapper, *, metric_prefix: str = 'ragas.', llm: Any = None, embeddings: Any = None, raise_exceptions: bool = False)

Evaluate one sample with optional RAGAS metrics without hard coupling.

Source code in src/tunarag/evaluators.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def __init__(
    self,
    metrics: Sequence[Any],
    sample_mapper: RagasSampleMapper,
    *,
    metric_prefix: str = "ragas.",
    llm: Any = None,
    embeddings: Any = None,
    raise_exceptions: bool = False,
) -> None:
    if not metrics:
        raise ValueError("RAGAS evaluator requires at least one metric")
    if not callable(sample_mapper):
        raise TypeError("RAGAS sample mapper must be callable")
    if metric_prefix == "tunarag.objective" or metric_prefix.startswith("tunarag.objective."):
        raise ValueError("RAGAS metric prefix uses a reserved name")
    self._metrics = tuple(metrics)
    self._sample_mapper = sample_mapper
    self._metric_prefix = metric_prefix
    self._llm = llm
    self._embeddings = embeddings
    self._raise_exceptions = raise_exceptions

evaluate async 

evaluate(output: Any, example: Any) -> Sequence[MetricValue]

Map and evaluate exactly one sample through RAGAS's async API.

Source code in src/tunarag/evaluators.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
async def evaluate(self, output: Any, example: Any) -> Sequence[MetricValue]:
    """Map and evaluate exactly one sample through RAGAS's async API."""

    try:
        sample_data = self._sample_mapper(output, example)
    except Exception as error:
        raise EvaluationError(
            "RAGAS sample mapping failed",
            stage="evaluation",
            details={"cause": type(error).__name__},
        ) from error
    if not isinstance(sample_data, Mapping):
        raise EvaluationError("RAGAS sample mapper must return a mapping", stage="evaluation")
    ragas, dataset_schema = _load_ragas()
    try:
        sample = dataset_schema.SingleTurnSample(**dict(sample_data))
        dataset = ragas.EvaluationDataset(samples=[sample])
        result = await ragas.aevaluate(
            dataset=dataset,
            metrics=list(self._metrics),
            llm=self._llm,
            embeddings=self._embeddings,
            raise_exceptions=self._raise_exceptions,
            show_progress=False,
        )
        scores = result.scores
    except Exception as error:
        raise EvaluationError(
            "RAGAS evaluation failed",
            stage="evaluation",
            details={"cause": type(error).__name__},
        ) from error
    if not isinstance(scores, Sequence) or len(scores) != 1:
        raise EvaluationError("RAGAS returned an invalid per-sample result", stage="evaluation")
    row = scores[0]
    if not isinstance(row, Mapping) or not row:
        raise EvaluationError("RAGAS returned no metric scores", stage="evaluation")
    observations: list[MetricValue] = []
    names: set[str] = set()
    for raw_name, raw_value in row.items():
        name = f"{self._metric_prefix}{raw_name}"
        if not name.strip() or name == "tunarag.objective":
            raise EvaluationError("RAGAS returned an invalid metric name", stage="evaluation")
        if name in names:
            raise EvaluationError("RAGAS returned duplicate metric names", stage="evaluation")
        names.add(name)
        value = _numeric_score(raw_value, metric=name)
        if value is None:
            observations.append(MetricValue(name, None, ValueStatus.UNAVAILABLE, 0.0))
        else:
            observations.append(MetricValue(name, value))
    return tuple(observations)

fingerprint 

fingerprint() -> Any

Return secret-safe semantic integration configuration.

Source code in src/tunarag/evaluators.py 
139
140
141
142
143
144
145
146
147
148
149
def fingerprint(self) -> Any:
    """Return secret-safe semantic integration configuration."""

    return {
        "metrics": [_component_identity(metric) for metric in self._metrics],
        "sample_mapper": _callable_identity(self._sample_mapper),
        "metric_prefix": self._metric_prefix,
        "llm": _component_identity(self._llm),
        "embeddings": _component_identity(self._embeddings),
        "raise_exceptions": self._raise_exceptions,
    }

Objectives

objective

Objective aggregation for quality, cost, and latency metrics.

ObjectiveDirection

Bases: str, Enum

Whether larger or smaller metric values are beneficial.

MAXIMIZE class-attribute instance-attribute 

MAXIMIZE = 'maximize'

MINIMIZE class-attribute instance-attribute 

MINIMIZE = 'minimize'

ObjectiveTerm dataclass 

ObjectiveTerm(metric: str, weight: float = 1.0, direction: ObjectiveDirection = MAXIMIZE, minimum: float | None = None, maximum: float | None = None)

One weighted metric contribution to an objective score.

metric instance-attribute 

metric: str

weight class-attribute instance-attribute 

weight: float = 1.0

direction class-attribute instance-attribute 

direction: ObjectiveDirection = ObjectiveDirection.MAXIMIZE

minimum class-attribute instance-attribute 

minimum: float | None = None

maximum class-attribute instance-attribute 

maximum: float | None = None

ObjectiveResult dataclass 

ObjectiveResult(score: float | None, status: ValueStatus)

Aggregated score and confidence state.

score instance-attribute 

score: float | None

status instance-attribute 

status: ValueStatus

Objective dataclass 

Objective(terms: tuple[ObjectiveTerm, ...])

Combine metric observations into a benefit-oriented weighted score.

terms instance-attribute 

terms: tuple[ObjectiveTerm, ...]

evaluate 

evaluate(metrics: Sequence[MetricValue]) -> ObjectiveResult
Source code in src/tunarag/objective.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def evaluate(self, metrics: Sequence[MetricValue]) -> ObjectiveResult:
    by_name = {metric.name: metric for metric in metrics}
    score = 0.0
    status = ValueStatus.EXACT
    for term in self.terms:
        metric = by_name.get(term.metric)
        if metric is None or metric.value is None:
            return ObjectiveResult(None, ValueStatus.UNAVAILABLE)
        value = float(metric.value)
        if term.minimum is not None and term.maximum is not None:
            value = (value - term.minimum) / (term.maximum - term.minimum)
            value = max(0.0, min(1.0, value))
        if term.direction is ObjectiveDirection.MINIMIZE:
            value = 1.0 - value if term.minimum is not None else -value
        score += term.weight * value
        if metric.status is ValueStatus.ESTIMATED:
            status = ValueStatus.ESTIMATED
    return ObjectiveResult(score, status)