Coverage for haystack/components/evaluators/llm_evaluator.py: 99%
160 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 asyncio
6import json
7from typing import Any
9from tqdm import tqdm
10from tqdm.asyncio import tqdm as async_tqdm
12from haystack import component, default_from_dict, default_to_dict, logging
13from haystack.components.builders import PromptBuilder
14from haystack.components.generators.chat.openai import OpenAIChatGenerator
15from haystack.components.generators.chat.types import ChatGenerator
16from haystack.components.generators.utils import _trace_chat_generator_run
17from haystack.core.serialization import component_to_dict
18from haystack.dataclasses.chat_message import ChatMessage
19from haystack.utils import deserialize_chatgenerator_inplace, deserialize_type, serialize_type
20from haystack.utils.misc import _parse_dict_from_json
22logger = logging.getLogger(__name__)
25@component
26class LLMEvaluator:
27 """
28 Uses an LLM to evaluate inputs based on a prompt containing instructions and examples.
30 The default API requires an OpenAI API key to be provided as an environment variable "OPENAI_API_KEY".
31 The inputs are lists that are user-defined depending on the desired metric.
32 The output is a dictionary with a key `results` containing a list of evaluation results.
33 Each result is a dictionary with user-defined keys and values of either 0 for FALSE or 1 for TRUE respectively.
35 Usage example:
36 ```python
37 from haystack.components.evaluators import LLMEvaluator
38 evaluator = LLMEvaluator(
39 instructions="Is this answer problematic for children?",
40 inputs=[("predicted_answers", list[str])],
41 outputs=["score"],
42 examples=[
43 {"inputs": {"predicted_answers": "Damn, this is straight outta hell!!!"}, "outputs": {"score": 1}},
44 {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}},
45 ],
46 )
47 predicted_answers = [
48 "Football is the most popular sport with around 4 billion followers worldwide",
49 "Python language was created by Guido van Rossum.",
50 ]
51 results = evaluator.run(predicted_answers=predicted_answers)
52 print(results)
53 # {'results': [{'score': 0}, {'score': 0}]}
54 ```
55 """
57 def __init__(
58 self,
59 instructions: str,
60 inputs: list[tuple[str, type[list]]],
61 outputs: list[str],
62 examples: list[dict[str, Any]],
63 progress_bar: bool = True,
64 *,
65 raise_on_failure: bool = True,
66 chat_generator: ChatGenerator | None = None,
67 ) -> None:
68 """
69 Creates an instance of LLMEvaluator.
71 If no LLM is specified using the `chat_generator` parameter, the component will use OpenAI in JSON mode.
73 :param instructions:
74 The prompt instructions to use for evaluation.
75 Should be a question about the inputs that can be answered with yes or no.
76 :param inputs:
77 The inputs that the component expects as incoming connections and that it evaluates.
78 Each input is a tuple of an input name and input type. Input types must be lists.
79 :param outputs:
80 Output names of the evaluation results. They correspond to keys in the output dictionary.
81 :param examples:
82 Few-shot examples conforming to the expected input and output format as defined in the `inputs` and
83 `outputs` parameters.
84 Each example is a dictionary with keys "inputs" and "outputs"
85 They contain the input and output as dictionaries respectively.
86 :param raise_on_failure:
87 If True, the component will raise an exception on an unsuccessful API call.
88 :param progress_bar:
89 Whether to show a progress bar during the evaluation.
90 :param chat_generator:
91 a ChatGenerator instance which represents the LLM.
92 In order for the component to work, the LLM should be configured to return a JSON object. For example,
93 when using the OpenAIChatGenerator, you should pass `{"response_format": {"type": "json_object"}}` in the
94 `generation_kwargs`.
95 """
96 self.validate_init_parameters(inputs, outputs, examples)
97 component.set_input_types(self, **dict(inputs))
99 self.raise_on_failure = raise_on_failure
100 self.instructions = instructions
101 self.inputs = inputs
102 self.outputs = outputs
103 self.examples = examples
104 self.progress_bar = progress_bar
106 template = self.prepare_template()
107 self.builder = PromptBuilder(template=template)
109 if chat_generator is not None:
110 self._chat_generator = chat_generator
111 else:
112 generation_kwargs = {"response_format": {"type": "json_object"}, "seed": 42}
113 self._chat_generator = OpenAIChatGenerator(generation_kwargs=generation_kwargs)
115 def warm_up(self) -> None:
116 """
117 Warm up the underlying chat generator.
118 """
119 if hasattr(self._chat_generator, "warm_up"):
120 self._chat_generator.warm_up()
122 async def warm_up_async(self) -> None:
123 """
124 Warm up the underlying chat generator on the serving event loop.
125 """
126 if hasattr(self._chat_generator, "warm_up_async"):
127 await self._chat_generator.warm_up_async()
128 elif hasattr(self._chat_generator, "warm_up"):
129 self._chat_generator.warm_up()
131 def close(self) -> None:
132 """
133 Release the underlying chat generator's resources.
134 """
135 if hasattr(self._chat_generator, "close"):
136 self._chat_generator.close()
138 async def close_async(self) -> None:
139 """
140 Release the underlying chat generator's async resources.
141 """
142 if hasattr(self._chat_generator, "close_async"):
143 await self._chat_generator.close_async()
144 elif hasattr(self._chat_generator, "close"):
145 self._chat_generator.close()
147 @staticmethod
148 def validate_init_parameters(
149 inputs: list[tuple[str, type[list]]], outputs: list[str], examples: list[dict[str, Any]]
150 ) -> None:
151 """
152 Validate the init parameters.
154 :param inputs:
155 The inputs to validate.
156 :param outputs:
157 The outputs to validate.
158 :param examples:
159 The examples to validate.
161 :raises ValueError:
162 If the inputs are not a list of tuples with a string and a type of list.
163 If the outputs are not a list of strings.
164 If the examples are not a list of dictionaries.
165 If any example does not have keys "inputs" and "outputs" with values that are dictionaries with string keys.
166 """
167 # Validate inputs
168 if (
169 not isinstance(inputs, list)
170 or not all(isinstance(_input, tuple) for _input in inputs)
171 or not all(isinstance(_input[0], str) and _input[1] is not list and len(_input) == 2 for _input in inputs)
172 ):
173 msg = (
174 f"LLM evaluator expects inputs to be a list of tuples. Each tuple must contain an input name and "
175 f"type of list but received {inputs}."
176 )
177 raise ValueError(msg)
179 # Validate outputs
180 if not isinstance(outputs, list) or not all(isinstance(output, str) for output in outputs):
181 msg = f"LLM evaluator expects outputs to be a list of str but received {outputs}."
182 raise ValueError(msg)
184 # Validate examples are lists of dicts
185 if not isinstance(examples, list) or not all(isinstance(example, dict) for example in examples):
186 msg = f"LLM evaluator expects examples to be a list of dictionaries but received {examples}."
187 raise ValueError(msg)
189 # Validate each example
190 for example in examples:
191 if (
192 {"inputs", "outputs"} != example.keys()
193 or not all(isinstance(example[param], dict) for param in ["inputs", "outputs"])
194 or not all(isinstance(key, str) for param in ["inputs", "outputs"] for key in example[param])
195 ):
196 msg = (
197 f"LLM evaluator expects each example to have keys `inputs` and `outputs` with values that are "
198 f"dictionaries with str keys but received {example}."
199 )
200 raise ValueError(msg)
202 @component.output_types(results=list[dict[str, Any]], meta=list[dict[str, Any]] | None)
203 def run(self, **inputs: Any) -> dict[str, Any]:
204 """
205 Run the LLM evaluator.
207 :param inputs:
208 The input values to evaluate. The keys are the input names and the values are lists of input values.
209 :returns:
210 A dictionary with a `results` entry that contains a list of results.
211 Each result is a dictionary containing the keys as defined in the `outputs` parameter of the LLMEvaluator
212 and the evaluation results as the values. If an exception occurs for a particular input value, the result
213 will be `None` for that entry.
214 If the API is "openai" and the response contains a "meta" key, the metadata from OpenAI will be included
215 in the output dictionary, under the key "meta".
216 :raises ValueError:
217 Only in the case that `raise_on_failure` is set to True and the received inputs are not lists or have
218 different lengths, or if the output is not a valid JSON or doesn't contain the expected keys.
219 """
220 self.warm_up()
222 self.validate_input_parameters(dict(self.inputs), inputs)
224 # inputs is a dictionary with keys being input names and values being a list of input values
225 # We need to iterate through the lists in parallel for all keys of the dictionary
226 input_names, values = inputs.keys(), list(zip(*inputs.values(), strict=True))
227 list_of_input_names_to_values = [dict(zip(input_names, v, strict=True)) for v in values]
229 results: list[dict[str, Any] | None] = []
230 metadata = []
231 errors = 0
232 for input_names_to_values in tqdm(list_of_input_names_to_values, disable=not self.progress_bar):
233 prompt = self.builder.run(**input_names_to_values)
234 messages = [ChatMessage.from_user(prompt["prompt"])]
235 try:
236 with _trace_chat_generator_run(self._chat_generator, {"messages": messages}) as span:
237 result = self._chat_generator.run(messages=messages)
238 span.set_content_tag("haystack.component.output", result)
239 except Exception as e:
240 if self.raise_on_failure:
241 raise ValueError(f"Error while generating response for prompt: {prompt}. Error: {e}") from e
242 logger.warning("Error while generating response for prompt: {prompt}. Error: {e}", prompt=prompt, e=e)
243 results.append(None)
244 errors += 1
245 continue
247 parsed_result = _parse_dict_from_json(
248 result["replies"][0].text, expected_keys=self.outputs, raise_on_failure=self.raise_on_failure
249 )
250 if parsed_result is None:
251 results.append(None)
252 errors += 1
253 else:
254 results.append(parsed_result)
256 if result["replies"][0].meta:
257 metadata.append(result["replies"][0].meta)
259 if errors > 0:
260 logger.warning(
261 "LLM evaluator failed for {errors} out of {len(list_of_input_names_to_values)} inputs.",
262 errors=errors,
263 len=len(list_of_input_names_to_values),
264 )
266 return {"results": results, "meta": metadata or None}
268 @component.output_types(results=list[dict[str, Any]], meta=list[dict[str, Any]] | None)
269 async def run_async(self, **inputs: Any) -> dict[str, Any]:
270 """
271 Run the LLM evaluator asynchronously
273 :param inputs:
274 The input values to evaluate. The keys are the input names and the values are lists of input values.
275 :returns:
276 A dictionary with a `results` entry that contains a list of results.
277 Each result is a dictionary containing the keys as defined in the `outputs` parameter of the LLMEvaluator
278 and the evaluation results as the values. If an exception occurs for a particular input value, the result
279 will be `None` for that entry.
280 If the API is "openai" and the response contains a "meta" key, the metadata from OpenAI will be included
281 in the output dictionary, under the key "meta".
282 :raises TypeError:
283 If the chat generator does not support async execution.
284 :raises ValueError:
285 Only in the case that `raise_on_failure` is set to True and the received inputs are not lists or have
286 different lengths, or if the output is not a valid JSON or doesn't contain the expected keys.
287 """
289 await self.warm_up_async()
291 self.validate_input_parameters(dict(self.inputs), inputs)
293 # inputs is a dictionary with keys being input names and values being a list of input values
294 # We need to iterate through the lists in parallel for all keys of the dictionary
295 input_names, values = inputs.keys(), list(zip(*inputs.values(), strict=True))
296 list_of_input_names_to_values = [dict(zip(input_names, v, strict=True)) for v in values]
298 results: list[dict[str, Any] | None] = []
299 metadata = []
300 errors = 0
302 generator_has_async = hasattr(self._chat_generator, "run_async")
303 for input_names_to_values in async_tqdm(list_of_input_names_to_values, disable=not self.progress_bar):
304 prompt = self.builder.run(**input_names_to_values)
305 messages = [ChatMessage.from_user(prompt["prompt"])]
306 try:
307 with _trace_chat_generator_run(self._chat_generator, {"messages": messages}) as span:
308 if generator_has_async:
309 result = await self._chat_generator.run_async(messages=messages) # type: ignore[attr-defined]
310 else:
311 logger.debug(
312 "{generator_type} does not implement 'run_async'."
313 " Running the synchronous 'run' method in a thread to avoid blocking the event loop.",
314 generator_type=type(self._chat_generator).__name__,
315 )
316 result = await asyncio.to_thread(self._chat_generator.run, messages=messages)
317 span.set_content_tag("haystack.component.output", result)
318 except Exception as e:
319 if self.raise_on_failure:
320 raise ValueError(f"Error while generating response for prompt: {prompt}. Error: {e}") from e
321 logger.warning("Error while generating response for prompt: {prompt}. Error: {e}", prompt=prompt, e=e)
322 results.append(None)
323 errors += 1
324 continue
326 parsed_result = _parse_dict_from_json(
327 result["replies"][0].text, expected_keys=self.outputs, raise_on_failure=self.raise_on_failure
328 )
329 if parsed_result is None:
330 results.append(None)
331 errors += 1
332 else:
333 results.append(parsed_result)
335 if result["replies"][0].meta:
336 metadata.append(result["replies"][0].meta)
338 if errors > 0:
339 logger.warning(
340 "LLM evaluator failed for {errors} out of {len(list_of_input_names_to_values)} inputs.",
341 errors=errors,
342 len=len(list_of_input_names_to_values),
343 )
345 return {"results": results, "meta": metadata or None}
347 def prepare_template(self) -> str:
348 """
349 Prepare the prompt template.
351 Combine instructions, inputs, outputs, and examples into one prompt template with the following format:
352 Instructions:
353 `<instructions>`
355 Generate the response in JSON format with the following keys:
356 `<list of output keys>`
357 Consider the instructions and the examples below to determine those values.
359 Examples:
360 `<examples>`
362 Inputs:
363 `<inputs>`
364 Outputs:
366 :returns:
367 The prompt template.
368 """
369 inputs_section = (
370 "{" + ", ".join([f'"{input_socket[0]}": {{{{ {input_socket[0]} }}}}' for input_socket in self.inputs]) + "}"
371 )
373 examples_section = "\n".join(
374 [
375 "Inputs:\n" + json.dumps(example["inputs"]) + "\nOutputs:\n" + json.dumps(example["outputs"])
376 for example in self.examples
377 ]
378 )
379 return (
380 f"Instructions:\n"
381 f"{self.instructions}\n\n"
382 f"Generate the response in JSON format with the following keys:\n"
383 f"{json.dumps(self.outputs)}\n"
384 f"Consider the instructions and the examples below to determine those values.\n\n"
385 f"Examples:\n"
386 f"{examples_section}\n\n"
387 f"Inputs:\n"
388 f"{inputs_section}\n"
389 f"Outputs:\n"
390 )
392 def to_dict(self) -> dict[str, Any]:
393 """
394 Serialize this component to a dictionary.
396 :returns:
397 The serialized component as a dictionary.
398 """
399 # Since we cannot currently serialize tuples, convert the inputs to a list.
400 inputs = [[name, serialize_type(type_)] for name, type_ in self.inputs]
401 return default_to_dict(
402 self,
403 instructions=self.instructions,
404 inputs=inputs,
405 outputs=self.outputs,
406 examples=self.examples,
407 chat_generator=component_to_dict(obj=self._chat_generator, name="chat_generator"),
408 progress_bar=self.progress_bar,
409 raise_on_failure=self.raise_on_failure,
410 )
412 @classmethod
413 def from_dict(cls, data: dict[str, Any]) -> "LLMEvaluator":
414 """
415 Deserialize this component from a dictionary.
417 :param data:
418 The dictionary representation of this component.
419 :returns:
420 The deserialized component instance.
421 """
422 data["init_parameters"]["inputs"] = [
423 (name, deserialize_type(type_)) for name, type_ in data["init_parameters"]["inputs"]
424 ]
426 if data["init_parameters"].get("chat_generator"):
427 deserialize_chatgenerator_inplace(data["init_parameters"], key="chat_generator")
429 return default_from_dict(cls, data)
431 @staticmethod
432 def validate_input_parameters(expected: dict[str, Any], received: dict[str, Any]) -> None:
433 """
434 Validate the input parameters.
436 :param expected:
437 The expected input parameters.
438 :param received:
439 The received input parameters.
441 :raises ValueError:
442 If not all expected inputs are present in the received inputs
443 If the received inputs are not lists or have different lengths
444 """
445 # Validate that all expected inputs are present in the received inputs
446 for param in expected:
447 if param not in received:
448 msg = f"LLM evaluator expected input parameter '{param}' but received only {received.keys()}."
449 raise ValueError(msg)
451 # Validate that all received inputs are lists
452 if not all(isinstance(_input, list) for _input in received.values()):
453 msg = (
454 "LLM evaluator expects all input values to be lists but received "
455 f"{[type(_input) for _input in received.values()]}."
456 )
457 raise ValueError(msg)
459 # Validate that all received inputs are of the same length
460 inputs = received.values()
461 length = len(next(iter(inputs)))
462 if not all(len(_input) == length for _input in inputs):
463 msg = (
464 f"LLM evaluator expects all input lists to have the same length but received {inputs} with lengths "
465 f"{[len(_input) for _input in inputs]}."
466 )
467 raise ValueError(msg)