Spaces:
Running on Zero
Running on Zero
File size: 1,994 Bytes
d047469 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 | """Select the evaluator implementation used by the hosted pipeline."""
import importlib
import os
DEFAULT_EVALUATOR_VERSION = "v1"
KNOWN_EVALUATOR_VERSIONS = ("v1", "v2")
_IMPLEMENTATIONS = {
"v1": ("graph", "evaluate_call"),
}
class EvaluatorConfigurationError(ValueError):
"""Raised when EVALUATOR_VERSION is not a recognized version."""
class EvaluatorUnavailableError(RuntimeError):
"""Raised when a recognized evaluator has not been implemented yet."""
def resolve_evaluator_version(explicit=None):
"""Return an available evaluator version or raise a descriptive error."""
version = (
explicit
if explicit is not None
else os.environ.get("EVALUATOR_VERSION", DEFAULT_EVALUATOR_VERSION)
)
version = str(version).strip().lower()
if version not in KNOWN_EVALUATOR_VERSIONS:
choices = ", ".join(KNOWN_EVALUATOR_VERSIONS)
raise EvaluatorConfigurationError(
f"Unknown EVALUATOR_VERSION={version!r}; expected one of: {choices}"
)
if version not in _IMPLEMENTATIONS:
raise EvaluatorUnavailableError(
f"Evaluator {version!r} is reserved but not implemented"
)
return version
def evaluate_call(call_id, results_dir="results", evaluator_version=None):
"""Run the selected evaluator and attach implementation provenance."""
version = resolve_evaluator_version(evaluator_version)
module_name, function_name = _IMPLEMENTATIONS[version]
implementation = importlib.import_module(module_name)
run = getattr(implementation, function_name)
evaluation = run(call_id, results_dir=results_dir)
if not isinstance(evaluation, dict):
raise TypeError(
f"Evaluator {version!r} returned {type(evaluation).__name__}, expected dict"
)
evaluation["_evaluator"] = {
"version": version,
"implementation": module_name,
"rubric_version": evaluation.get("rubric_version"),
}
return evaluation
|