Skip to content

core

genlm.eval.core

Dataset

Bases: Generic[T], ABC

Base class for datasets that yield instances conforming to a Pydantic schema.

Source code in genlm/eval/core/dataset.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class Dataset(Generic[T], ABC):
    """Base class for datasets that yield instances conforming to a Pydantic schema."""

    @abstractmethod
    def __iter__(self) -> Iterator[T]:
        """Iterate over dataset instances.

        Returns:
            Iterator[T]: An iterator over instances conforming to schema T.
        """
        pass  # pragma: no cover

    @property
    @abstractmethod
    def schema(self) -> type[T]:
        """Get the Pydantic schema class for this dataset.

        Returns:
            type[T]: The Pydantic model class defining the schema.
        """
        pass  # pragma: no cover

__iter__() abstractmethod

Iterate over dataset instances.

Returns:

Type Description
Iterator[T]

Iterator[T]: An iterator over instances conforming to schema T.
Source code in genlm/eval/core/dataset.py 
18
19
20
21
22
23
24
25
@abstractmethod
def __iter__(self) -> Iterator[T]:
    """Iterate over dataset instances.

    Returns:
        Iterator[T]: An iterator over instances conforming to schema T.
    """
    pass  # pragma: no cover

schema abstractmethod property

Get the Pydantic schema class for this dataset.

Returns:

Type Description
type[T]

type[T]: The Pydantic model class defining the schema.

Instance

Bases: BaseModel

Base class for dataset instances that conform to a Pydantic schema.

Source code in genlm/eval/core/dataset.py 
6
7
8
9
class Instance(BaseModel):
    """Base class for dataset instances that conform to a Pydantic schema."""

    instance_id: Union[int, str]

ModelAdaptor

Bases: Protocol

Protocol for async model adapters that process instances into model outputs.

Source code in genlm/eval/core/model.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
class ModelAdaptor(Protocol):
    """Protocol for async model adapters that process instances into model outputs."""

    async def __call__(
        self, instance: Instance, output_dir: str, replicate: int
    ) -> ModelOutput:
        """Process an instance and generate model outputs.

        Args:
            instance (Instance): Input dataset instance to process
            output_dir (str): Directory for saving any intermediate results
            replicate (int): Replicate index for multiple evaluation runs

        Returns:
            (ModelOutput): Model output containing responses and runtime information
        """
        ...  # pragma: no cover

__call__(instance, output_dir, replicate) async

Process an instance and generate model outputs.

Parameters:

Name Type Description Default
instance Instance

Input dataset instance to process

 required
output_dir str

Directory for saving any intermediate results

 required
replicate int

Replicate index for multiple evaluation runs

 required

Returns:

Type Description
ModelOutput

Model output containing responses and runtime information
Source code in genlm/eval/core/model.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
async def __call__(
    self, instance: Instance, output_dir: str, replicate: int
) -> ModelOutput:
    """Process an instance and generate model outputs.

    Args:
        instance (Instance): Input dataset instance to process
        output_dir (str): Directory for saving any intermediate results
        replicate (int): Replicate index for multiple evaluation runs

    Returns:
        (ModelOutput): Model output containing responses and runtime information
    """
    ...  # pragma: no cover

ModelOutput

Bases: BaseModel

Collection of model responses with execution metadata.

Source code in genlm/eval/core/model.py 
14
15
16
17
18
19
class ModelOutput(BaseModel):
    """Collection of model responses with execution metadata."""

    responses: List[ModelResponse]
    runtime_seconds: Optional[float] = None
    metadata: Optional[Dict[str, Any]] = None

ModelResponse

Bases: BaseModel

Single model response containing generated text, probability, and optional metadata.

Source code in genlm/eval/core/model.py 
 6
 7
 8
 9
10
11
class ModelResponse(BaseModel):
    """Single model response containing generated text, probability, and optional metadata."""

    response: Any
    weight: float
    metadata: Optional[Dict[str, Any]] = None

Evaluator

Bases: Generic[T], ABC

Base class for evaluators that handle response evaluation.

