RAG METRICS

AGENT METRICS

GENERAL LLM METRICS

When you can't write deterministic evaluation rules (most LLM outputs), use a capable LLM as the judge. The key is calibration — your judge must agree with human raters on a validation set.

import anthropic
from pydantic import BaseModel
import instructor

judge_client = instructor.from_anthropic(anthropic.Anthropic())

class JudgeVerdict(BaseModel):
    score:      float   # 0.0 to 1.0
    reasoning:  str
    passed:     bool    # True if score >= threshold

# ── Faithfulness judge ────────────────────────────────
FAITHFULNESS_JUDGE = """You are an expert evaluator. Determine whether every factual
claim in the ANSWER is directly supported by the CONTEXT.

Score 1.0: All claims are explicitly stated in the context.
Score 0.5: Most claims supported, some extrapolation.
Score 0.0: Major claims not in context — hallucination present.

CONTEXT:
{context}

ANSWER:
{answer}"""

def judge_faithfulness(context: str, answer: str,
                        threshold: float = 0.7) -> JudgeVerdict:
    result = judge_client.messages.create(
        model="claude-3-5-sonnet-20241022",   # strong model for judging
        max_tokens=300,
        temperature=0.0,
        messages=[{"role": "user",
                   "content": FAITHFULNESS_JUDGE.format(context=context, answer=answer)}],
        response_model=JudgeVerdict
    )
    result.passed = result.score >= threshold
    return result

# ── Task success judge ────────────────────────────────
TASK_SUCCESS_JUDGE = """Did the AI agent successfully complete the following task?

ORIGINAL TASK: {task}
AGENT'S OUTPUT: {output}

Score 1.0: Task fully completed — all requirements met.
Score 0.5: Task partially completed — some requirements missing.
Score 0.0: Task failed — output does not address the task."""

def judge_task_success(task: str, output: str) -> JudgeVerdict:
    return judge_client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=200, temperature=0.0,
        messages=[{"role": "user",
                   "content": TASK_SUCCESS_JUDGE.format(task=task, output=output)}],
        response_model=JudgeVerdict
    )

# Build a ground truth dataset for RAG evaluation
# Each test case: question + expected answer + expected source
RAG_TEST_SET = [
    {
        "question": "How does DPDK mempool initialisation work?",
        "ground_truth": "DPDK mempool uses rte_mempool_create() with a fixed pool of memory objects pre-allocated at startup.",
        "expected_source": "dpdk-guide-mempool.pdf",
    },
    # ... 20+ test cases
]

# Run full eval loop
async def evaluate_rag_pipeline(pipeline, test_set: list) -> dict:
    scores = {"faithfulness": [], "relevancy": [], "hit_rate": [],
              "cost_usd": [], "latency_ms": []}

    for case in test_set:
        import time
        t_start = time.perf_counter()
        result  = await pipeline.query(case["question"])
        latency = (time.perf_counter() - t_start) * 1000

        # Metric 1: Faithfulness
        faith = judge_faithfulness(
            context=" ".join(s["text"] for s in result["sources"]),
            answer=result["answer"]
        )
        scores["faithfulness"].append(faith.score)

        # Metric 2: Answer relevancy (does answer address the question?)
        relevancy = judge_answer_relevancy(case["question"], result["answer"])
        scores["relevancy"].append(relevancy.score)

        # Metric 3: Source hit rate
        expected = case["expected_source"]
        hit = any(expected in s.get("source", "") for s in result["sources"])
        scores["hit_rate"].append(float(hit))

        scores["latency_ms"].append(latency)

    def avg(lst): return round(sum(lst) / len(lst), 3) if lst else 0
    return {k: avg(v) for k, v in scores.items()}

# Agent evaluation is harder than RAG eval because:
# 1. The "right answer" may not be unique
# 2. The path matters, not just the destination
# 3. Tool calls have side effects that are hard to undo

@dataclass
class AgentTestCase:
    task:              str
    expected_outcome:  str                # what a successful completion looks like
    required_tools:    list[str] = None   # tools that MUST be called
    forbidden_tools:   list[str] = None   # tools that must NOT be called
    max_turns:         int = 10
    max_cost_usd:      float = 0.50

AGENT_TEST_SET = [
    AgentTestCase(
        task="Find the square root of 1764 and the current time",
        expected_outcome="Answer mentions 42 and current time",
        required_tools=["calculate", "get_current_time"],
        max_turns=5
    ),
    AgentTestCase(
        task="Search for DPDK documentation on hugepages",
        expected_outcome="Returns information about hugepage configuration",
        required_tools=["search_web"],
        forbidden_tools=["send_email"],   # should not email anyone
    ),
]

