Complex RAG evaluation

In this example, we will walk through a complex RAG application and evaluate it’s performance.

Try out full example at example repo.

1) Define Pipeline

pipeline.py

from continuous_eval.eval import Module, Pipeline, Dataset, ModuleOutput
from continuous_eval.metrics.retrieval import (
    # === Deterministic
    PrecisionRecallF1,
    RankedRetrievalMetrics,
    # === LLM-based
    LLMBasedContextCoverage,
    LLMBasedContextPrecision,
)
from continuous_eval.metrics.generation.text import (
    # === Deterministic
    DeterministicFaithfulness,
    DeterministicAnswerCorrectness,
    FleschKincaidReadability,
    # === Semantic
    BertAnswerRelevance,
    BertAnswerSimilarity,
    DebertaAnswerScores,
    # === LLM-based
    LLMBasedFaithfulness,
    LLMBasedAnswerCorrectness,
    LLMBasedAnswerRelevance,
    LLMBasedStyleConsistency,
)
from typing import List, Dict
from continuous_eval.eval.tests import GreaterOrEqualThan, MeanGreaterOrEqualThan

dataset = Dataset("examples/langchain/rag_data/eval_golden_dataset")

Documents = List[Dict[str, str]]
DocumentsContent = ModuleOutput(lambda x: [z["page_content"] for z in x])

base_retriever = Module(
    name="base_retriever",
    input=dataset.question,
    output=Documents,
    eval=[
        PrecisionRecallF1().use(
            retrieved_context=DocumentsContent,
            ground_truth_context=dataset.ground_truth_context,
        ),
        RankedRetrievalMetrics().use(
            retrieved_context=DocumentsContent,
            ground_truth_context=dataset.ground_truth_context,
        ),
    ],
    tests=[
        GreaterOrEqualThan(
            test_name="Context Recall", metric_name="context_recall", min_value=0.9
        ),
    ],
)

bm25_retriever = Module(
    name="bm25_retriever",
    input=dataset.question,
    output=Documents,
    eval=[
        PrecisionRecallF1().use(
            retrieved_context=DocumentsContent,
            ground_truth_context=dataset.ground_truth_context,
        ),
        RankedRetrievalMetrics().use(
            retrieved_context=DocumentsContent,
            ground_truth_context=dataset.ground_truth_context,
        ),
    ],
    tests=[
        GreaterOrEqualThan(
            test_name="Context Recall", metric_name="context_recall", min_value=0.9
        ),
    ],
)

hyde_generator = Module(
    name="HyDE_generator",
    input=dataset.question,
    output=str,
)

hyde_retriever = Module(
    name="HyDE_retriever",
    input=hyde_generator,
    output=Documents,
    eval=[
        PrecisionRecallF1().use(
            retrieved_context=DocumentsContent,
            ground_truth_context=dataset.ground_truth_context,
        ),
        RankedRetrievalMetrics().use(
            retrieved_context=DocumentsContent,
            ground_truth_context=dataset.ground_truth_context,
        ),
    ],
    tests=[
        MeanGreaterOrEqualThan(
            test_name="Context Recall", metric_name="context_recall", min_value=0.9
        ),
    ],
)


reranker = Module(
    name="cohere_reranker",
    input=(base_retriever, hyde_retriever, bm25_retriever),
    output=Documents,
    eval=[
        PrecisionRecallF1().use(
            retrieved_context=DocumentsContent,
            ground_truth_context=dataset.ground_truth_context,
        ),
        RankedRetrievalMetrics().use(
            retrieved_context=DocumentsContent,
            ground_truth_context=dataset.ground_truth_context,
        ),
    ],
    tests=[
        MeanGreaterOrEqualThan(
            test_name="Context Recall", metric_name="context_recall", min_value=0.9
        ),
    ],
)

llm = Module(
    name="answer_generator",
    input=reranker,
    output=str,
    eval=[
        FleschKincaidReadability().use(answer=ModuleOutput()),
        DeterministicAnswerCorrectness().use(
            answer=ModuleOutput(), ground_truth_answers=dataset.ground_truths
        ),
        DeterministicFaithfulness().use(
            answer=ModuleOutput(),
            retrieved_context=ModuleOutput(DocumentsContent, module=reranker),
        ),
        BertAnswerRelevance().use(answer=ModuleOutput(), question=dataset.question),
        BertAnswerSimilarity().use(
            answer=ModuleOutput(), ground_truth_answers=dataset.ground_truths
        ),
        DebertaAnswerScores().use(
            answer=ModuleOutput(), ground_truth_answers=dataset.ground_truths
        ),
        # LLMBasedFaithfulness().use(
        #     answer=ModuleOutput(),
        #     retrieved_context=ModuleOutput(DocumentsContent, module=reranker),
        #     question=dataset.question,
        # ),
        # LLMBasedAnswerCorrectness().use(
        #     question=dataset.question,
        #     answer=ModuleOutput(),
        #     ground_truth_answers=dataset.ground_truths,
        # ),
        # LLMBasedAnswerRelevance().use(
        #     question=dataset.question,
        #     answer=ModuleOutput(),
        # ),
        # LLMBasedStyleConsistency().use(
        #     answer=ModuleOutput(), ground_truth_answers=dataset.ground_truths
        # ),
    ],
    tests=[
        GreaterOrEqualThan(
            test_name="Readability", metric_name="flesch_reading_ease", min_value=20.0
        ),
        GreaterOrEqualThan(
            test_name="Answer Correctness", metric_name="rouge_l_recall", min_value=0.8
        ),
    ],
)

pipeline = Pipeline([base_retriever, hyde_generator, hyde_retriever, bm25_retriever, reranker, llm], dataset=dataset)

