RAG & Agent Evaluation

# Implementation example
# Key operations: results display, RAG pipeline, evaluation logic
from ragas import evaluate
from ragas.metrics import (
 faithfulness,
 answer_relevancy,
 context_precision,
 context_recall,
)
from datasets import Dataset

# Prepare evaluation dataset with required columns
eval_data = {
 "question": [
 "What is the capital of France?",
 "How does photosynthesis work?",
 "What causes tides?",
 ],
 "answer": [
 "The capital of France is Paris, which is also its largest city.",
 "Photosynthesis converts CO2 and water into glucose using sunlight.",
 "Tides are caused by gravitational pull of the moon and sun.",
 ],
 "contexts": [
 ["Paris is the capital and most populous city of France."],
 ["Photosynthesis is a process by which plants convert light energy into chemical energy."],
 ["Ocean tides are caused by the gravitational forces of the moon."],
 ],
 "ground_truth": [
 "Paris is the capital of France.",
 "Photosynthesis converts light energy, CO2, and water into glucose and oxygen.",
 "Tides are primarily caused by the moon's gravitational pull on Earth's oceans.",
 ],
}

dataset = Dataset.from_dict(eval_data)

# Run RAGAS evaluation
results = evaluate(
 dataset,
 metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
)

print(results)
print("\nPer-question breakdown:")
df = results.to_pandas()
print(df.to_string(index=False))

# implement score_faithfulness
# Key operations: results display, prompt construction, evaluation logic
from openai import OpenAI
import json

client = OpenAI()

def score_faithfulness(question: str, answer: str, contexts: list[str]) -> dict:
 """Score faithfulness: is the answer grounded in the provided context?

 Uses an LLM judge to decompose the answer into claims and verify
 each claim against the context.
 """
 context_text = "\n\n".join(f"Context {i+1}: {c}" for i, c in enumerate(contexts))

 prompt = f"""Evaluate the faithfulness of an answer to the provided context.

QUESTION: {question}
CONTEXT:
{context_text}

ANSWER: {answer}

Instructions:
1. Decompose the answer into individual factual claims
2. For each claim, determine if it is SUPPORTED or NOT SUPPORTED by the context
3. Return JSON with:
 - "claims": list of {{"claim": str, "supported": bool, "evidence": str}}
 - "faithfulness_score": fraction of supported claims (0.0 to 1.0)"""

 response = client.chat.completions.create(
 model="gpt-4o",
 messages=[{"role": "user", "content": prompt}],
 response_format={"type": "json_object"},
 temperature=0,
 )
 return json.loads(response.choices[0].message.content)

# Example usage
result = score_faithfulness(
 question="What is the capital of France?",
 answer="Paris is the capital of France with a population of 2.1 million.",
 contexts=["Paris is the capital and most populous city of France."]
)
print(json.dumps(result, indent=2))

# Define ToolCall, AgentTrajectory; implement evaluate_agent_trajectory
# Key operations: agent orchestration, tool integration, evaluation logic
from dataclasses import dataclass
from typing import Optional

@dataclass
class ToolCall:
 """A single tool call in an agent trajectory."""
 tool_name: str
 parameters: dict
 result: Optional[str] = None
 is_correct_tool: Optional[bool] = None

@dataclass
class AgentTrajectory:
 """Complete record of an agent's execution."""
 task: str
 tool_calls: list[ToolCall]
 final_answer: str
 total_tokens: int = 0
 total_latency_ms: float = 0

def evaluate_agent_trajectory(
 trajectory: AgentTrajectory,
 ideal_tools: list[str],
 expected_answer: str,
 answer_checker=None
) -> dict:
 """Evaluate an agent trajectory against an ideal reference.

 Args:
 trajectory: The actual execution trajectory
 ideal_tools: Ordered list of expected tool names
 expected_answer: The ground truth answer
 answer_checker: Optional function for fuzzy answer matching
 """
 actual_tools = [tc.tool_name for tc in trajectory.tool_calls]

 # Task completion: did the agent get the right answer?
 if answer_checker:
 task_complete = answer_checker(trajectory.final_answer, expected_answer)
 else:
 task_complete = trajectory.final_answer.strip().lower() == expected_answer.strip().lower()

 # Tool accuracy: fraction of calls using correct tools
 correct_tools = sum(
 1 for t in actual_tools if t in ideal_tools
 )
 tool_accuracy = correct_tools / len(actual_tools) if actual_tools else 0

 # Trajectory efficiency: ideal steps / actual steps
 efficiency = min(len(ideal_tools) / len(actual_tools), 1.0) if actual_tools else 0

 # Redundancy: duplicate consecutive tool calls
 redundant = sum(
 1 for i in range(1, len(actual_tools))
 if actual_tools[i] == actual_tools[i - 1]
 )

 return {
 "task_completed": task_complete,
 "tool_accuracy": round(tool_accuracy, 3),
 "trajectory_efficiency": round(efficiency, 3),
 "num_steps": len(actual_tools),
 "ideal_steps": len(ideal_tools),
 "redundant_calls": redundant,
 "total_tokens": trajectory.total_tokens,
 }

# implement test_rag_with_deepeval
# Key operations: retrieval pipeline, RAG pipeline, evaluation logic
from deepeval import assert_test
from deepeval.test_case import LLMTestCase
from deepeval.metrics import (
 FaithfulnessMetric,
 AnswerRelevancyMetric,
 HallucinationMetric,
)

def test_rag_with_deepeval():
 """Example: Testing RAG output quality with DeepEval."""
 test_case = LLMTestCase(
 input="What are the benefits of solar energy?",
 actual_output="Solar energy is renewable, reduces electricity bills, "
 "and has low maintenance costs.",
 retrieval_context=[
 "Solar energy is a renewable source of power that reduces "
 "dependence on fossil fuels.",
 "Solar panels require minimal maintenance and can reduce "
 "electricity bills by up to 50%.",
 ],
 )

 # Define metrics with thresholds
 faithfulness = FaithfulnessMetric(threshold=0.7, model="gpt-4o")
 relevancy = AnswerRelevancyMetric(threshold=0.7, model="gpt-4o")
 hallucination = HallucinationMetric(threshold=0.5, model="gpt-4o")

 # Assert all metrics pass (integrates with pytest)
 assert_test(test_case, [faithfulness, relevancy, hallucination])

# Run with: pytest test_rag.py -v