class AgentEvaluator:
    def evaluate(self, agent_fn, test_case: AgentTestCase) -> dict:
        result = agent_fn(test_case.task)
        tools_called = result.get("tools_called", [])
        output       = result.get("answer", "")
        turns        = result.get("turns_used", 0)
        cost         = result.get("cost_usd", 0)

        # Task success — LLM judge
        success = judge_task_success(test_case.task, output)

        # Required tools coverage
        tool_coverage = 1.0
        if test_case.required_tools:
            called_set  = set(tools_called)
            required    = set(test_case.required_tools)
            tool_coverage = len(called_set & required) / len(required)

        # Forbidden tools check
        forbidden_used = []
        if test_case.forbidden_tools:
            forbidden_used = [t for t in tools_called if t in test_case.forbidden_tools]

        # Efficiency: did it use more turns than needed?
        efficiency = min(1.0, (test_case.max_turns - turns) / test_case.max_turns + 0.5)

        return {
            "task_success":   success.score,
            "task_passed":    success.passed,
            "tool_coverage":  tool_coverage,
            "forbidden_used": forbidden_used,
            "turns_used":     turns,
            "cost_usd":       cost,
            "efficiency":     efficiency,
            "judge_reasoning": success.reasoning,
        }

    def evaluate_batch(self, agent_fn, test_set) -> dict:
        results   = [self.evaluate(agent_fn, tc) for tc in test_set]
        successes = [r["task_success"] for r in results]
        return {
            "task_success_rate": sum(r["task_passed"] for r in results) / len(results),
            "avg_success_score": sum(successes) / len(successes),
            "avg_turns":         sum(r["turns_used"] for r in results) / len(results),
            "avg_cost_usd":      sum(r["cost_usd"] for r in results) / len(results),
            "forbidden_violations": sum(1 for r in results if r["forbidden_used"]),
            "n":                 len(results),
        }

pip install deepeval

from deepeval import evaluate
from deepeval.metrics import (
    AnswerRelevancyMetric, FaithfulnessMetric,
    ContextualPrecisionMetric, ContextualRecallMetric,
    HallucinationMetric, ToxicityMetric,
)
from deepeval.test_case import LLMTestCase

# Define metrics
metrics = [
    AnswerRelevancyMetric(threshold=0.7, model="gpt-4o"),
    FaithfulnessMetric(threshold=0.7, model="gpt-4o"),
    ContextualPrecisionMetric(threshold=0.7, model="gpt-4o"),
    ContextualRecallMetric(threshold=0.7, model="gpt-4o"),
]

# Create a test case
test_case = LLMTestCase(
    input="How does DPDK mempool work?",
    actual_output="DPDK mempool pre-allocates a fixed pool of memory objects...",
    expected_output="rte_mempool_create() creates a fixed-size pool...",   # optional
    retrieval_context=["The DPDK mempool library provides an API to allocate..."]
)

# Run evaluation
results = evaluate([test_case], metrics)

# Use in pytest for CI/CD regression testing
from deepeval import assert_test
import pytest

@pytest.mark.parametrize("test_case", my_test_cases)
def test_rag_quality(test_case):
    assert_test(test_case, metrics)

pip install langsmith

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"]      = os.environ["LANGSMITH_API_KEY"]
os.environ["LANGCHAIN_PROJECT"]       = "my-rag-project"

# All LangChain calls now auto-trace to LangSmith
# Go to smith.langchain.com → see every run

# ── Manual tracing (without LangChain) ───────────────
from langsmith import Client, traceable

ls_client = Client()

@traceable(name="rag_query", run_type="chain")
def traced_rag_query(question: str) -> dict:
    # Your RAG pipeline here — every call is auto-logged
    result = rag_query(question)
    return result

# ── Dataset-based evaluation ──────────────────────────
# Create a dataset in LangSmith
dataset = ls_client.create_dataset("rag-eval-set")
ls_client.create_examples(
    inputs=[{"question": t["question"]} for t in RAG_TEST_SET],
    outputs=[{"answer": t["ground_truth"]} for t in RAG_TEST_SET],
    dataset_id=dataset.id
)

# Define evaluator function
def faithfulness_evaluator(run, example) -> dict:
    verdict = judge_faithfulness(
        context=run.outputs.get("context", ""),
        answer=run.outputs.get("answer", "")
    )
    return {"key": "faithfulness", "score": verdict.score,
            "comment": verdict.reasoning}

# Run evaluation against the dataset
from langsmith.evaluation import evaluate as ls_evaluate

results = ls_evaluate(
    traced_rag_query,
    data="rag-eval-set",
    evaluators=[faithfulness_evaluator],
    experiment_prefix="v2-reranked"
)
# Results visible in LangSmith UI with charts, per-example scores, diffs