Tutorial 10: Evaluation Metrics in LLMs

# Cohen's Kappa example
# Scenario: Two human raters evaluate 10 LLM responses as Helpful (1) or Not Helpful (0)

from sklearn.metrics import cohen_kappa_score
import numpy as np

# Rater A and Rater B scores for 10 LLM responses
rater_A = [1, 1, 0, 1, 0, 1, 1, 0, 0, 1]
rater_B = [1, 1, 0, 1, 1, 1, 0, 0, 0, 1]

kappa = cohen_kappa_score(rater_A, rater_B)
print(f"Cohen's Kappa: {kappa:.3f}")
print(f"Raw agreement: {np.mean(np.array(rater_A) == np.array(rater_B)):.1%}")

# Fleiss' Kappa example
# Scenario: 3 raters evaluate 5 LLM responses on a 3-point scale (0=Bad, 1=Ok, 2=Good)
# Each row = one response; each column = count of raters choosing that category

# !pip install statsmodels
from statsmodels.stats.inter_rater import fleiss_kappa, aggregate_raters

# Raw ratings: rows = items, columns = rater scores
ratings = np.array([
    [2, 2, 2],   # Response 1: all raters say Good
    [0, 1, 0],   # Response 2: two say Bad, one says Ok
    [1, 1, 2],   # Response 3: two say Ok, one says Good
    [0, 0, 1],   # Response 4: two say Bad, one says Ok
    [2, 1, 2],   # Response 5: two say Good, one says Ok
])

# Aggregate into category counts
table, categories = aggregate_raters(ratings)
fk = fleiss_kappa(table)
print(f"Fleiss' Kappa: {fk:.3f}")

# BLEU Score Example
# !pip install nltk
from nltk.translate.bleu_score import sentence_bleu, corpus_bleu, SmoothingFunction
import nltk
nltk.download('punkt', quiet=True)

# Example: Evaluating a machine translation output
reference = "the cat sat on the mat".split()
hypothesis_good = "the cat sat on the mat".split()       # Perfect match
hypothesis_ok   = "a cat was sitting on a mat".split()   # Partial match
hypothesis_bad  = "dog runs quickly through park".split() # No match

smoother = SmoothingFunction().method1

for name, hyp in [("Perfect", hypothesis_good), ("Partial", hypothesis_ok), ("Bad", hypothesis_bad)]:
    score = sentence_bleu([reference], hyp, smoothing_function=smoother)
    print(f"BLEU ({name:7s}): {score:.4f}  |  hypothesis: '{' '.join(hyp)}'")

# ROUGE Score Example
# !pip install rouge-score
from rouge_score import rouge_scorer

scorer = rouge_scorer.RougeScorer(['rouge1', 'rouge2', 'rougeL'], use_stemmer=True)

# Summarization example
reference_summary = (
    "The researchers developed a new language model that achieves state-of-the-art "
    "performance on multiple benchmarks using a novel training approach."
)

generated_summary = (
    "Scientists created an improved language model with record-breaking benchmark "
    "results through a new training method."
)

scores = scorer.score(reference_summary, generated_summary)

print("ROUGE Scores (Precision / Recall / F1)")
print("-" * 45)
for metric, score in scores.items():
    print(f"{metric.upper():10s}: P={score.precision:.3f}  R={score.recall:.3f}  F1={score.fmeasure:.3f}")

# METEOR Score Example
# pip install nltk
import nltk
nltk.download('wordnet', quiet=True)
nltk.download('punkt_tab', quiet=True)
from nltk.translate.meteor_score import meteor_score

reference = "the cat sat on the mat"

examples = [
    ("the cat sat on the mat",   "Perfect match"),
    ("a cat was sitting on a mat", "Synonym + morphology"),
    ("the mat sat on the cat",    "Same words, wrong order"),
    ("dog runs quickly in park",  "Completely different"),
]

print(f"Reference: '{reference}'\n")
print(f"{'Hypothesis':<35} {'METEOR':>7}  Description")
print("-" * 70)
for hyp, desc in examples:
    score = meteor_score([reference.split()], hyp.split())
    print(f"'{hyp}'")
    print(f"  → METEOR: {score:.4f}  ({desc})\n")

from dotenv import load_dotenv
import os

# Load environment variables from .env file
load_dotenv()

