|
|
"""LLM Chains for evaluating question answering.""" |
|
|
|
|
|
from __future__ import annotations |
|
|
|
|
|
import re |
|
|
import string |
|
|
from typing import Any, List, Optional, Sequence, Tuple |
|
|
|
|
|
from langchain_core.callbacks.manager import Callbacks |
|
|
from langchain_core.language_models import BaseLanguageModel |
|
|
from langchain_core.prompts import PromptTemplate |
|
|
from pydantic import ConfigDict |
|
|
|
|
|
from langchain.chains.llm import LLMChain |
|
|
from langchain.evaluation.qa.eval_prompt import CONTEXT_PROMPT, COT_PROMPT, PROMPT |
|
|
from langchain.evaluation.schema import LLMEvalChain, StringEvaluator |
|
|
from langchain.schema import RUN_KEY |
|
|
|
|
|
|
|
|
def _get_score(text: str) -> Optional[Tuple[str, int]]: |
|
|
match = re.search(r"grade:\s*(correct|incorrect)", text.strip(), re.IGNORECASE) |
|
|
if match: |
|
|
if match.group(1).upper() == "CORRECT": |
|
|
return "CORRECT", 1 |
|
|
elif match.group(1).upper() == "INCORRECT": |
|
|
return "INCORRECT", 0 |
|
|
try: |
|
|
first_word = ( |
|
|
text.strip().split()[0].translate(str.maketrans("", "", string.punctuation)) |
|
|
) |
|
|
if first_word.upper() == "CORRECT": |
|
|
return "CORRECT", 1 |
|
|
elif first_word.upper() == "INCORRECT": |
|
|
return "INCORRECT", 0 |
|
|
last_word = ( |
|
|
text.strip() |
|
|
.split()[-1] |
|
|
.translate(str.maketrans("", "", string.punctuation)) |
|
|
) |
|
|
if last_word.upper() == "CORRECT": |
|
|
return "CORRECT", 1 |
|
|
elif last_word.upper() == "INCORRECT": |
|
|
return "INCORRECT", 0 |
|
|
except IndexError: |
|
|
pass |
|
|
return None |
|
|
|
|
|
|
|
|
def _parse_string_eval_output(text: str) -> dict: |
|
|
"""Parse the output text. |
|
|
|
|
|
Args: |
|
|
text (str): The output text to parse. |
|
|
|
|
|
Returns: |
|
|
Any: The parsed output. |
|
|
""" |
|
|
reasoning = text.strip() |
|
|
parsed_scores = _get_score(reasoning) |
|
|
if parsed_scores is None: |
|
|
value, score = None, None |
|
|
else: |
|
|
value, score = parsed_scores |
|
|
return { |
|
|
"reasoning": reasoning, |
|
|
"value": value, |
|
|
"score": score, |
|
|
} |
|
|
|
|
|
|
|
|
class QAEvalChain(LLMChain, StringEvaluator, LLMEvalChain): |
|
|
"""LLM Chain for evaluating question answering.""" |
|
|
|
|
|
output_key: str = "results" |
|
|
|
|
|
model_config = ConfigDict( |
|
|
extra="ignore", |
|
|
) |
|
|
|
|
|
@classmethod |
|
|
def is_lc_serializable(cls) -> bool: |
|
|
return False |
|
|
|
|
|
@property |
|
|
def evaluation_name(self) -> str: |
|
|
return "correctness" |
|
|
|
|
|
@property |
|
|
def requires_reference(self) -> bool: |
|
|
return True |
|
|
|
|
|
@property |
|
|
def requires_input(self) -> bool: |
|
|
return True |
|
|
|
|
|
@classmethod |
|
|
def from_llm( |
|
|
cls, |
|
|
llm: BaseLanguageModel, |
|
|
prompt: Optional[PromptTemplate] = None, |
|
|
**kwargs: Any, |
|
|
) -> QAEvalChain: |
|
|
"""Load QA Eval Chain from LLM. |
|
|
|
|
|
Args: |
|
|
llm (BaseLanguageModel): the base language model to use. |
|
|
|
|
|
prompt (PromptTemplate): A prompt template containing the input_variables: |
|
|
'input', 'answer' and 'result' that will be used as the prompt |
|
|
for evaluation. |
|
|
Defaults to PROMPT. |
|
|
|
|
|
**kwargs: additional keyword arguments. |
|
|
|
|
|
Returns: |
|
|
QAEvalChain: the loaded QA eval chain. |
|
|
""" |
|
|
prompt = prompt or PROMPT |
|
|
expected_input_vars = {"query", "answer", "result"} |
|
|
if expected_input_vars != set(prompt.input_variables): |
|
|
raise ValueError( |
|
|
f"Input variables should be {expected_input_vars}, " |
|
|
f"but got {prompt.input_variables}" |
|
|
) |
|
|
return cls(llm=llm, prompt=prompt, **kwargs) |
|
|
|
|
|
def evaluate( |
|
|
self, |
|
|
examples: Sequence[dict], |
|
|
predictions: Sequence[dict], |
|
|
question_key: str = "query", |
|
|
answer_key: str = "answer", |
|
|
prediction_key: str = "result", |
|
|
*, |
|
|
callbacks: Callbacks = None, |
|
|
) -> List[dict]: |
|
|
"""Evaluate question answering examples and predictions.""" |
|
|
inputs = [ |
|
|
{ |
|
|
"query": example[question_key], |
|
|
"answer": example[answer_key], |
|
|
"result": predictions[i][prediction_key], |
|
|
} |
|
|
for i, example in enumerate(examples) |
|
|
] |
|
|
|
|
|
return self.apply(inputs, callbacks=callbacks) |
|
|
|
|
|
def _prepare_output(self, result: dict) -> dict: |
|
|
parsed_result = _parse_string_eval_output(result[self.output_key]) |
|
|
if RUN_KEY in result: |
|
|
parsed_result[RUN_KEY] = result[RUN_KEY] |
|
|
return parsed_result |
|
|
|
|
|
def _evaluate_strings( |
|
|
self, |
|
|
*, |
|
|
prediction: str, |
|
|
reference: Optional[str] = None, |
|
|
input: Optional[str] = None, |
|
|
callbacks: Callbacks = None, |
|
|
include_run_info: bool = False, |
|
|
**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 |
|
|
callbacks (Callbacks, optional): the callbacks to use for tracing. |
|
|
include_run_info (bool, optional): whether to include run info in the |
|
|
returned results. |
|
|
**kwargs: additional keyword arguments, including callbacks, tags, etc. |
|
|
Returns: |
|
|
dict: The evaluation results containing the score or value. |
|
|
""" |
|
|
result = self( |
|
|
{ |
|
|
"query": input, |
|
|
"answer": reference, |
|
|
"result": prediction, |
|
|
}, |
|
|
callbacks=callbacks, |
|
|
include_run_info=include_run_info, |
|
|
) |
|
|
return self._prepare_output(result) |
|
|
|
|
|
async def _aevaluate_strings( |
|
|
self, |
|
|
*, |
|
|
prediction: str, |
|
|
reference: Optional[str] = None, |
|
|
input: Optional[str] = None, |
|
|
callbacks: Callbacks = None, |
|
|
include_run_info: bool = False, |
|
|
**kwargs: Any, |
|
|
) -> dict: |
|
|
result = await self.acall( |
|
|
inputs={"query": input, "answer": reference, "result": prediction}, |
|
|
callbacks=callbacks, |
|
|
include_run_info=include_run_info, |
|
|
) |
|
|
return self._prepare_output(result) |
|
|
|
|
|
|
|
|
class ContextQAEvalChain(LLMChain, StringEvaluator, LLMEvalChain): |
|
|
"""LLM Chain for evaluating QA w/o GT based on context""" |
|
|
|
|
|
@classmethod |
|
|
def is_lc_serializable(cls) -> bool: |
|
|
return False |
|
|
|
|
|
@property |
|
|
def requires_reference(self) -> bool: |
|
|
"""Whether the chain requires a reference string.""" |
|
|
return True |
|
|
|
|
|
@property |
|
|
def requires_input(self) -> bool: |
|
|
"""Whether the chain requires an input string.""" |
|
|
return True |
|
|
|
|
|
model_config = ConfigDict( |
|
|
extra="ignore", |
|
|
) |
|
|
|
|
|
@classmethod |
|
|
def _validate_input_vars(cls, prompt: PromptTemplate) -> None: |
|
|
expected_input_vars = {"query", "context", "result"} |
|
|
if expected_input_vars != set(prompt.input_variables): |
|
|
raise ValueError( |
|
|
f"Input variables should be {expected_input_vars}, " |
|
|
f"but got {prompt.input_variables}" |
|
|
) |
|
|
|
|
|
@property |
|
|
def evaluation_name(self) -> str: |
|
|
return "Contextual Accuracy" |
|
|
|
|
|
@classmethod |
|
|
def from_llm( |
|
|
cls, |
|
|
llm: BaseLanguageModel, |
|
|
prompt: Optional[PromptTemplate] = None, |
|
|
**kwargs: Any, |
|
|
) -> ContextQAEvalChain: |
|
|
"""Load QA Eval Chain from LLM. |
|
|
|
|
|
Args: |
|
|
llm (BaseLanguageModel): the base language model to use. |
|
|
|
|
|
prompt (PromptTemplate): A prompt template containing the input_variables: |
|
|
'query', 'context' and 'result' that will be used as the prompt |
|
|
for evaluation. |
|
|
Defaults to PROMPT. |
|
|
|
|
|
**kwargs: additional keyword arguments. |
|
|
|
|
|
Returns: |
|
|
ContextQAEvalChain: the loaded QA eval chain. |
|
|
""" |
|
|
prompt = prompt or CONTEXT_PROMPT |
|
|
cls._validate_input_vars(prompt) |
|
|
return cls(llm=llm, prompt=prompt, **kwargs) |
|
|
|
|
|
def evaluate( |
|
|
self, |
|
|
examples: List[dict], |
|
|
predictions: List[dict], |
|
|
question_key: str = "query", |
|
|
context_key: str = "context", |
|
|
prediction_key: str = "result", |
|
|
*, |
|
|
callbacks: Callbacks = None, |
|
|
) -> List[dict]: |
|
|
"""Evaluate question answering examples and predictions.""" |
|
|
inputs = [ |
|
|
{ |
|
|
"query": example[question_key], |
|
|
"context": example[context_key], |
|
|
"result": predictions[i][prediction_key], |
|
|
} |
|
|
for i, example in enumerate(examples) |
|
|
] |
|
|
|
|
|
return self.apply(inputs, callbacks=callbacks) |
|
|
|
|
|
def _prepare_output(self, result: dict) -> dict: |
|
|
parsed_result = _parse_string_eval_output(result[self.output_key]) |
|
|
if RUN_KEY in result: |
|
|
parsed_result[RUN_KEY] = result[RUN_KEY] |
|
|
return parsed_result |
|
|
|
|
|
def _evaluate_strings( |
|
|
self, |
|
|
*, |
|
|
prediction: str, |
|
|
reference: Optional[str] = None, |
|
|
input: Optional[str] = None, |
|
|
callbacks: Callbacks = None, |
|
|
include_run_info: bool = False, |
|
|
**kwargs: Any, |
|
|
) -> dict: |
|
|
result = self( |
|
|
{ |
|
|
"query": input, |
|
|
"context": reference, |
|
|
"result": prediction, |
|
|
}, |
|
|
callbacks=callbacks, |
|
|
include_run_info=include_run_info, |
|
|
) |
|
|
return self._prepare_output(result) |
|
|
|
|
|
async def _aevaluate_strings( |
|
|
self, |
|
|
*, |
|
|
prediction: str, |
|
|
reference: Optional[str] = None, |
|
|
input: Optional[str] = None, |
|
|
callbacks: Callbacks = None, |
|
|
include_run_info: bool = False, |
|
|
**kwargs: Any, |
|
|
) -> dict: |
|
|
result = await self.acall( |
|
|
inputs={"query": input, "context": reference, "result": prediction}, |
|
|
callbacks=callbacks, |
|
|
include_run_info=include_run_info, |
|
|
) |
|
|
return self._prepare_output(result) |
|
|
|
|
|
|
|
|
class CotQAEvalChain(ContextQAEvalChain): |
|
|
"""LLM Chain for evaluating QA using chain of thought reasoning.""" |
|
|
|
|
|
@classmethod |
|
|
def is_lc_serializable(cls) -> bool: |
|
|
return False |
|
|
|
|
|
@property |
|
|
def evaluation_name(self) -> str: |
|
|
return "COT Contextual Accuracy" |
|
|
|
|
|
@classmethod |
|
|
def from_llm( |
|
|
cls, |
|
|
llm: BaseLanguageModel, |
|
|
prompt: Optional[PromptTemplate] = None, |
|
|
**kwargs: Any, |
|
|
) -> CotQAEvalChain: |
|
|
"""Load QA Eval Chain from LLM.""" |
|
|
prompt = prompt or COT_PROMPT |
|
|
cls._validate_input_vars(prompt) |
|
|
return cls(llm=llm, prompt=prompt, **kwargs) |
|
|
|
