Ragas — Testing & CI¶

import asyncio

import pytest
from openai import AsyncOpenAI
from ragas.llms import InstructorBaseRagasLLM, llm_factory
from ragas.metrics.collections import Faithfulness
from ragas.metrics.collections.faithfulness.util import (
    NLIStatementOutput, StatementFaithfulnessAnswer, StatementGeneratorOutput,
)

from evals.fake_judge import FakeJudge


class ScriptedJudge(InstructorBaseRagasLLM):
    """Returns prepared Pydantic objects keyed by response-model name and records the prompts."""

    def __init__(self, script):
        self.script, self.prompts = script, []

    def generate(self, prompt, response_model):
        raise NotImplementedError("metrics call agenerate")

    async def agenerate(self, prompt, response_model):
        self.prompts.append(prompt)
        return self.script[response_model.__name__]


def test_faithfulness_is_fraction_of_supported_statements():
    flags = [1, 1, 1, 0]
    judge = ScriptedJudge({
        "StatementGeneratorOutput": StatementGeneratorOutput(statements=["a", "b", "c", "d"]),
        "NLIStatementOutput": NLIStatementOutput(statements=[
            StatementFaithfulnessAnswer(statement=s, reason="r", verdict=v) for s, v in zip("abcd", flags)]),
    })
    result = Faithfulness(llm=judge).score(
        user_input="q", response="Paris is in France.", retrieved_contexts=["The capital is Paris."])
    assert result.value == pytest.approx(0.75)
    assert "Paris is in France." in judge.prompts[0]          # statement generation sees the response
    assert "The capital is Paris." in judge.prompts[1]        # verdicts see the retrieved context


def test_request_path_through_instructor_and_the_openai_sdk():
    fake = FakeJudge({
        "StatementGeneratorOutput": {"statements": ["a", "b"]},
        "NLIStatementOutput": {"statements": [{"statement": "a", "reason": "r", "verdict": 1},
                                              {"statement": "b", "reason": "r", "verdict": 0}]},
    })

    async def run():
        client = AsyncOpenAI(base_url=fake.base_url, api_key="sk-test", max_retries=0)      # created inside the loop
        llm = llm_factory("gpt-4o-mini", client=client)
        return await Faithfulness(llm=llm).ascore(user_input="q", response="r", retrieved_contexts=["c"])

    assert asyncio.run(run()).value == 0.5
    assert [title for title, _ in fake.calls] == ["StatementGeneratorOutput", "NLIStatementOutput"]

import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer


class FakeJudge:
    """OpenAI-compatible server; `script` maps a response-model name to a dict or a callable(prompt) -> dict."""

    def __init__(self, script):
        self.script, self.calls = script, []
        outer = self

        class Handler(BaseHTTPRequestHandler):
            def log_message(self, *args):
                pass

            def do_POST(self):
                body = json.loads(self.rfile.read(int(self.headers["content-length"])))
                system = next(m["content"] for m in body["messages"] if "json_schema:" in str(m["content"]))
                title = json.loads(system.split("json_schema:")[1].split("Make sure to return")[0])["title"]
                prompt = body["messages"][-1]["content"]
                outer.calls.append((title, prompt))
                answer = outer.script[title]
                content = json.dumps(answer(prompt) if callable(answer) else answer)
                data = json.dumps({
                    "id": "fake", "object": "chat.completion", "created": 0, "model": body["model"],
                    "choices": [{"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": content}}],
                    "usage": {"prompt_tokens": 100, "completion_tokens": 20, "total_tokens": 120},
                }).encode()
                self.send_response(200)
                self.send_header("content-type", "application/json")
                self.send_header("content-length", str(len(data)))
                self.end_headers()
                self.wfile.write(data)

        self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
        threading.Thread(target=self.server.serve_forever, daemon=True).start()
        self.base_url = f"http://127.0.0.1:{self.server.server_address[1]}/v1"

import json
import os
from pathlib import Path

import pandas as pd
from openai import AsyncOpenAI
from ragas import EvaluationDataset, SingleTurnSample
from ragas.cache import DiskCacheBackend
from ragas.llms import llm_factory
from ragas.metrics.collections import ContextRecall, Faithfulness

from evals.harness import ascore_dataset

GOLDEN = Path("evals/datasets/golden.jsonl")             # {"user_input", "reference", "retrieved_contexts"} per line