# Access the environment variables
os.environ['LANGCHAIN_TRACING_V2'] = os.getenv('LANGSMITH_TRACING')
os.environ['LANGCHAIN_ENDPOINT'] = 'https://api.smith.langchain.com'
os.environ['LANGCHAIN_API_KEY'] = os.getenv('LANGSMITH_API_KEY')
os.environ['GOOGLE_API_KEY'] = os.getenv('GOOGLE_API_KEY')

from langchain_google_genai import GoogleGenerativeAIEmbeddings, ChatGoogleGenerativeAI

embeddings = GoogleGenerativeAIEmbeddings(model="models/gemini-embedding-001")
llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash-lite")

from dotenv import load_dotenv
import os
from pydantic import BaseModel, Field
from typing import List, Literal, Optional

# Load environment variables from .env file
load_dotenv()
os.environ['LANGCHAIN_TRACING_V2'] = os.getenv('LANGSMITH_TRACING')
os.environ['LANGCHAIN_ENDPOINT'] = 'https://api.smith.langchain.com'
os.environ['LANGCHAIN_API_KEY'] = os.getenv('LANGSMITH_API_KEY')
os.environ['GOOGLE_API_KEY'] = os.getenv('GOOGLE_API_KEY')

from google import genai

# ── Pydantic schemas for structured judge output ────────────────────────────

class CriterionScore(BaseModel):
    criterion: str = Field(description="The evaluation criterion being scored.")
    score: int = Field(description="Score from 1 to 5.")
    justification: str = Field(description="Brief justification for the score.")

class JudgementResult(BaseModel):
    overall_score: float = Field(description="Weighted overall score from 1.0 to 5.0.")
    verdict: Literal["excellent", "good", "acceptable", "poor", "fail"] = Field(
        description="Final categorical verdict."
    )
    criterion_scores: List[CriterionScore] = Field(
        description="Per-criterion breakdown of the evaluation."
    )
    strengths: List[str] = Field(description="Key strengths of the response.")
    weaknesses: List[str] = Field(description="Key weaknesses or areas for improvement.")
    reasoning: str = Field(description="Overall reasoning for the judgement.")


# ── Helper: call Gemini and parse structured output ─────────────────────────

client = genai.Client()

def judge_response(
    original_question: str,
    candidate_response: str,
    reference_answer: Optional[str] = None,
) -> JudgementResult:
    """Use Gemini as a judge with structured output."""

    ref_section = (
        f"\n\nReference Answer (ground truth):\n{reference_answer}"
        if reference_answer
        else "\n\n(No reference answer provided — judge on general quality.)"
    )

    prompt = f"""You are an expert evaluator assessing the quality of an AI-generated response.

Question asked:
{original_question}

Candidate Response to evaluate:
{candidate_response}
{ref_section}

Evaluate the response on these criteria (score each 1–5):
1. Accuracy       – Is the information correct and factually grounded?
2. Completeness   – Does it fully address the question?
3. Clarity        – Is it well-structured and easy to understand?
4. Conciseness    – Is it appropriately brief without omitting key content?
5. Helpfulness    – Would this response genuinely help the user?

Return a structured JSON judgement following the schema exactly.
"""

    response = client.models.generate_content(
        model="gemini-2.5-flash-lite",
        contents=prompt,
        config={
            "response_mime_type": "application/json",
            "response_json_schema": JudgementResult.model_json_schema(),
        },
    )

    return JudgementResult.model_validate_json(response.text)

# ── Example 1: Judge with a reference answer (RAG / QA scenario) ────────────

question_1 = "What causes the northern lights?"

candidate_1 = """The northern lights, or aurora borealis, are caused by collisions between
electrically charged particles from the sun and gases in Earth's atmosphere.
These collisions produce colourful lights in the sky, most visible near the poles."""

reference_1 = """Aurora borealis occurs when solar wind particles (electrons and protons)
are guided by Earth's magnetic field into the upper atmosphere, where they excite
oxygen and nitrogen atoms. Oxygen emits green and red light; nitrogen emits blue and purple."""

result_1 = judge_response(question_1, candidate_1, reference_answer=reference_1)

print("=" * 60)
print("EXAMPLE 1 — With Reference Answer")
print("=" * 60)
print(f"Verdict : {result_1.verdict.upper()}  (overall: {result_1.overall_score}/5)")
print(f"Reasoning: {result_1.reasoning}\n")
for cs in result_1.criterion_scores:
    print(f"  [{cs.score}/5] {cs.criterion}: {cs.justification}")
print(f"\nStrengths : {result_1.strengths}")
print(f"Weaknesses: {result_1.weaknesses}")

