Evaluation of RAG pipelines with Ragas

import os
 
# get keys for your project from https://cloud.langfuse.com
os.environ["LANGFUSE_PUBLIC_KEY"] = ""
os.environ["LANGFUSE_SECRET_KEY"] = ""
 
# your openai key
os.environ["OPENAI_API_KEY"] = ""
 
# Your host, defaults to https://cloud.langfuse.com
# For US data region, set to "https://us.cloud.langfuse.com"
# os.environ["LANGFUSE_HOST"] = "http://localhost:3000"

%pip install langfuse datasets ragas llama_index python-dotenv openai --upgrade

from datasets import load_dataset
 
fiqa_eval = load_dataset("explodinggradients/fiqa", "ragas_eval")['baseline']
fiqa_eval

# import metrics
from ragas.metrics import faithfulness, answer_relevancy, context_precision
from ragas.metrics.critique import SUPPORTED_ASPECTS, harmfulness
 
# metrics you chose
metrics = [faithfulness, answer_relevancy, context_precision, harmfulness]
 
for m in metrics:
    print(m.name)
    # also init the metrics
    m.init_model()

row = fiqa_eval[0]
row['question'], row['answer']

('How to deposit a cheque issued to an associate in my business into my business account?',
 '\nThe best way to deposit a cheque issued to an associate in your business into your business account is to open a business account with the bank. You will need a state-issued "dba" certificate from the county clerk\'s office as well as an Employer ID Number (EIN) issued by the IRS. Once you have opened the business account, you can have the associate sign the back of the cheque and deposit it into the business account.')

from langfuse import Langfuse
 
langfuse = Langfuse()

def score_with_ragas(query, chunks, answer):
    scores = {}
    for m in metrics:
        print(f"calculating {m.name}")
        scores[m.name] = m.score_single(
            {'question': query, 'contexts': chunks, 'answer': answer}
        )
    return scores

from langfuse.model import CreateTrace, CreateSpan, CreateGeneration, CreateEvent, CreateScore
 
# start a new trace when you get a question
question = row['question']
trace = langfuse.trace(CreateTrace(name = "rag"))
 
# retrieve the relevant chunks
# chunks = get_similar_chunks(question)
contexts = row['contexts']
# pass it as span
trace.span(CreateSpan(
    name = "retrieval", input={'question': question}, output={'contexts': contexts}
))
 
# use llm to generate a answer with the chunks
# answer = get_response_from_llm(question, chunks)
answer = row['answer']
trace.span(CreateSpan(
    name = "generation", input={'question': question, 'contexts': contexts}, output={'answer': answer}
))
 
# compute scores for the question, context, answer tuple
ragas_scores = score_with_ragas(question, contexts, answer)
ragas_scores

# send the scores
for m in metrics:
    trace.score(CreateScore(name=m.name, value=ragas_scores[m.name]))

from langfuse.model import CreateTrace, CreateSpan, CreateGeneration, CreateEvent, CreateScore
# fiqa traces
for interaction in fiqa_eval.select(range(10, 20)):
    trace = langfuse.trace(CreateTrace(name = "rag"))
    trace.span(CreateSpan(
        name = "retrieval",
        input={'question': question},
        output={'contexts': contexts}
    ))
    trace.span(CreateSpan(
        name = "generation",
        input={'question': question, 'contexts': contexts},
        output={'answer': answer}
    ))
 
# await that Langfuse SDK has processed all events before trying to retrieve it in the next step
langfuse.flush()

def get_traces(name=None, limit=None, user_id=None):
    all_data = []
    page = 1
 
    while True:
        response = langfuse.client.trace.list(
            name=name, page=page, user_id=user_id
        )
        if not response.data:
            break
        page += 1
        all_data.extend(response.data)
        if len(all_data) > limit:
            break
 
    return all_data[:limit]

from random import sample
 
NUM_TRACES_TO_SAMPLE = 3
traces = get_traces(name='rag', limit=5)
traces_sample = sample(traces, NUM_TRACES_TO_SAMPLE)
 
len(traces_sample)

# score on a sample
from random import sample
 
evaluation_batch = {
    "question": [],
    "contexts": [],
    "answer": [],
    "trace_id": [],
}
 
for t in traces_sample:
    observations = [langfuse.client.observations.get(o) for o in t.observations]
    for o in observations:
        if o.name == 'retrieval':
            question = o.input['question']
            contexts = o.output['contexts']
        if o.name=='generation':
            answer = o.output['answer']
    evaluation_batch['question'].append(question)
    evaluation_batch['contexts'].append(contexts)
    evaluation_batch['answer'].append(answer)
    evaluation_batch['trace_id'].append(t.id)

# run ragas evaluate
from datasets import Dataset
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy
 
ds = Dataset.from_dict(evaluation_batch)
r = evaluate(ds, metrics=[faithfulness, answer_relevancy])

r

df = r.to_pandas()
 
# add the langfuse trace_id to the result dataframe
df["trace_id"] = ds["trace_id"]
 
df.head()

from langfuse.model import InitialScore
 
for _, row in df.iterrows():
    for metric_name in ["faithfulness", "answer_relevancy"]:
        langfuse.score(InitialScore(
            name=metric_name,
            value=row[metric_name],
            trace_id=row["trace_id"]))