Ragas — Test Sets & Experiments¶

from langchain_community.document_loaders import DirectoryLoader      # needs langchain-community<0.4.2
from openai import AsyncOpenAI
from ragas.embeddings.base import embedding_factory
from ragas.llms import llm_factory
from ragas.testset import TestsetGenerator

client = AsyncOpenAI()
llm = llm_factory("gpt-4o-mini", client=client)
emb = embedding_factory("openai", model="text-embedding-3-small", client=client)

docs = DirectoryLoader("kb/", glob="**/*.md").load()                  # LangChain Documents
generator = TestsetGenerator(llm=llm, embedding_model=emb)
testset = generator.generate_with_langchain_docs(docs, testset_size=30, raise_exceptions=False)
df = testset.to_pandas()

from ragas.testset.persona import Persona
from ragas.testset.synthesizers.single_hop.specific import SingleHopSpecificQuerySynthesizer

chunks = ["Refunds are accepted within 30 days of purchase. Sale items are final.", "Shipping takes 3-5 business days within the EU."]
generator = TestsetGenerator(
    llm=llm, embedding_model=emb,
    persona_list=[Persona(name="Support agent", role_description="Answers customer questions about orders")],
)
testset = generator.generate_with_chunks(
    chunks, testset_size=20,
    query_distribution=[(SingleHopSpecificQuerySynthesizer(llm=llm), 1.0)],
)

from ragas.testset.graph import KnowledgeGraph

generator.knowledge_graph.save("kb_graph.json")
kg = KnowledgeGraph.load("kb_graph.json")                              # KnowledgeGraph(nodes: 3, relationships: 4)
more = TestsetGenerator(llm=llm, embedding_model=emb, knowledge_graph=kg).generate(testset_size=20)

import asyncio

from openai import AsyncOpenAI
from ragas.llms import llm_factory
from ragas.metrics.collections import Faithfulness


async def ground_check(df, concurrency=8):
    llm = llm_factory("gpt-4o-mini", client=AsyncOpenAI(), temperature=0)     # client created inside the loop
    faith, sem = Faithfulness(llm=llm), asyncio.Semaphore(concurrency)

    async def one(row):
        async with sem:
            result = await faith.ascore(user_input=row.user_input, response=row.reference,
                                        retrieved_contexts=row.reference_contexts)
            return result.value

    return await asyncio.gather(*[one(r) for r in df.itertuples()])


df["ref_grounded"] = asyncio.run(ground_check(df))
suspect = df[df["ref_grounded"] < 0.8]

import asyncio

from openai import AsyncOpenAI
from pydantic import BaseModel
from ragas import Dataset, experiment
from ragas.llms import llm_factory
from ragas.metrics.collections import ContextRecall, Faithfulness


class Result(BaseModel):
    user_input: str
    response: str
    faithfulness: float
    context_recall: float


sem = asyncio.Semaphore(8)                                    # arun starts every row at once: bound it yourself


def build(llm):
    faith, recall = Faithfulness(llm=llm), ContextRecall(llm=llm)

    @experiment(Result, name_prefix="rag")
    async def run(row, model: str):
        async with sem:
            response, contexts = await my_rag_app(row["user_input"], model=model)       # your system under test
            f = await faith.ascore(user_input=row["user_input"], response=response, retrieved_contexts=contexts)
            r = await recall.ascore(user_input=row["user_input"], retrieved_contexts=contexts, reference=row["reference"])
            return Result(user_input=row["user_input"], response=response, faithfulness=f.value, context_recall=r.value)

    return run


async def main():
    llm = llm_factory("gpt-4o-mini", client=AsyncOpenAI(), temperature=0)
    run = build(llm)
    ds = Dataset.load("golden", "local/jsonl", root_dir="evals")                       # rows are dicts
    base = await run.arun(ds, name="baseline", model="small")                           # saved as evals/experiments/rag-baseline.jsonl
    cand = await run.arun(ds, name="candidate", model="large")
    return base, cand


base, cand = asyncio.run(main())

assert len(base) == len(ds) and len(cand) == len(ds), "rows were dropped by failures"

b, c = base.to_pandas().set_index("user_input"), cand.to_pandas().set_index("user_input")
cmp = b[["faithfulness", "context_recall"]].join(c[["faithfulness", "context_recall"]], lsuffix="_base", rsuffix="_cand")
cmp["d_faithfulness"] = cmp["faithfulness_cand"] - cmp["faithfulness_base"]
print(cmp[["faithfulness_base", "faithfulness_cand", "d_faithfulness"]].sort_values("d_faithfulness").head(10))   # worst regressions first
print(cmp.mean())

def to_sample(question: str, docs, answer: str) -> dict:
    return {"user_input": question, "retrieved_contexts": [d.page_content for d in docs], "response": answer}