Benchmark: Ragas 0.1 vs. LangSmith 2.0: RAG Evaluation Speed for 1k Queries

import os
import time
import logging
from typing import List, Dict
import pandas as pd
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevance, context_relevance
from ragas.dataset import Dataset
from datasets import load_dataset

# Configure logging for error tracking
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

def load_rag_eval_dataset(num_samples: int = 1000) -> Dataset:
    """
    Load SQuAD 2.0 RAG eval dataset, truncated to num_samples.
    Returns Ragas-compatible Dataset object.
    """
    try:
        # Load base SQuAD 2.0 dataset
        squad = load_dataset("squad_v2", split="validation")
        # Sample first num_samples to ensure consistent benchmark
        sampled = squad.select(range(min(num_samples, len(squad))))

        # Format into Ragas required schema: question, answer, contexts, ground_truth
        eval_data = {
            "question": [item["question"] for item in sampled],
            "answer": [item["answers"]["text"][0] if item["answers"]["text"] else "No answer" for item in sampled],
            "contexts": [[item["context"]] for item in sampled],  # Wrap context in list per Ragas schema
            "ground_truth": [item["answers"]["text"][0] if item["answers"]["text"] else "No answer" for item in sampled]
        }
        return Dataset.from_dict(eval_data)
    except Exception as e:
        logger.error(f"Failed to load dataset: {str(e)}")
        raise

def run_ragas_benchmark(dataset: Dataset) -> Dict:
    """
    Run Ragas 0.1 evaluation on 1k queries, return timing and result metrics.
    """
    metrics = [faithfulness, answer_relevance, context_relevance]
    start_time = time.perf_counter()

    try:
        # Run async batch evaluation (Ragas 0.1 native support)
        result = evaluate(dataset=dataset, metrics=metrics, raise_exceptions=False)
        end_time = time.perf_counter()
        elapsed = end_time - start_time

        # Extract aggregate metrics
        return {
            "elapsed_seconds": round(elapsed, 2),
            "faithfulness_avg": round(result["faithfulness"], 4),
            "answer_relevance_avg": round(result["answer_relevance"], 4),
            "context_relevance_avg": round(result["context_relevance"], 4),
            "total_queries": len(dataset)
        }
    except Exception as e:
        logger.error(f"Ragas evaluation failed: {str(e)}")
        raise

if __name__ == "__main__":
    # Benchmark configuration
    NUM_QUERIES = 1000
    logger.info(f"Starting Ragas 0.1 benchmark for {NUM_QUERIES} queries")

    # Load dataset
    eval_dataset = load_rag_eval_dataset(NUM_QUERIES)
    logger.info(f"Loaded dataset with {len(eval_dataset)} samples")

    # Run benchmark
    benchmark_results = run_ragas_benchmark(eval_dataset)

    # Log results
    logger.info(f"Ragas 0.1 Benchmark Results:")
    logger.info(f"Total Queries: {benchmark_results['total_queries']}")
    logger.info(f"Elapsed Time: {benchmark_results['elapsed_seconds']} seconds")
    logger.info(f"Faithfulness Avg: {benchmark_results['faithfulness_avg']}")
    logger.info(f"Answer Relevance Avg: {benchmark_results['answer_relevance_avg']}")
    logger.info(f"Context Relevance Avg: {benchmark_results['context_relevance_avg']}")

    # Save results to CSV for comparison
    pd.DataFrame([benchmark_results]).to_csv("ragas_0_1_benchmark.csv", index=False)

import os
import time
import logging
from typing import List, Dict
import pandas as pd
from langsmith import Client
from langsmith.evaluation import evaluate as langsmith_evaluate
from langsmith.schemas import Example, Run
import requests
from datasets import load_dataset

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# LangSmith API configuration (set via environment variables)
LANGSMITH_API_KEY = os.getenv("LANGSMITH_API_KEY")
LANGSMITH_PROJECT = os.getenv("LANGSMITH_PROJECT", "rag-eval-benchmark")
if not LANGSMITH_API_KEY:
    raise ValueError("LANGSMITH_API_KEY environment variable is required")

def load_langsmith_eval_dataset(num_samples: int = 1000) -> List[Example]:
    """
    Load SQuAD 2.0 dataset and format into LangSmith Example objects.
    """
    try:
        squad = load_dataset("squad_v2", split="validation")
        sampled = squad.select(range(min(num_samples, len(squad))))

        examples = []
        for item in sampled:
            # LangSmith example schema for RAG eval
            example = Example(
                inputs={
                    "question": item["question"],
                    "context": item["context"]
                },
                outputs={
                    "ground_truth": item["answers"]["text"][0] if item["answers"]["text"] else "No answer"
                }
            )
            examples.append(example)
        return examples
    except Exception as e:
        logger.error(f"Failed to load dataset for LangSmith: {str(e)}")
        raise

