Introduction to Creating Metrics in DSPy

Welcome to the third lesson of the "Evaluation in DSPy" course. In this lesson, we will focus on creating metrics, a crucial aspect of evaluating the quality of system outputs in DSPy. Metrics allow us to quantify how well a system performs, providing a basis for improvement and optimization. Building on the data handling skills you acquired in the previous lesson, you will now learn how to define and implement metrics to assess output quality effectively. By the end of this lesson, you will be equipped with the knowledge to create and use metrics in DSPy, setting the stage for practical applications and further exploration.

Basic Metric Functions

To begin, let's explore some basic metric functions that are foundational in evaluating system responses. One such function is validate_answer, which checks if the predicted answer matches the expected answer. Here's how you can use it in DSPy:

def validate_answer(example, pred, trace=None):
    return example.answer.lower() == pred.answer.lower()

This function compares the predicted answer with the example's answer, ignoring case differences. It returns True if they match and False otherwise. This basic validation is useful for tasks where exact matches are required.

Next, let's look at the answer_exact_match and answer_passage_match built-in functions. These functions provide more flexibility by allowing partial matches and checking if the answer is present in a passage. Here's how you can use them:

from dspy.metrics import answer_exact_match, answer_passage_match

exact_match_result = answer_exact_match(example, pred, frac=1.0)
passage_match_result = answer_passage_match(example, pred)

Here its the underlying implementation of these functions:

def _answer_match(prediction, answers, frac=1.0):
    """Returns True if the prediction matches any of the answers."""
    from dspy.dsp.utils import EM, F1

    if frac >= 1.0:
        return EM(prediction, answers)

    return F1(prediction, answers) >= frac
    
def answer_exact_match(example, pred, trace=None, frac=1.0):
    if isinstance(example.answer, str):
        return _answer_match(pred.answer, [example.answer], frac=frac)
    elif isinstance(example.answer, list):
        return _answer_match(pred.answer, example.answer, frac=frac)
    
    raise ValueError(f"Invalid answer type: {type(example.answer)}")

This function uses a helper function _answer_match to determine if the prediction matches any of the answers, allowing for partial matches based on the frac parameter. The answer_passage_match is also implemented in a similar way:

def _passage_match(passages: list[str], answers: list[str]) -> bool:
    """Returns True if any of the passages contains the answer."""
    from dspy.dsp.utils import DPR_normalize, has_answer, normalize_text

    def passage_has_answers(passage: str, answers: list[str]) -> bool:
        """Returns True if the passage contains the answer."""
        return has_answer(
            tokenized_answers=[DPR_normalize(normalize_text(ans)) for ans in answers],
            text=normalize_text(passage),
        )

    return any(passage_has_answers(psg, answers) for psg in passages)

def answer_passage_match(example, pred, trace=None):   
    if isinstance(example.answer, str):
        return _passage_match(pred.context, [example.answer])
    elif isinstance(example.answer, list):
        return _passage_match(pred.context, example.answer)
    
    raise ValueError(f"Invalid answer type: {type(example.answer)}")

The answer_passage_match function is designed to evaluate whether the predicted answer is present within a given passage. It works by checking if any of the expected answers are found within the context of the predicted response. The function uses a helper function _passage_match to perform the actual matching process.

These functions are essential for evaluating the accuracy of system responses, especially in tasks involving text passages.

Completeness and Groundedness Evaluation

In DSPy, evaluating the completeness and groundedness of system responses is crucial for understanding their quality. The CompleteAndGrounded built-in class provides a structured way to perform this evaluation.

Here's how you can use the CompleteAndGrounded class:

from dspy.metrics import CompleteAndGrounded

complete_and_grounded = CompleteAndGrounded(threshold=0.66)
score = complete_and_grounded(example, pred)

This class calculates a score based on the completeness and groundedness of the response, using an F1 score to combine these aspects.

Here's how the CompleteAndGrounded class is implemented:

class AnswerCompleteness(dspy.Signature):
    """
    Estimate the completeness of a system's responses, against the ground truth.
    You will first enumerate key ideas in each response, discuss their overlap, and then report completeness.
    """

    question: str = dspy.InputField()
    ground_truth: str = dspy.InputField()
    system_response: str = dspy.InputField()
    ground_truth_key_ideas: str = dspy.OutputField(desc="enumeration of key ideas in the ground truth")
    system_response_key_ideas: str = dspy.OutputField(desc="enumeration of key ideas in the system response")
    discussion: str = dspy.OutputField(desc="discussion of the overlap between ground truth and system response")
    completeness: float = dspy.OutputField(desc="fraction (out of 1.0) of ground truth covered by the system response")

class AnswerGroundedness(dspy.Signature):
    """
    Estimate the groundedness of a system's responses, against real retrieved documents written by people.
    You will first enumerate whatever non-trivial or check-worthy claims are made in the system response, and then
    discuss the extent to which some or all of them can be deduced from the retrieved context and basic commonsense.
    """

    question: str = dspy.InputField()
    retrieved_context: str = dspy.InputField()
    system_response: str = dspy.InputField()
    system_response_claims: str = dspy.OutputField(desc="enumeration of non-trivial or check-worthy claims in the system response")
    discussion: str = dspy.OutputField(desc="discussion of how supported the claims are by the retrieved context")
    groundedness: float = dspy.OutputField(desc="fraction (out of 1.0) of system response supported by the retrieved context")

class CompleteAndGrounded(dspy.Module):
    def __init__(self, threshold=0.66):
        self.threshold = threshold
        self.completeness_module = dspy.ChainOfThought(AnswerCompleteness)
        self.groundedness_module = dspy.ChainOfThought(AnswerGroundedness)

    def forward(self, example, pred, trace=None):
        completeness = self.completeness_module(question=example.question, ground_truth=example.response, system_response=pred.response)
        groundedness = self.groundedness_module(question=example.question, retrieved_context=pred.context, system_response=pred.response)
        score = f1_score(groundedness.groundedness, completeness.completeness)

        return score if trace is None else score >= self.threshold