Source code in genlm/eval/core/evaluator.py 
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
61
class Evaluator(Generic[T], ABC):
    """Base class for evaluators that handle response evaluation."""

    @abstractmethod
    def evaluate_sample(self, instance, response):
        """Evaluate a single response for correctness.

        Args:
            instance (T): The dataset instance being evaluated.
            response (Any): The model's response, which is given by the response attribute of a `ModelOutput` object.

        Returns:
            (EvaluationResult): The evaluation result.
        """
        pass  # pragma: no cover

    def evaluate_ensemble(self, instance: T, output: ModelOutput) -> Dict[str, Any]:
        """Evaluate the complete ensemble of weighted samples using weighted accuracy.

        Args:
            instance (T): The dataset instance being evaluated.
            output (ModelOutput): The complete model output including ensemble responses.

        Returns:
            (Dict[str, Any]): Dictionary containing evaluation metrics.
        """
        weighted_accuracy = 0.0
        results = []
        for response in output.responses:
            result = self.evaluate_sample(instance, response.response)
            weighted_accuracy += result.score * response.weight
            results.append(
                {
                    "score": result.score,
                    "desc": result.desc,
                    "metadata": result.metadata,
                }
            )

        return {
            "weighted_accuracy": weighted_accuracy,
            "runtime_seconds": output.runtime_seconds,
            "results": results,
        }

evaluate_sample(instance, response) abstractmethod

Evaluate a single response for correctness.

Parameters:

Name Type Description Default
instance T

The dataset instance being evaluated.

 required
response Any

The model's response, which is given by the response attribute of a ModelOutput object.

 required

Returns:

Type Description
EvaluationResult

The evaluation result.
Source code in genlm/eval/core/evaluator.py 
21
22
23
24
25
26
27
28
29
30
31
32
@abstractmethod
def evaluate_sample(self, instance, response):
    """Evaluate a single response for correctness.

    Args:
        instance (T): The dataset instance being evaluated.
        response (Any): The model's response, which is given by the response attribute of a `ModelOutput` object.

    Returns:
        (EvaluationResult): The evaluation result.
    """
    pass  # pragma: no cover

evaluate_ensemble(instance, output)

Evaluate the complete ensemble of weighted samples using weighted accuracy.

Parameters:

Name Type Description Default
instance T

The dataset instance being evaluated.

 required
output ModelOutput

The complete model output including ensemble responses.

 required

Returns:

Type Description
Dict[str, Any]

Dictionary containing evaluation metrics.
Source code in genlm/eval/core/evaluator.py 
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
61
def evaluate_ensemble(self, instance: T, output: ModelOutput) -> Dict[str, Any]:
    """Evaluate the complete ensemble of weighted samples using weighted accuracy.

    Args:
        instance (T): The dataset instance being evaluated.
        output (ModelOutput): The complete model output including ensemble responses.

    Returns:
        (Dict[str, Any]): Dictionary containing evaluation metrics.
    """
    weighted_accuracy = 0.0
    results = []
    for response in output.responses:
        result = self.evaluate_sample(instance, response.response)
        weighted_accuracy += result.score * response.weight
        results.append(
            {
                "score": result.score,
                "desc": result.desc,
                "metadata": result.metadata,
            }
        )

    return {
        "weighted_accuracy": weighted_accuracy,
        "runtime_seconds": output.runtime_seconds,
        "results": results,
    }

EvaluationResult

Bases: BaseModel

Class for storing evaluation results.

Source code in genlm/eval/core/evaluator.py 
10
11
12
13
14
15
class EvaluationResult(BaseModel):
    """Class for storing evaluation results."""

    score: float
    desc: str
    metadata: Dict[str, Any] = {}

run_evaluation(dataset, model, evaluator, output_dir=None, n_replicates=1, overwrite_results=False, overwrite_outputs=False, max_instances=float('inf'), verbosity=0) async

Run evaluation on a dataset using the provided model and evaluator.

Parameters:

Name Type Description Default
dataset Dataset

The dataset to evaluate on.

 required
model ModelAdaptor

The model adaptor to use for generation.

 required
evaluator Evaluator

The evaluator to use for prompt generation and evaluation.

 required
output_dir str

The directory to save the results. Defaults to None, in which case results are not saved.

 None
n_replicates int

Number of times to replicate the evaluation. Defaults to 1.

 1
