| """Interfaces to be implemented by general evaluators.""" |
|
|
| from __future__ import annotations |
|
|
| import logging |
| from abc import ABC, abstractmethod |
| from enum import Enum |
| from typing import Any, Optional, Sequence, Tuple, Union |
| from warnings import warn |
|
|
| from langchain_core.agents import AgentAction |
| from langchain_core.language_models import BaseLanguageModel |
| from langchain_core.runnables.config import run_in_executor |
|
|
| from langchain.chains.base import Chain |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| class EvaluatorType(str, Enum): |
| """The types of the evaluators.""" |
|
|
| QA = "qa" |
| """Question answering evaluator, which grades answers to questions |
| directly using an LLM.""" |
| COT_QA = "cot_qa" |
| """Chain of thought question answering evaluator, which grades |
| answers to questions using |
| chain of thought 'reasoning'.""" |
| CONTEXT_QA = "context_qa" |
| """Question answering evaluator that incorporates 'context' in the response.""" |
| PAIRWISE_STRING = "pairwise_string" |
| """The pairwise string evaluator, which predicts the preferred prediction from |
| between two models.""" |
| SCORE_STRING = "score_string" |
| """The scored string evaluator, which gives a score between 1 and 10 |
| to a prediction.""" |
| LABELED_PAIRWISE_STRING = "labeled_pairwise_string" |
| """The labeled pairwise string evaluator, which predicts the preferred prediction |
| from between two models based on a ground truth reference label.""" |
| LABELED_SCORE_STRING = "labeled_score_string" |
| """The labeled scored string evaluator, which gives a score between 1 and 10 |
| to a prediction based on a ground truth reference label.""" |
| AGENT_TRAJECTORY = "trajectory" |
| """The agent trajectory evaluator, which grades the agent's intermediate steps.""" |
| CRITERIA = "criteria" |
| """The criteria evaluator, which evaluates a model based on a |
| custom set of criteria without any reference labels.""" |
| LABELED_CRITERIA = "labeled_criteria" |
| """The labeled criteria evaluator, which evaluates a model based on a |
| custom set of criteria, with a reference label.""" |
| STRING_DISTANCE = "string_distance" |
| """Compare predictions to a reference answer using string edit distances.""" |
| EXACT_MATCH = "exact_match" |
| """Compare predictions to a reference answer using exact matching.""" |
| REGEX_MATCH = "regex_match" |
| """Compare predictions to a reference answer using regular expressions.""" |
| PAIRWISE_STRING_DISTANCE = "pairwise_string_distance" |
| """Compare predictions based on string edit distances.""" |
| EMBEDDING_DISTANCE = "embedding_distance" |
| """Compare a prediction to a reference label using embedding distance.""" |
| PAIRWISE_EMBEDDING_DISTANCE = "pairwise_embedding_distance" |
| """Compare two predictions using embedding distance.""" |
| JSON_VALIDITY = "json_validity" |
| """Check if a prediction is valid JSON.""" |
| JSON_EQUALITY = "json_equality" |
| """Check if a prediction is equal to a reference JSON.""" |
| JSON_EDIT_DISTANCE = "json_edit_distance" |
| """Compute the edit distance between two JSON strings after canonicalization.""" |
| JSON_SCHEMA_VALIDATION = "json_schema_validation" |
| """Check if a prediction is valid JSON according to a JSON schema.""" |
|
|
|
|
| class LLMEvalChain(Chain): |
| """A base class for evaluators that use an LLM.""" |
|
|
| @classmethod |
| @abstractmethod |
| def from_llm(cls, llm: BaseLanguageModel, **kwargs: Any) -> LLMEvalChain: |
| """Create a new evaluator from an LLM.""" |
|
|
|
|
| class _EvalArgsMixin: |
| """Mixin for checking evaluation arguments.""" |
|
|
| @property |
| def requires_reference(self) -> bool: |
| """Whether this evaluator requires a reference label.""" |
| return False |
|
|
| @property |
| def requires_input(self) -> bool: |
| """Whether this evaluator requires an input string.""" |
| return False |
|
|
| @property |
| def _skip_input_warning(self) -> str: |
| """Warning to show when input is ignored.""" |
| return f"Ignoring input in {self.__class__.__name__}, as it is not expected." |
|
|
| @property |
| def _skip_reference_warning(self) -> str: |
| """Warning to show when reference is ignored.""" |
| return ( |
| f"Ignoring reference in {self.__class__.__name__}, as it is not expected." |
| ) |
|
|
| def _check_evaluation_args( |
| self, |
| reference: Optional[str] = None, |
| input: Optional[str] = None, |
| ) -> None: |
| """Check if the evaluation arguments are valid. |
| |
| Args: |
| reference (Optional[str], optional): The reference label. |
| input (Optional[str], optional): The input string. |
| Raises: |
| ValueError: If the evaluator requires an input string but none is provided, |
| or if the evaluator requires a reference label but none is provided. |
| """ |
| if self.requires_input and input is None: |
| raise ValueError(f"{self.__class__.__name__} requires an input string.") |
| elif input is not None and not self.requires_input: |
| warn(self._skip_input_warning) |
| if self.requires_reference and reference is None: |
| raise ValueError(f"{self.__class__.__name__} requires a reference string.") |
| elif reference is not None and not self.requires_reference: |
| warn(self._skip_reference_warning) |
|
|
|
|
| class StringEvaluator(_EvalArgsMixin, ABC): |
| """Grade, tag, or otherwise evaluate predictions relative to their inputs |
| and/or reference labels.""" |
|
|
| @property |
| def evaluation_name(self) -> str: |
| """The name of the evaluation.""" |
| return self.__class__.__name__ |
|
|
| @property |
| def requires_reference(self) -> bool: |
| """Whether this evaluator requires a reference label.""" |
| return False |
|
|
| @abstractmethod |
| def _evaluate_strings( |
| self, |
| *, |
| prediction: Union[str, Any], |
| reference: Optional[Union[str, Any]] = None, |
| input: Optional[Union[str, Any]] = None, |
| **kwargs: Any, |
| ) -> dict: |
| """Evaluate Chain or LLM output, based on optional input and label. |
| |
| Args: |
| prediction (str): The LLM or chain prediction to evaluate. |
| reference (Optional[str], optional): The reference label to evaluate against. |
| input (Optional[str], optional): The input to consider during evaluation. |
| kwargs: Additional keyword arguments, including callbacks, tags, etc. |
| Returns: |
| dict: The evaluation results containing the score or value. |
| It is recommended that the dictionary contain the following keys: |
| - score: the score of the evaluation, if applicable. |
| - value: the string value of the evaluation, if applicable. |
| - reasoning: the reasoning for the evaluation, if applicable. |
| """ |
|
|
| async def _aevaluate_strings( |
| self, |
| *, |
| prediction: Union[str, Any], |
| reference: Optional[Union[str, Any]] = None, |
| input: Optional[Union[str, Any]] = None, |
| **kwargs: Any, |
| ) -> dict: |
| """Asynchronously evaluate Chain or LLM output, based on optional input and label. |
| |
| Args: |
| prediction (str): The LLM or chain prediction to evaluate. |
| reference (Optional[str], optional): The reference label to evaluate against. |
| input (Optional[str], optional): The input to consider during evaluation. |
| kwargs: Additional keyword arguments, including callbacks, tags, etc. |
| Returns: |
| dict: The evaluation results containing the score or value. |
| It is recommended that the dictionary contain the following keys: |
| - score: the score of the evaluation, if applicable. |
| - value: the string value of the evaluation, if applicable. |
| - reasoning: the reasoning for the evaluation, if applicable. |
| """ |
| return await run_in_executor( |
| None, |
| self._evaluate_strings, |
| prediction=prediction, |
| reference=reference, |
| input=input, |
| **kwargs, |
| ) |
|
|
| def evaluate_strings( |
| self, |
| *, |
| prediction: str, |
| reference: Optional[str] = None, |
| input: Optional[str] = None, |
| **kwargs: Any, |
| ) -> dict: |
| """Evaluate Chain or LLM output, based on optional input and label. |
| |
| Args: |
| prediction (str): The LLM or chain prediction to evaluate. |
| reference (Optional[str], optional): The reference label to evaluate against. |
| input (Optional[str], optional): The input to consider during evaluation. |
| kwargs: Additional keyword arguments, including callbacks, tags, etc. |
| Returns: |
| dict: The evaluation results containing the score or value. |
| """ |
| self._check_evaluation_args(reference=reference, input=input) |
| return self._evaluate_strings( |
| prediction=prediction, reference=reference, input=input, **kwargs |
| ) |
|
|
| async def aevaluate_strings( |
| self, |
| *, |
| prediction: str, |
| reference: Optional[str] = None, |
| input: Optional[str] = None, |
| **kwargs: Any, |
| ) -> dict: |
| """Asynchronously evaluate Chain or LLM output, based on optional input and label. |
| |
| Args: |
| prediction (str): The LLM or chain prediction to evaluate. |
| reference (Optional[str], optional): The reference label to evaluate against. |
| input (Optional[str], optional): The input to consider during evaluation. |
| kwargs: Additional keyword arguments, including callbacks, tags, etc. |
| Returns: |
| dict: The evaluation results containing the score or value. |
| """ |
| self._check_evaluation_args(reference=reference, input=input) |
| return await self._aevaluate_strings( |
| prediction=prediction, reference=reference, input=input, **kwargs |
| ) |
|
|
|
|
| class PairwiseStringEvaluator(_EvalArgsMixin, ABC): |
| """Compare the output of two models (or two outputs of the same model).""" |
|
|
| @abstractmethod |
| def _evaluate_string_pairs( |
| self, |
| *, |
| prediction: str, |
| prediction_b: str, |
| reference: Optional[str] = None, |
| input: Optional[str] = None, |
| **kwargs: Any, |
| ) -> dict: |
| """Evaluate the output string pairs. |
| |
| Args: |
| prediction (str): The output string from the first model. |
| prediction_b (str): The output string from the second model. |
| reference (Optional[str], optional): The expected output / reference string. |
| input (Optional[str], optional): The input string. |
| kwargs: Additional keyword arguments, such as callbacks and optional reference strings. |
| Returns: |
| dict: A dictionary containing the preference, scores, and/or other information. |
| """ |
|
|
| async def _aevaluate_string_pairs( |
| self, |
| *, |
| prediction: str, |
| prediction_b: str, |
| reference: Optional[str] = None, |
| input: Optional[str] = None, |
| **kwargs: Any, |
| ) -> dict: |
| """Asynchronously evaluate the output string pairs. |
| |
| Args: |
| prediction (str): The output string from the first model. |
| prediction_b (str): The output string from the second model. |
| reference (Optional[str], optional): The expected output / reference string. |
| input (Optional[str], optional): The input string. |
| kwargs: Additional keyword arguments, such as callbacks and optional reference strings. |
| Returns: |
| dict: A dictionary containing the preference, scores, and/or other information. |
| """ |
| return await run_in_executor( |
| None, |
| self._evaluate_string_pairs, |
| prediction=prediction, |
| prediction_b=prediction_b, |
| reference=reference, |
| input=input, |
| **kwargs, |
| ) |
|
|
| def evaluate_string_pairs( |
| self, |
| *, |
| prediction: str, |
| prediction_b: str, |
| reference: Optional[str] = None, |
| input: Optional[str] = None, |
| **kwargs: Any, |
| ) -> dict: |
| """Evaluate the output string pairs. |
| |
| Args: |
| prediction (str): The output string from the first model. |
| prediction_b (str): The output string from the second model. |
| reference (Optional[str], optional): The expected output / reference string. |
| input (Optional[str], optional): The input string. |
| kwargs: Additional keyword arguments, such as callbacks and optional reference strings. |
| Returns: |
| dict: A dictionary containing the preference, scores, and/or other information. |
| """ |
| self._check_evaluation_args(reference=reference, input=input) |
| return self._evaluate_string_pairs( |
| prediction=prediction, |
| prediction_b=prediction_b, |
| reference=reference, |
| input=input, |
| **kwargs, |
| ) |
|
|
| async def aevaluate_string_pairs( |
| self, |
| *, |
| prediction: str, |
| prediction_b: str, |
| reference: Optional[str] = None, |
| input: Optional[str] = None, |
| **kwargs: Any, |
| ) -> dict: |
| """Asynchronously evaluate the output string pairs. |
| |
| Args: |
| prediction (str): The output string from the first model. |
| prediction_b (str): The output string from the second model. |
| reference (Optional[str], optional): The expected output / reference string. |
| input (Optional[str], optional): The input string. |
| kwargs: Additional keyword arguments, such as callbacks and optional reference strings. |
| Returns: |
| dict: A dictionary containing the preference, scores, and/or other information. |
| """ |
| self._check_evaluation_args(reference=reference, input=input) |
| return await self._aevaluate_string_pairs( |
| prediction=prediction, |
| prediction_b=prediction_b, |
| reference=reference, |
| input=input, |
| **kwargs, |
| ) |
|
|
|
|
| class AgentTrajectoryEvaluator(_EvalArgsMixin, ABC): |
| """Interface for evaluating agent trajectories.""" |
|
|
| @property |
| def requires_input(self) -> bool: |
| """Whether this evaluator requires an input string.""" |
| return True |
|
|
| @abstractmethod |
| def _evaluate_agent_trajectory( |
| self, |
| *, |
| prediction: str, |
| agent_trajectory: Sequence[Tuple[AgentAction, str]], |
| input: str, |
| reference: Optional[str] = None, |
| **kwargs: Any, |
| ) -> dict: |
| """Evaluate a trajectory. |
| |
| Args: |
| prediction (str): The final predicted response. |
| agent_trajectory (List[Tuple[AgentAction, str]]): |
| The intermediate steps forming the agent trajectory. |
| input (str): The input to the agent. |
| reference (Optional[str]): The reference answer. |
| |
| Returns: |
| dict: The evaluation result. |
| """ |
|
|
| async def _aevaluate_agent_trajectory( |
| self, |
| *, |
| prediction: str, |
| agent_trajectory: Sequence[Tuple[AgentAction, str]], |
| input: str, |
| reference: Optional[str] = None, |
| **kwargs: Any, |
| ) -> dict: |
| """Asynchronously evaluate a trajectory. |
| |
| Args: |
| prediction (str): The final predicted response. |
| agent_trajectory (List[Tuple[AgentAction, str]]): |
| The intermediate steps forming the agent trajectory. |
| input (str): The input to the agent. |
| reference (Optional[str]): The reference answer. |
| |
| Returns: |
| dict: The evaluation result. |
| """ |
| return await run_in_executor( |
| None, |
| self._evaluate_agent_trajectory, |
| prediction=prediction, |
| agent_trajectory=agent_trajectory, |
| reference=reference, |
| input=input, |
| **kwargs, |
| ) |
|
|
| def evaluate_agent_trajectory( |
| self, |
| *, |
| prediction: str, |
| agent_trajectory: Sequence[Tuple[AgentAction, str]], |
| input: str, |
| reference: Optional[str] = None, |
| **kwargs: Any, |
| ) -> dict: |
| """Evaluate a trajectory. |
| |
| Args: |
| prediction (str): The final predicted response. |
| agent_trajectory (List[Tuple[AgentAction, str]]): |
| The intermediate steps forming the agent trajectory. |
| input (str): The input to the agent. |
| reference (Optional[str]): The reference answer. |
| |
| Returns: |
| dict: The evaluation result. |
| """ |
| self._check_evaluation_args(reference=reference, input=input) |
| return self._evaluate_agent_trajectory( |
| prediction=prediction, |
| input=input, |
| agent_trajectory=agent_trajectory, |
| reference=reference, |
| **kwargs, |
| ) |
|
|
| async def aevaluate_agent_trajectory( |
| self, |
| *, |
| prediction: str, |
| agent_trajectory: Sequence[Tuple[AgentAction, str]], |
| input: str, |
| reference: Optional[str] = None, |
| **kwargs: Any, |
| ) -> dict: |
| """Asynchronously evaluate a trajectory. |
| |
| Args: |
| prediction (str): The final predicted response. |
| agent_trajectory (List[Tuple[AgentAction, str]]): |
| The intermediate steps forming the agent trajectory. |
| input (str): The input to the agent. |
| reference (Optional[str]): The reference answer. |
| |
| Returns: |
| dict: The evaluation result. |
| """ |
| self._check_evaluation_args(reference=reference, input=input) |
| return await self._aevaluate_agent_trajectory( |
| prediction=prediction, |
| input=input, |
| agent_trajectory=agent_trajectory, |
| reference=reference, |
| **kwargs, |
| ) |
|
|