print(pipeline.graph_repr())```
</TabItem>
<TabItem label="2) Define Dataset">
```yaml title="manifest.yaml"
name: Paul Graham's Essays
description: Paul Graham's Essays Q&A
format: jsonl
license: CC0
fields:
question:
    description: The question asked by the user
    type: str
    ground_truth: false
ground_truths:
    description: The answer(s) to the question
    type: List[str]
    ground_truth: true
ground_truth_context:
    description: Ground truth contexts
    type: List[str]
    ground_truth: true
metadata:
    description: Info about the question
    type: List[Dict[str,str]]
    ground_truth: false
question_type:
    description: The type of question
    type: str
    ground_truth: false

3) Run Complex RAG App

complex_rag_app.py

import os
from pathlib import Path

from continuous_eval.eval.manager import eval_manager
from langchain_community.document_loaders.directory import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.retrievers.document_compressors import CohereRerank
from langchain_community.vectorstores import Chroma
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_openai import OpenAIEmbeddings
from langchain_community.retrievers import BM25Retriever
from dotenv import load_dotenv
from examples.langchain.complex_rag.pipeline import pipeline

load_dotenv()

eval_manager.set_pipeline(pipeline)

# Load documents and split
loader = DirectoryLoader("examples/langchain/rag_data/documents/208_219_graham_essays")
docs = loader.load()
TextSplitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=0)
split_docs = TextSplitter.split_documents(docs)

# Set up Vectorstore
db = Chroma(
    persist_directory=str("examples/langchain/rag_data/vectorstore/208_219_chroma_db"),
    embedding_function=OpenAIEmbeddings(),
)

# Set up LLM
model = ChatGoogleGenerativeAI(model="gemini-pro")

def base_retrieve(q):
    # Basic retriever
    basic_retriever = db.as_retriever(
        search_type="similarity", search_kwargs={"k": 4}
    )
    return basic_retriever.invoke(q)

def bm25_retrieve(q):
    # bm25 retriever
    bm25_retriever = BM25Retriever.from_documents(documents=docs)
    bm25_retriever.k = 2
    return bm25_retriever.invoke(q)

def hyde_generator(q):
    # HyDE generator
    system_prompt = (
        "Generate a hypothetical document paragraph that contains an answer to the question below."
    )
    user_prompt = f"Question: {q}\n\n"
    try:
        result = model.invoke(system_prompt + user_prompt).content
    except Exception as e:
        print(f"{e} unable to generate hypothetical document, using question as HyDE")
        result = q
    return result

def hyde_retrieve(hypothetical_doc):
    # HyDE retriever
    hyde_retriever = db.as_retriever(
        search_type="similarity", search_kwargs={"k": 3}
    )
    return hyde_retriever.invoke(hypothetical_doc)

def rerank(q, retrieved_docs):
    # reranker
    compressor = CohereRerank(cohere_api_key=os.getenv("COHERE_API_KEY"))
    compressor.top_n = 5
    return compressor.compress_documents(retrieved_docs, q)


def ask(q, retrieved_docs):
    system_prompt = (
        "You are and expert of the life of Paul Graham.\n"
        "Answer the question below based on the context provided."
    )
    user_prompt = f"Question: {q}\n\n"
    user_prompt += "Contexts:\n" + "\n".join(
        [doc.page_content for doc in retrieved_docs]
    )
    try:
        result = model.invoke(system_prompt + user_prompt).content
    except Exception as e:
        print(e)
        result = "Sorry, I cannot answer this question."
    return result


if __name__ == "__main__":
    eval_manager.start_run()
    while eval_manager.is_running():
        if eval_manager.curr_sample is None:
            break
        q = eval_manager.curr_sample["question"]
        # Base Retriever
        base_retrieved_docs = base_retrieve(q)
        eval_manager.log("base_retriever", [doc.__dict__ for doc in base_retrieved_docs])
        # BM25 Retriever
        bm25_retrieved_docs = bm25_retrieve(q)
        eval_manager.log("bm25_retriever", [doc.__dict__ for doc in bm25_retrieved_docs])
        # HyDE Generator
        hypothetical_doc = hyde_generator(q)
        eval_manager.log("HyDE_generator", hypothetical_doc)
        # HyDE Retriever
        hyde_retrieved_docs = hyde_retrieve(hypothetical_doc)
        eval_manager.log("HyDE_retriever", [doc.__dict__ for doc in hyde_retrieved_docs])

        fused_docs = base_retrieved_docs + bm25_retrieved_docs + hyde_retrieved_docs
        # Reranker
        reranked_docs = rerank(q, fused_docs)
        eval_manager.log("cohere_reranker", [doc.__dict__ for doc in reranked_docs])
        # Generator
        response = ask(q, reranked_docs)
        eval_manager.log("answer_generator", response)
        print(f"Q: {q}\nA: {response}\n")
        eval_manager.next_sample()

    eval_manager.evaluation.save(Path("results.jsonl"))

4) Run Eval

eval.py

from pathlib import Path

from continuous_eval.eval.manager import eval_manager
from examples.langchain.complex_rag.pipeline import pipeline

if __name__ == "__main__":
    eval_manager.set_pipeline(pipeline)

    # Evaluation
    eval_manager.evaluation.load(Path("results.jsonl"))
    eval_manager.run_metrics()
    eval_manager.metrics.save(Path("metrics_results.json"))

    # Tests
    eval_manager.metrics.load(Path("metrics_results.json"))
    agg = eval_manager.metrics.aggregate() # optional
    print(agg)
    eval_manager.run_tests()
    eval_manager.tests.save(Path("test_results.json"))

    # eval_manager.tests.load(Path("test_results.json"))
    for module_name, test_results in eval_manager.tests.results.items():
        print(f"{module_name}")
        for test_name in test_results:
            print(f" - {test_name}: {test_results[test_name]}")