overwrite_results bool

Whether to overwrite existing evaluation results. Defaults to False.

 False
overwrite_outputs bool

Whether to overwrite existing output. Defaults to False.

 False
max_instances int

The maximum number of instances to evaluate. Defaults to float("inf").

 float('inf')
verbosity int

The verbosity of the evaluation. Defaults to 0, which is silent.

 0

Returns:

Type Description
Dict[str, Any]

Aggregated evaluation results.
Source code in genlm/eval/core/runner.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 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
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
async def run_evaluation(
    dataset,
    model,
    evaluator,
    output_dir=None,
    n_replicates=1,
    overwrite_results=False,
    overwrite_outputs=False,
    max_instances=float("inf"),
    verbosity=0,
):
    """Run evaluation on a dataset using the provided model and evaluator.

    Args:
        dataset (Dataset): The dataset to evaluate on.
        model (ModelAdaptor): The model adaptor to use for generation.
        evaluator (Evaluator): The evaluator to use for prompt generation and evaluation.
        output_dir (str, optional): The directory to save the results. Defaults to None, in which case results are not saved.
        n_replicates (int, optional): Number of times to replicate the evaluation. Defaults to 1.
        overwrite_results (bool, optional): Whether to overwrite existing evaluation results. Defaults to False.
        overwrite_outputs (bool, optional): Whether to overwrite existing output. Defaults to False.
        max_instances (int, optional): The maximum number of instances to evaluate. Defaults to float("inf").
        verbosity (int, optional): The verbosity of the evaluation. Defaults to 0, which is silent.

    Returns:
        (Dict[str, Any]): Aggregated evaluation results.
    """
    all_results = []
    all_instance_results = []
    all_instance_outputs = []

    if overwrite_outputs and not overwrite_results:
        raise ValueError(
            "Cannot overwrite outputs without overwriting results. (Hint: set overwrite_results=True)"
        )

    if output_dir is not None and not os.path.exists(output_dir):
        os.makedirs(output_dir)  # pragma: no cover

    n_instances = 0
    for instance in dataset:
        n_instances += 1

        instance_results = []
        instance_outputs = []
        instance_id = instance.instance_id

        for i in range(n_replicates):
            output = None
            result = None
            if output_dir is not None:
                instance_output_path = os.path.join(
                    output_dir, f"{instance_id}-{i}-output.json"
                )
                instance_results_path = os.path.join(
                    output_dir, f"{instance_id}-{i}-results.json"
                )

                # Try loading cached files if not overwriting
                if not overwrite_outputs:
                    output = _load_cached_output(instance_output_path)
                if not overwrite_results:
                    result = _load_cached_results(instance_results_path)
            else:
                instance_output_path = None
                instance_results_path = None

            # Generate new output if needed
            wrote_output = False
            if output is None:
                output = await model(instance, output_dir, replicate=i)
                if instance_output_path is not None:
                    wrote_output = True
                    _save_output(output, instance_output_path)

            # Evaluate if we need new results (no results, overwriting results, or wrote new output)
            if result is None or overwrite_results or wrote_output:
                result = evaluator.evaluate_ensemble(instance, output)
                if instance_results_path is not None:
                    _save_results(result, instance_results_path)

            instance_results.append(result)
            instance_outputs.append(output)

        avg_instance_result = {
            "weighted_accuracy": sum(r["weighted_accuracy"] for r in instance_results)
            / n_replicates,
        }
        all_results.append(avg_instance_result)
        all_instance_results.append(instance_results)
        all_instance_outputs.append(instance_outputs)

        if verbosity > 0:
            print(f"Instance {instance}")
            print(
                f"Mean weighted accuracy (instance): {avg_instance_result['weighted_accuracy']}"
            )
            print(
                f"Mean weighted accuracy (total): {sum(r['weighted_accuracy'] for r in all_results) / len(all_results)}"
            )
            print()

        if n_instances >= max_instances:
            break

    return {
        "average_weighted_accuracy": sum(r["weighted_accuracy"] for r in all_results)
        / len(all_results),
        "n_instances": len(all_results),
        "all_instance_results": all_instance_results,
        "all_instance_outputs": all_instance_outputs,
    }