File size: 6,372 Bytes
b462f85 f418928 b462f85 058c80a b462f85 f418928 b462f85 f418928 b462f85 f418928 b462f85 f418928 b462f85 f418928 b462f85 4d23392 b462f85 f418928 b462f85 f418928 b462f85 058c80a b462f85 058c80a b462f85 058c80a b462f85 f418928 b462f85 100c2eb 058c80a 100c2eb |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 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 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 |
from typing import Any, Dict, List, Literal, Optional
from .api import evaluate, produce
from .artifact import Artifact, settings
from .inference import InferenceEngine, OpenAiInferenceEngine
from .metrics import BulkInstanceMetric
from .operator import SequentialOperator
class LLMAsJudge(BulkInstanceMetric):
"""LLM as judge based metric class for evaluating correctness.
Attributes:
main_score (str): The main score label used for evaluation.
task (Literal["rating.single_turn"]): The type of task the llm-as-judge runs. This defines the output and input
format of the jude model.
template (str): The template used when generating inputs for the judge llm.
format (str): The format used when generating inputs for judge llm.
system_prompt (str): The system prompt used when generating inputs for judge llm.
strip_system_prompt_and_format_from_inputs (bool): Whether to strip the system prompt and formatting from the
inputs that the models that is being judges received, when they are inserted to the llm-as-judge prompt.
inference_model (InferenceEngine): the module that creates the inference of the judge llm.
reduction_map (dict): A dictionary specifying the reduction method for the metric.
batch_size (int): The size of the bulk.
"""
main_score: str = "llm_as_judge"
task: Literal["rating.single_turn", "single_turn_with_reference"]
template: str
format: Optional[str] = None
system_prompt: Optional[str] = None
strip_system_prompt_and_format_from_inputs: bool = True
inference_model: InferenceEngine
reduction_map: Optional[Dict[str, List[str]]] = None
batch_size: int = 32
def _get_input_instances(self, task_data: List[Dict]) -> List:
if self.strip_system_prompt_and_format_from_inputs:
instances = []
for task_data_instance in task_data:
template = task_data_instance["metadata"]["template"]
instance = SequentialOperator(
steps=[template, "formats.empty"]
).process_instance(
{"inputs": task_data_instance, "outputs": task_data_instance}
)
instances.append(instance["source"])
"""
We also have access to: instance["target"]
instance["references"]
"""
return instances
return [t["source"] for t in task_data]
def _get_instance_for_judge_model(
self, input_instances: List[str], predictions: List, references: List
) -> List[Dict]:
if self.task == "rating.single_turn":
instances = [
{
"question": input_instance,
"answer": prediction,
"rating": 5.0, # This is a dummy value that is not used in practice
}
for input_instance, prediction, reference in zip(
input_instances, predictions, references
)
]
elif self.task == "rating.single_turn_with_reference":
instances = [
{
"question": input_instance,
"answer": prediction,
"reference_answer": reference[0],
"rating": 5.0, # This is a dummy value that is not used in practice
}
for input_instance, prediction, reference in zip(
input_instances, predictions, references
)
]
else:
raise NotImplementedError(
f"Error in 'LLMAsJudge' metric. {self.task} is not a supported task type."
)
return instances
def prepare(self):
super().prepare()
if self.reduction_map is None:
self.reduction_map = {"mean": [self.main_score]}
supported_tasks = ["rating.single_turn", "rating.single_turn_with_reference"]
assert self.task in supported_tasks, (
f"Error in 'LLMAsJudge' metric. {self.task} is not a supported task type."
f"The supported tasks types are: {', '.join(supported_tasks)}."
)
if isinstance(self.inference_model, OpenAiInferenceEngine):
if self.format:
raise ValueError(
"Error in 'LLMAsJudge' metric. Inference model 'OpenAiInferenceEngine' does "
"not support formatting. Please remove the format definition from the recipe"
" (OpenAi Chat API take care of the formatting automatically)."
)
if self.system_prompt:
raise ValueError(
"Error in 'LLMAsJudge' metric. Inference model 'OpenAiInferenceEngine' does "
"not support system prompt. Please remove the system_prompt definition from the recipe"
" (Current implementation of Unitxt does not support this."
" Support will be added in future updates)."
)
def compute(
self,
references: List[List[Any]],
predictions: List[Any],
task_data: List[Dict],
) -> List[Dict[str, Any]]:
input_instances = self._get_input_instances(task_data)
instances = self._get_instance_for_judge_model(
input_instances, predictions, references
)
card = f"cards.dynamic_cards_for_llm_judges.{self.task}"
recipe_args = {
"card": card,
"template": self.template,
"demos_pool_size": 0,
"num_demos": 0,
"__type__": settings.default_recipe,
}
if self.system_prompt:
recipe_args["system_prompt"] = self.system_prompt
if self.format:
recipe_args["format"] = self.format
recipe = Artifact.from_dict(recipe_args)
dataset = produce(instances, recipe)
verdicts = self.inference_model.infer(dataset)
meta_scores = evaluate(predictions=verdicts, data=dataset)
return [
{
self.main_score: instance["processed_prediction"],
"judge_raw_output": verdict,
}
for instance, verdict in zip(meta_scores, verdicts)
]
|