def ragas_metric_wrapper(run: Run, example: Example) -> Dict:
    """
    Wrapper to use Ragas metrics within LangSmith 2.0 evaluation pipeline.
    LangSmith 2.0 does not natively support all Ragas metrics, so we wrap them.
    """
    try:
        from ragas.metrics import faithfulness, answer_relevance, context_relevance
        from ragas import evaluate
        from ragas.dataset import Dataset

        # Format single sample into Ragas dataset
        single_data = {
            "question": [example.inputs["question"]],
            "answer": [run.outputs.get("answer", "")],
            "contexts": [example.inputs["context"]],
            "ground_truth": [example.outputs["ground_truth"]]
        }
        ragas_dataset = Dataset.from_dict(single_data)

        # Evaluate single sample
        result = evaluate(dataset=ragas_dataset, metrics=[faithfulness, answer_relevance, context_relevance])
        return {
            "faithfulness": result["faithfulness"],
            "answer_relevance": result["answer_relevance"],
            "context_relevance": result["context_relevance"]
        }
    except Exception as e:
        logger.error(f"Metric wrapper failed: {str(e)}")
        return {"faithfulness": 0.0, "answer_relevance": 0.0, "context_relevance": 0.0}

def run_langsmith_benchmark(examples: List[Example]) -> Dict:
    """
    Run LangSmith 2.0 evaluation on 1k queries, return timing and metrics.
    """
    client = Client(api_key=LANGSMITH_API_KEY)
    start_time = time.perf_counter()

    try:
        # Create evaluation project
        client.create_project(project_name=LANGSMITH_PROJECT)

        # Run evaluation (LangSmith 2.0 sequential processing for 1k samples)
        eval_results = langsmith_evaluate(
            examples=examples,
            evaluators=[ragas_metric_wrapper],
            project_name=LANGSMITH_PROJECT,
            max_concurrency=4  # LangSmith 2.0 max concurrency for eval
        )

        end_time = time.perf_counter()
        elapsed = end_time - start_time

        # Aggregate results
        faith_scores = [r.get("faithfulness", 0.0) for r in eval_results]
        ans_rel_scores = [r.get("answer_relevance", 0.0) for r in eval_results]
        ctx_rel_scores = [r.get("context_relevance", 0.0) for r in eval_results]

        return {
            "elapsed_seconds": round(elapsed, 2),
            "faithfulness_avg": round(sum(faith_scores)/len(faith_scores), 4),
            "answer_relevance_avg": round(sum(ans_rel_scores)/len(ans_rel_scores), 4),
            "context_relevance_avg": round(sum(ctx_rel_scores)/len(ctx_rel_scores), 4),
            "total_queries": len(examples)
        }
    except Exception as e:
        logger.error(f"LangSmith evaluation failed: {str(e)}")
        raise

if __name__ == "__main__":
    NUM_QUERIES = 1000
    logger.info(f"Starting LangSmith 2.0 benchmark for {NUM_QUERIES} queries")

    # Load dataset
    eval_examples = load_langsmith_eval_dataset(NUM_QUERIES)
    logger.info(f"Loaded {len(eval_examples)} examples for LangSmith")

    # Run benchmark
    benchmark_results = run_langsmith_benchmark(eval_examples)

    # Log results
    logger.info(f"LangSmith 2.0 Benchmark Results:")
    logger.info(f"Total Queries: {benchmark_results['total_queries']}")
    logger.info(f"Elapsed Time: {benchmark_results['elapsed_seconds']} seconds")
    logger.info(f"Faithfulness Avg: {benchmark_results['faithfulness_avg']}")
    logger.info(f"Answer Relevance Avg: {benchmark_results['answer_relevance_avg']}")
    logger.info(f"Context Relevance Avg: {benchmark_results['context_relevance_avg']}")

    # Save results
    pd.DataFrame([benchmark_results]).to_csv("langsmith_2_0_benchmark.csv", index=False)

import os
import time
import logging
import pandas as pd
from typing import Dict
from datasets import load_dataset
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevance, context_relevance
from ragas.dataset import Dataset
from langsmith import Client
from langsmith.evaluation import evaluate as langsmith_evaluate
from langsmith.schemas import Example

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