# ── Example 2: Judge without a reference answer (open-ended) ────────────────

question_2 = "Explain the bias-variance tradeoff in machine learning."

candidate_2 = """Bias is when your model is too simple and underfits. Variance is when
it's too complex and overfits. You want to find the sweet spot in between."""

result_2 = judge_response(question_2, candidate_2)

print("\n" + "=" * 60)
print("EXAMPLE 2 — No Reference Answer (open-ended)")
print("=" * 60)
print(f"Verdict : {result_2.verdict.upper()}  (overall: {result_2.overall_score}/5)")
print(f"Reasoning: {result_2.reasoning}\n")
for cs in result_2.criterion_scores:
    print(f"  [{cs.score}/5] {cs.criterion}: {cs.justification}")
print(f"\nStrengths : {result_2.strengths}")
print(f"Weaknesses: {result_2.weaknesses}")

# ── Example 3: Pairwise comparison (A/B judge) ──────────────────────────────

class PairwiseResult(BaseModel):
    winner: Literal["A", "B", "tie"] = Field(description="Which response is better.")
    confidence: Literal["high", "medium", "low"]
    reasoning: str = Field(description="Why this response was chosen.")
    response_a_score: float = Field(description="Score for response A (1–5).")
    response_b_score: float = Field(description="Score for response B (1–5).")

def judge_pairwise(question: str, response_a: str, response_b: str) -> PairwiseResult:
    """Compare two candidate responses head-to-head."""

    prompt = f"""You are an expert evaluator doing a pairwise comparison of two AI responses.

Question: {question}

Response A:
{response_a}

Response B:
{response_b}

Decide which response is better overall, or declare a tie.
Score each response individually (1–5) and explain your reasoning.
"""

    response = client.models.generate_content(
        model="gemini-2.5-flash-lite",
        contents=prompt,
        config={
            "response_mime_type": "application/json",
            "response_json_schema": PairwiseResult.model_json_schema(),
        },
    )

    return PairwiseResult.model_validate_json(response.text)


question_3 = "How does gradient descent work?"

response_a = """Gradient descent minimizes a loss function by iteratively updating model
parameters in the direction opposite to the gradient. The learning rate controls step size.
Variants include SGD, mini-batch GD, Adam, and RMSProp."""

response_b = """It's an optimization algorithm. You compute gradients and update weights.
If the learning rate is too high it diverges, too low it's slow."""

pairwise = judge_pairwise(question_3, response_a, response_b)

print("\n" + "=" * 60)
print("EXAMPLE 3 — Pairwise A/B Comparison")
print("=" * 60)
print(f"Winner    : Response {pairwise.winner.upper()} ({pairwise.confidence} confidence)")
print(f"A score   : {pairwise.response_a_score}/5")
print(f"B score   : {pairwise.response_b_score}/5")
print(f"Reasoning : {pairwise.reasoning}")

# Claim decomposition and verification pipeline (conceptual)
# In production, each step would use an LLM or search API

def extract_claims(text):
    """Step 1: Decompose text into individual factual claims."""
    # In practice: prompt an LLM to extract atomic claims
    # Example simulation:
    return [
        "The Eiffel Tower was built in 1889.",
        "The Eiffel Tower is located in London.",   # WRONG
        "The Eiffel Tower is 330 metres tall.",
        "The Eiffel Tower was designed by Gustave Eiffel.",
    ]

def verify_claim(claim, knowledge_base):
    """Step 2: Verify each claim against a knowledge source."""
    # In practice: RAG lookup or search API
    return knowledge_base.get(claim, None)

knowledge_base = {
    "The Eiffel Tower was built in 1889.": True,
    "The Eiffel Tower is located in London.": False,   # Paris, not London
    "The Eiffel Tower is 330 metres tall.": True,
    "The Eiffel Tower was designed by Gustave Eiffel.": True,
}

generated_text = "The Eiffel Tower, built in 1889 and located in London, stands 330 metres tall and was designed by Gustave Eiffel."

claims = extract_claims(generated_text)
results = [(claim, verify_claim(claim, knowledge_base)) for claim in claims]

print("Claim Verification Results:")
print("-" * 60)
for claim, is_true in results:
    status = "✓ TRUE" if is_true else "✗ FALSE" if is_true is False else "? UNKNOWN"
    print(f"{status}: {claim}")

true_count = sum(1 for _, v in results if v is True)
total = len(results)
print(f"\nFactuality score: {true_count}/{total} = {true_count/total:.0%}")