Coverage for haystack/evaluation/eval_run_result.py: 62%
93 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 13:53 +0000
1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2#
3# SPDX-License-Identifier: Apache-2.0
5import csv
6from copy import deepcopy
7from typing import Any, Literal, Union
9from haystack import logging
10from haystack.lazy_imports import LazyImport
12with LazyImport("Run 'pip install pandas'") as pandas_import:
13 from pandas import DataFrame
15logger = logging.getLogger(__name__)
18class EvaluationRunResult:
19 """
20 Contains the inputs and the outputs of an evaluation pipeline and provides methods to inspect them.
21 """
23 def __init__(self, run_name: str, inputs: dict[str, list[Any]], results: dict[str, dict[str, Any]]) -> None:
24 """
25 Initialize a new evaluation run result.
27 :param run_name:
28 Name of the evaluation run.
30 :param inputs:
31 Dictionary containing the inputs used for the run. Each key is the name of the input and its value is a list
32 of input values. The length of the lists should be the same.
34 :param results:
35 Dictionary containing the results of the evaluators used in the evaluation pipeline. Each key is the name
36 of the metric and its value is dictionary with the following keys:
37 - 'score': The aggregated score for the metric.
38 - 'individual_scores': A list of scores for each input sample.
39 """
40 self.run_name = run_name
41 self.inputs = deepcopy(inputs)
42 self.results = deepcopy(results)
44 if len(inputs) == 0:
45 raise ValueError("No inputs provided.")
46 if len({len(lst) for lst in inputs.values()}) != 1:
47 raise ValueError("Lengths of the inputs should be the same.")
49 expected_len = len(next(iter(inputs.values())))
51 for metric, outputs in results.items():
52 if "score" not in outputs:
53 raise ValueError(f"Aggregate score missing for {metric}.")
54 if "individual_scores" not in outputs:
55 raise ValueError(f"Individual scores missing for {metric}.")
57 if len(outputs["individual_scores"]) != expected_len:
58 raise ValueError(
59 f"Length of individual scores for '{metric}' should be the same as the inputs. "
60 f"Got {len(outputs['individual_scores'])} but expected {expected_len}."
61 )
63 @staticmethod
64 def _write_to_csv(csv_file: str, data: dict[str, list[Any]]) -> str:
65 """
66 Write data to a CSV file.
68 :param csv_file: Path to the CSV file to write
69 :param data: Dictionary containing the data to write
70 :return: Status message indicating success or failure
71 """
72 list_lengths = [len(value) for value in data.values()]
74 if len(set(list_lengths)) != 1:
75 raise ValueError("All lists in the JSON must have the same length")
77 try:
78 headers = list(data.keys())
79 num_rows = list_lengths[0]
80 rows = []
82 for i in range(num_rows):
83 row = [data[header][i] for header in headers]
84 rows.append(row)
86 with open(csv_file, "w", newline="") as csvfile:
87 writer = csv.writer(csvfile)
88 writer.writerow(headers)
89 writer.writerows(rows)
91 return f"Data successfully written to {csv_file}"
92 except PermissionError:
93 return f"Error: Permission denied when writing to {csv_file}"
94 except OSError as e:
95 return f"Error writing to {csv_file}: {str(e)}"
96 except Exception as e:
97 return f"Error: {str(e)}"
99 @staticmethod
100 def _handle_output(
101 data: dict[str, list[Any]], output_format: Literal["json", "csv", "df"] = "csv", csv_file: str | None = None
102 ) -> Union[str, "DataFrame", dict[str, list[Any]]]:
103 """
104 Handles output formatting based on `output_format`.
106 :returns: DataFrame for 'df', dict for 'json', or confirmation message for 'csv'
107 """
108 if output_format == "json":
109 return data
111 if output_format == "df":
112 pandas_import.check()
113 return DataFrame(data)
115 if output_format == "csv":
116 if not csv_file:
117 raise ValueError("A file path must be provided in 'csv_file' parameter to save the CSV output.")
118 return EvaluationRunResult._write_to_csv(csv_file, data)
120 raise ValueError(f"Invalid output format '{output_format}' provided. Choose from 'json', 'csv', or 'df'.")
122 def aggregated_report(
123 self, output_format: Literal["json", "csv", "df"] = "json", csv_file: str | None = None
124 ) -> Union[dict[str, list[Any]], "DataFrame", str]:
125 """
126 Generates a report with aggregated scores for each metric.
128 :param output_format: The output format for the report, "json", "csv", or "df", default to "json".
129 :param csv_file: Filepath to save CSV output if `output_format` is "csv", must be provided.
131 :returns:
132 JSON or DataFrame with aggregated scores, in case the output is set to a CSV file, a message confirming the
133 successful write or an error message.
134 """
135 results = {k: v["score"] for k, v in self.results.items()}
136 data = {"metrics": list(results.keys()), "score": list(results.values())}
137 return self._handle_output(data, output_format, csv_file)
139 def detailed_report(
140 self, output_format: Literal["json", "csv", "df"] = "json", csv_file: str | None = None
141 ) -> Union[dict[str, list[Any]], "DataFrame", str]:
142 """
143 Generates a report with detailed scores for each metric.
145 :param output_format: The output format for the report, "json", "csv", or "df", default to "json".
146 :param csv_file: Filepath to save CSV output if `output_format` is "csv", must be provided.
148 :returns:
149 JSON or DataFrame with the detailed scores, in case the output is set to a CSV file, a message confirming
150 the successful write or an error message.
151 """
153 combined_data = {col: self.inputs[col] for col in self.inputs}
155 # enforce columns type consistency
156 scores_columns = list(self.results.keys())
157 for col in scores_columns:
158 col_values = self.results[col]["individual_scores"]
159 if any(isinstance(v, float) for v in col_values):
160 col_values = [float(v) for v in col_values]
161 combined_data[col] = col_values
163 return self._handle_output(combined_data, output_format, csv_file)
165 def comparative_detailed_report(
166 self,
167 other: "EvaluationRunResult",
168 keep_columns: list[str] | None = None,
169 output_format: Literal["json", "csv", "df"] = "json",
170 csv_file: str | None = None,
171 ) -> Union[str, "DataFrame", None]:
172 """
173 Generates a report with detailed scores for each metric from two evaluation runs for comparison.
175 :param other: Results of another evaluation run to compare with.
176 :param keep_columns: List of common column names to keep from the inputs of the evaluation runs to compare.
177 :param output_format: The output format for the report, "json", "csv", or "df", default to "json".
178 :param csv_file: Filepath to save CSV output if `output_format` is "csv", must be provided.
180 :returns:
181 JSON or DataFrame with a comparison of the detailed scores, in case the output is set to a CSV file,
182 a message confirming the successful write or an error message.
183 :raises TypeError: If `other` is not an EvaluationRunResult instance, or if the detailed reports are not
184 dictionaries.
185 :raises ValueError: If the `other` parameter is missing required attributes.
186 """
188 if not isinstance(other, EvaluationRunResult):
189 raise TypeError("Comparative scores can only be computed between EvaluationRunResults.")
191 if not hasattr(other, "run_name") or not hasattr(other, "inputs") or not hasattr(other, "results"):
192 raise ValueError("The 'other' parameter must have 'run_name', 'inputs', and 'results' attributes.")
194 if self.run_name == other.run_name:
195 logger.warning(
196 "The run names of the two evaluation results are the same ('{run_name}')", run_name=self.run_name
197 )
199 if self.inputs.keys() != other.inputs.keys():
200 logger.warning(
201 "The input columns differ between the results; using the input columns of '{run_name}'",
202 run_name=self.run_name,
203 )
205 # got both detailed reports
206 detailed_a = self.detailed_report(output_format="json")
207 detailed_b = other.detailed_report(output_format="json")
209 # ensure both detailed reports are in dictionaries format
210 if not isinstance(detailed_a, dict) or not isinstance(detailed_b, dict):
211 raise TypeError("Detailed reports must be dictionaries.")
213 # determine which columns to ignore
214 if keep_columns is None:
215 ignore = list(self.inputs.keys())
216 else:
217 ignore = [col for col in list(self.inputs.keys()) if col not in keep_columns]
219 # filter out ignored columns from pipe_b_dict
220 filtered_detailed_b = {
221 f"{other.run_name}_{key}": value for key, value in detailed_b.items() if key not in ignore
222 }
224 # rename columns in pipe_a_dict based on ignore list
225 renamed_detailed_a = {
226 (key if key in ignore else f"{self.run_name}_{key}"): value for key, value in detailed_a.items()
227 }
229 # combine both detailed reports
230 combined_results = {**renamed_detailed_a, **filtered_detailed_b}
231 return self._handle_output(combined_results, output_format, csv_file)