def load_ragas_dataset(num_samples: int = 1000) -> Dataset:
    squad = load_dataset("squad_v2", split="validation").select(range(num_samples))
    return Dataset.from_dict({
        "question": [item["question"] for item in squad],
        "answer": [item["answers"]["text"][0] if item["answers"]["text"] else "No answer" for item in squad],
        "contexts": [[item["context"]] for item in squad],
        "ground_truth": [item["answers"]["text"][0] if item["answers"]["text"] else "No answer" for item in squad]
    })

def load_langsmith_examples(num_samples: int = 1000) -> List[Example]:
    squad = load_dataset("squad_v2", split="validation").select(range(num_samples))
    return [Example(
        inputs={"question": item["question"], "context": item["context"]},
        outputs={"ground_truth": item["answers"]["text"][0] if item["answers"]["text"] else "No answer"}
    ) for item in squad]

def run_ragas_eval(dataset: Dataset) -> Dict:
    start = time.perf_counter()
    result = evaluate(dataset=dataset, metrics=[faithfulness, answer_relevance, context_relevance], raise_exceptions=False)
    elapsed = time.perf_counter() - start
    return {
        "tool": "Ragas 0.1",
        "elapsed_seconds": round(elapsed, 2),
        "faithfulness": round(result["faithfulness"], 4),
        "answer_relevance": round(result["answer_relevance"], 4),
        "context_relevance": round(result["context_relevance"], 4)
    }

def run_langsmith_eval(examples: List[Example]) -> Dict:
    client = Client(api_key=os.getenv("LANGSMITH_API_KEY"))
    start = time.perf_counter()
    def metric_wrapper(run: Run, example: Example) -> Dict:
        single_data = {
            "question": [example.inputs["question"]],
            "answer": [run.outputs.get("answer", "")],
            "contexts": [example.inputs["context"]],
            "ground_truth": [example.outputs["ground_truth"]]
        }
        ragas_dataset = Dataset.from_dict(single_data)
        res = evaluate(dataset=ragas_dataset, metrics=[faithfulness, answer_relevance, context_relevance])
        return {"faithfulness": res["faithfulness"], "answer_relevance": res["answer_relevance"], "context_relevance": res["context_relevance"]}
    results = langsmith_evaluate(examples=examples, evaluators=[metric_wrapper], project_name="benchmark", max_concurrency=4)
    elapsed = time.perf_counter() - start
    faith = [r["faithfulness"] for r in results]
    ans_rel = [r["answer_relevance"] for r in results]
    ctx_rel = [r["context_relevance"] for r in results]
    return {
        "tool": "LangSmith 2.0",
        "elapsed_seconds": round(elapsed, 2),
        "faithfulness": round(sum(faith)/len(faith), 4),
        "answer_relevance": round(sum(ans_rel)/len(ans_rel), 4),
        "context_relevance": round(sum(ctx_rel)/len(ctx_rel), 4)
    }

if __name__ == "__main__":
    logger.info("Running comparative benchmark for 1k queries")
    ragas_dataset = load_ragas_dataset(1000)
    langsmith_examples = load_langsmith_examples(1000)

    ragas_res = run_ragas_eval(ragas_dataset)
    langsmith_res = run_langsmith_eval(langsmith_examples)

    combined = pd.DataFrame([ragas_res, langsmith_res])
    combined.to_csv("comparative_benchmark.csv", index=False)
    logger.info(f"Results saved. Ragas: {ragas_res['elapsed_seconds']}s, LangSmith: {langsmith_res['elapsed_seconds']}s")

from ragas import evaluate
from ragas.metrics import faithfulness

# Optimized batch size for 8-core CPU
result = evaluate(
    dataset=your_dataset,
    metrics=[faithfulness],
    batch_size=8,  # Match CPU core count
    raise_exceptions=False
)

from langsmith import Client

client = Client(api_key="your-api-key")

def get_cached_eval(question: str, context: str, project: str = "rag-eval"):
    # Check for existing runs with matching inputs
    runs = client.list_runs(
        project_name=project,
        filter=f"inputs.question == '{question}' AND inputs.context == '{context}'",
        limit=1
    )
    return runs[0] if runs else None

from ragas.metrics import BaseMetric
from ragas.dataset import Dataset

class ClinicalTerminologyMetric(BaseMetric):
    name = "clinical_terminology"

    def __init__(self, icd_10_codes: list):
        self.icd_10_codes = icd_10_codes

    def score(self, row: dict) -> float:
        # Check if answer contains valid ICD-10 codes
        answer = row["answer"]
        matches = [code for code in self.icd_10_codes if code in answer]
        return len(matches) / len(self.icd_10_codes) if self.icd_10_codes else 0.0