async def run_suite(answer_fn, concurrency: int = 4) -> pd.DataFrame:
    rows = [json.loads(line) for line in GOLDEN.read_text().splitlines() if line.strip()]
    samples = [SingleTurnSample(response=answer_fn(r["user_input"]), **r) for r in rows]     # run the system under test
    model = os.getenv("JUDGE_MODEL", "gpt-4o-mini")
    llm = llm_factory(
        model,
        client=AsyncOpenAI(timeout=60, max_retries=2),   # created inside the running loop; reads OPENAI_API_KEY / OPENAI_BASE_URL
        temperature=0,
        cache=None if os.getenv("RAGAS_NO_CACHE") else DiskCacheBackend(f".ragas_cache/{model}"),   # key ignores the model: one dir per model
    )
    metrics = [Faithfulness(llm=llm), ContextRecall(llm=llm)]
    return await ascore_dataset(EvaluationDataset(samples=samples), metrics, concurrency=concurrency)

import asyncio
import os

import pytest

from evals.scoring import run_suite


def my_rag_app(question: str) -> str:
    return "Refunds are accepted within 30 days of purchase."        # replace with your pipeline


@pytest.fixture(scope="session")
def scores():
    if not os.getenv("OPENAI_API_KEY"):
        pytest.skip("OPENAI_API_KEY not set")
    df = asyncio.run(run_suite(my_rag_app))                          # one event loop, one run, many assertions
    df.to_csv("eval-results.csv", index=False)                       # uploaded as a CI artifact
    return df

import json
from pathlib import Path

import pytest

pytestmark = pytest.mark.ragas

THRESHOLDS = {"faithfulness": 0.85, "context_recall": 0.80}   # aggregate gate: mean over the golden set
SAMPLE_FLOOR = {"faithfulness": 0.50}                        # per-sample gate: no single answer below this
MAX_NAN_RATE = 0.05
BASELINE = Path(__file__).parent / "baseline.json"
TOLERANCE = 0.03


def test_judge_calls_succeeded(scores):
    for metric in THRESHOLDS:
        nan_rate = scores[metric].isna().mean()
        assert nan_rate <= MAX_NAN_RATE, f"{metric}: {nan_rate:.0%} of judge calls failed"


@pytest.mark.parametrize("metric,threshold", THRESHOLDS.items())
def test_mean_score(scores, metric, threshold):
    mean = scores[metric].mean()                              # pandas skips NaN, so test_judge_calls_succeeded guards them
    assert mean >= threshold, f"{metric} mean {mean:.3f} < {threshold}"


@pytest.mark.parametrize("metric,floor", SAMPLE_FLOOR.items())
def test_no_catastrophic_sample(scores, metric, floor):
    bad = scores[scores[metric] < floor]
    assert bad.empty, "below floor:\n" + bad[["user_input", metric]].to_string()


def test_no_regression_against_baseline(scores):
    baseline = json.loads(BASELINE.read_text())
    for metric, old in baseline.items():
        assert scores[metric].mean() >= old - TOLERANCE, f"{metric} dropped from {old:.3f}"

name: rag-evals
on:
  pull_request:
    types: [opened, synchronize, reopened, labeled]
  schedule:
    - cron: "0 3 * * *"
  workflow_dispatch:

env:
  RAGAS_DO_NOT_TRACK: "true"

jobs:
  unit:                                    # no judge calls, no secrets
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v5
      - run: uv sync --locked
      - run: uv run pytest -q              # pytest.ini deselects -m ragas

  judged:                                  # nightly, on demand, or on PRs labelled "run-evals"
    if: github.event_name != 'pull_request' || contains(github.event.pull_request.labels.*.name, 'run-evals')
    needs: unit
    runs-on: ubuntu-latest
    timeout-minutes: 30
    concurrency: { group: "rag-evals-${{ github.ref }}", cancel-in-progress: true }
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v5
      - run: uv sync --locked
      - uses: actions/cache@v4
        with:
          path: .ragas_cache
          key: ragas-${{ hashFiles('uv.lock', 'evals/datasets/golden.jsonl') }}
          restore-keys: ragas-
      - run: uv run pytest -q -m ragas -o addopts=""
        env: { OPENAI_API_KEY: "${{ secrets.OPENAI_API_KEY }}", JUDGE_MODEL: gpt-4o-mini }
      - uses: actions/upload-artifact@v4
        if: always()
        with: { name: eval-results, path: eval-results.csv }

import httpx
from openai import AsyncOpenAI

tokens = {"calls": 0, "input": 0, "output": 0}


async def count(response: httpx.Response):                      # runs for every judge response
    await response.aread()
    usage = response.json().get("usage") or {}
    tokens["calls"] += 1
    tokens["input"] += usage.get("prompt_tokens", 0)
    tokens["output"] += usage.get("completion_tokens", 0)


client = AsyncOpenAI(http_client=httpx.AsyncClient(event_hooks={"response": [count]}))
# build the client inside the coroutine that runs the suite, pass it to llm_factory, compare `tokens` with a budget