Custom LLM judge: a model you can run for free
Exact match needed no model. Nearly every other metric asks a model to judge, and by default that means an OpenAI model and a key. This lesson writes a judge that needs neither.
A metric talks to its judge through one method, generate(prompt, schema). The prompt is a question in plain text. The schema is a Pydantic class describing the shape of the answer, and the judge must return an instance of it. A judge that does this, and inherits DeepEvalBaseLLM, is a model as far as DeepEval is concerned.
Reading a prompt
A real judge reads the prompt with a language model. This one reads it with three small functions, and these are its imports.
import ast
import json
import re
from deepeval.models import DeepEvalBaseLLMSKIP = {"the", "and", "for", "was", "are", "you", "your", "with", "what", "where",
"when", "how", "this", "that", "will", "can", "has", "have", "its", "our",
"from", "into", "should", "does", "did", "also", "order"}
def words(text):
"""The words that carry meaning, lower case, without the common ones."""
return {w for w in re.findall(r"[a-z0-9]+", text.lower()) if len(w) > 2 and w not in SKIP}The first function keeps only the words that carry meaning, so two sentences can be compared by what they say rather than how they are worded.
The second finds a named part of a prompt. DeepEval's prompts put each piece of the test case under a heading such as Actual Output:, so the text under the last such heading is the part that belongs to this test case, after any examples the prompt starts with.
def section(prompt, name):
"""The text under the last line of the prompt that starts with name."""
heads = list(re.finditer(r"(?m)^" + name + r"[^\n]*:[ \t]*\n", prompt))
if not heads:
return ""
rest = prompt[heads[-1].end():]
return re.split(r"\n[A-Z][^\n:]{0,60}:[ \t]*\n|\nJSON:|\n===|\n\*\*", rest)[0].strip()The third says how much of one text also appears in another, from 0 to 1.
def share(part, whole):
"""How much of what part says also appears in whole, from 0 to 1."""
need = words(part)
return len(need & words(whole)) / len(need) if need else 1.0prompt = """Evaluation Steps:
1. Compare the actual output with the expected output.
Actual Output:
Order A17 is on its way.
Expected Output:
Order A17 shipped on 3 March.
JSON:"""
print(section(prompt, "Actual Output"))
print(sorted(words(section(prompt, "Expected Output"))))
print(share(section(prompt, "Expected Output"), section(prompt, "Actual Output")))The actual output was found under its heading. Of the three words that matter in the expected answer, only a17 is in the actual one, so the share is a third. Every decision this judge makes in the course is a rule built on those three functions.
The judge itself
Answers are kept in a dictionary, one function per question a metric can ask. A decorator puts a function in it under a name.
ANSWERS = {}
def answers(key):
def register(answer):
ANSWERS[key] = answer
return answer
return registerclass PretendJudge(DeepEvalBaseLLM):
def __init__(self, score=None):
self.score = score
def load_model(self):
return self
def get_model_name(self):
return "pretend judge"These three methods are part of what DeepEval requires of any model: somewhere to keep settings, load_model, and get_model_name, which is the name printed in reports. score is this judge's own setting, and lesson 10 uses it.
def generate(self, prompt, schema=None):
key = schema.__module__.split(".")[2] + "." + schema.__name__
if schema.__name__.endswith("ScoreReason"):
return schema(reason=read_back_score(prompt))
if key not in ANSWERS:
raise NotImplementedError(f"The pretend judge has no answer for {key} yet")
return schema(**ANSWERS[key](prompt, self))
async def a_generate(self, prompt, schema=None):
return self.generate(prompt, schema)generate turns the schema into a key: the metric the schema belongs to, and the schema's name, like g_eval.Steps. It finds the answer under that key and returns an instance of the schema built from it. A request for a reason on its own is answered by reading the score back out of the prompt, which lesson 13 shows.
a_generate is the same thing for metrics running asynchronously, which is the default. With no network involved it simply calls generate.
Asking it something
from deepeval.metrics.g_eval.schema import Steps
judge = PretendJudge()
print(judge.get_model_name())
try:
judge.generate("Evaluation Criteria:\nGives the shipping date.\n\nJSON:", Steps)
except NotImplementedError as error:
print(error)The judge exists, has a name, and cannot answer anything yet, and it says exactly which question it was missing. Steps is the first thing G-Eval asks a judge, and the next lesson gives the judge its answer.
- Call
sharethe other way round, actual output first, and explain the new number. - Add
"shipped"toSKIPand run the reading example again.
Slow is fine. Stopping is the only problem.