It consists of two main components: AnswerCompleteness and AnswerGroundedness.

The AnswerCompleteness component estimates how well a system's response covers the ground truth. It involves enumerating key ideas in both the ground truth and the system response, discussing their overlap, and reporting completeness. Similarly, AnswerGroundedness assesses the extent to which a system's response is supported by retrieved documents and commonsense reasoning.

Semantic Evaluation with F1 Score

Semantic evaluation involves assessing the quality of system responses based on their semantic content. The built-inSemanticF1 class provides a way to perform this evaluation using recall, precision, and F1 score. Here's how you can use it:

from dspy.metrics import SemanticF1

semantic_f1 = SemanticF1(threshold=0.66, decompositional=False)
score = semantic_f1(example, pred)

Recall measures the fraction of ground truth covered by the system response, while precision measures the fraction of the system response covered by the ground truth. The F1 score combines these two metrics to provide a balanced evaluation.

Here's the implementation of the SemanticF1 class:

class SemanticRecallPrecision(dspy.Signature):
    """
    Compare a system's response to the ground truth to compute its recall and precision.
    If asked to reason, enumerate key ideas in each response, and whether they are present in the other response.
    """

    question: str = dspy.InputField()
    ground_truth: str = dspy.InputField()
    system_response: str = dspy.InputField()
    recall: float = dspy.OutputField(desc="fraction (out of 1.0) of ground truth covered by the system response")
    precision: float = dspy.OutputField(desc="fraction (out of 1.0) of system response covered by the ground truth")


class DecompositionalSemanticRecallPrecision(dspy.Signature):
    """
    Compare a system's response to the ground truth to compute recall and precision of key ideas.
    You will first enumerate key ideas in each response, discuss their overlap, and then report recall and precision.
    """

    question: str = dspy.InputField()
    ground_truth: str = dspy.InputField()
    system_response: str = dspy.InputField()
    ground_truth_key_ideas: str = dspy.OutputField(desc="enumeration of key ideas in the ground truth")
    system_response_key_ideas: str = dspy.OutputField(desc="enumeration of key ideas in the system response")
    discussion: str = dspy.OutputField(desc="discussion of the overlap between ground truth and system response")
    recall: float = dspy.OutputField(desc="fraction (out of 1.0) of ground truth covered by the system response")
    precision: float = dspy.OutputField(desc="fraction (out of 1.0) of system response covered by the ground truth")


def f1_score(precision, recall):
    precision, recall = max(0.0, min(1.0, precision)), max(0.0, min(1.0, recall))
    return 0.0 if precision + recall == 0 else 2 * (precision * recall) / (precision + recall)


class SemanticF1(dspy.Module):
    def __init__(self, threshold=0.66, decompositional=False):
        self.threshold = threshold

        if decompositional:
            self.module = dspy.ChainOfThought(DecompositionalSemanticRecallPrecision)
        else:
            self.module = dspy.ChainOfThought(SemanticRecallPrecision)

    def forward(self, example, pred, trace=None):
        scores = self.module(question=example.question, ground_truth=example.response, system_response=pred.response)
        score = f1_score(scores.precision, scores.recall)

        return score if trace is None else score >= self.threshold

This class allows for both standard and decompositional semantic evaluation, providing flexibility in assessing the quality of responses. The forward method calculates the F1 score based on precision and recall, offering a comprehensive evaluation metric.

Practical Example: Evaluating a Tweet

To illustrate the creation of custom metrics, let's consider a practical example of evaluating a tweet. The goal is to assess whether a generated tweet answers a given question correctly, is engaging, and adheres to the character limit. Here's how you can implement such a metric:

class Assess(dspy.Signature):
    """Assess the quality of a tweet along the specified dimension."""

    assessed_text = dspy.InputField()
    assessment_question = dspy.InputField()
    assessment_answer: bool = dspy.OutputField()

def metric(gold, pred, trace=None):
    question, answer, tweet = gold.question, gold.answer, pred.output

    engaging = "Does the assessed text make for a self-contained, engaging tweet?"
    correct = f"The text should answer `{question}` with `{answer}`. Does the assessed text contain this answer?"

    correct =  dspy.Predict(Assess)(assessed_text=tweet, assessment_question=correct)
    engaging = dspy.Predict(Assess)(assessed_text=tweet, assessment_question=engaging)

    correct, engaging = [m.assessment_answer for m in [correct, engaging]]
    score = (correct + engaging) if correct and (len(tweet) <= 280) else 0

    if trace is not None: return score >= 2
    return score / 2.0

In this example, the Assess class defines the signature for automatic assessments, and the metric function evaluates the tweet based on correctness, engagement, and length. The function returns a score that reflects the quality of the tweet, providing a practical application of custom metrics.

Summary and Preparation for Practice

In this lesson, you learned how to create and use metrics in DSPy to evaluate the quality of system outputs. We covered basic metric functions, explored completeness and groundedness evaluation, and introduced semantic evaluation with F1 scores. Additionally, we walked through a practical example of evaluating a tweet using custom metrics. These skills are essential for assessing the performance of DSPy systems and will serve as a foundation for more advanced topics in the course. As you move on to the practice exercises, I encourage you to apply what you've learned and experiment with creating your own metrics in the CodeSignal IDE. This hands-on practice will reinforce your understanding and prepare you for the next steps in your DSPy journey.

Sign up
Join the 1M+ learners on CodeSignal
Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal