evaluation.py

import logging
import sys
import os
import pandas as pd
from dotenv import load_dotenv
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Response
from llama_index.core.evaluation import (
    DatasetGenerator,
    RelevancyEvaluator,
    FaithfulnessEvaluator,
    EvaluationResult,
)
from llama_index.llms.openai import OpenAI
from tabulate import tabulate
import textwrap
import argparse
import traceback
from httpx import ReadTimeout  

logging.basicConfig(stream=sys.stdout, level=logging.INFO)
logging.getLogger().addHandler(logging.StreamHandler(stream=sys.stdout))

parser = argparse.ArgumentParser(description='Process documents and questions for evaluation.')
parser.add_argument('--num_documents', type=int, default=None,
                    help='Number of documents to process (default: all)')
parser.add_argument('--skip_documents', type=int, default=0,
                    help='Number of documents to skip at the beginning (default: 0)')
parser.add_argument('--num_questions', type=int, default=None,
                    help='Number of questions to process (default: all)')
parser.add_argument('--skip_questions', type=int, default=0,
                    help='Number of questions to skip at the beginning (default: 0)')
parser.add_argument('--process_last_questions', action='store_true',
                    help='Process last N questions instead of first N')
args = parser.parse_args()

load_dotenv('.env')

reader = SimpleDirectoryReader("/tmp/elastic/production-readiness-review")
documents = reader.load_data()
print(f"First document: {documents[0].text}")
print(f"Second document: {documents[1].text}")
print(f"Thrid document: {documents[2].text}")


if args.skip_documents > 0:
    documents = documents[args.skip_documents:]

if args.num_documents is not None:
    documents = documents[:args.num_documents]

print(f"Number of documents loaded: {len(documents)}")

llm = OpenAI(model="gpt-4o", request_timeout=120)  

data_generator = DatasetGenerator.from_documents(documents, llm=llm)

try:
    eval_questions = data_generator.generate_questions_from_nodes()
    if isinstance(eval_questions, str):
        eval_questions_list = eval_questions.strip().split('\n')
    else:
        eval_questions_list = eval_questions
    eval_questions_list = [q for q in eval_questions_list if q.strip()]

    if args.skip_questions > 0:
        eval_questions_list = eval_questions_list[args.skip_questions:]

    if args.num_questions is not None:
        if args.process_last_questions:
            eval_questions_list = eval_questions_list[-args.num_questions:]
        else:
            eval_questions_list = eval_questions_list[:args.num_questions]

    print("\All available questions generated:")
    for idx, q in enumerate(eval_questions):
        print(f"{idx}. {q}")

    print("\nGenerated questions:")
    for idx, q in enumerate(eval_questions_list, start=1):
        print(f"{idx}. {q}")
except ReadTimeout as e:
    print("Request to Ollama timed out during question generation. Please check the server or increase the timeout duration.")
    traceback.print_exc()
    sys.exit(1)
except Exception as e:
    print(f"An error occurred while generating questions: {e}")
    traceback.print_exc()
    sys.exit(1)

print(f"\nTotal number of questions generated: {len(eval_questions_list)}")

evaluator_relevancy = RelevancyEvaluator(llm=llm)
evaluator_faith = FaithfulnessEvaluator(llm=llm)

vector_index = VectorStoreIndex.from_documents(documents)

def display_eval_df(
    query: str,
    response: Response,
    eval_result_relevancy: EvaluationResult,
    eval_result_faith: EvaluationResult,
) -> None:
    relevancy_feedback = getattr(eval_result_relevancy, 'feedback', '')
    relevancy_passing = getattr(eval_result_relevancy, 'passing', False)
    relevancy_passing_str = 'Pass' if relevancy_passing else 'Fail'

    relevancy_score = 1.0 if relevancy_passing else 0.0

    faithfulness_feedback = getattr(eval_result_faith, 'feedback', '')
    faithfulness_passing_bool = getattr(eval_result_faith, 'passing', False)
    faithfulness_passing = 'Pass' if faithfulness_passing_bool else 'Fail'

    def wrap_text(text, width=50):
        if text is None:
            return ''
        text = str(text)
        text = text.replace('\r', '')
        lines = text.split('\n')
        wrapped_lines = []
        for line in lines:
            wrapped_lines.extend(textwrap.wrap(line, width=width))
            wrapped_lines.append('')
        return '\n'.join(wrapped_lines)

    if response.source_nodes:
        source_content = wrap_text(response.source_nodes[0].node.get_content())
    else:
        source_content = ''

    eval_data = {
        "Query": wrap_text(query),
        "Response": wrap_text(str(response)),
        "Source": source_content,
        "Relevancy Response": relevancy_passing_str,
        "Relevancy Feedback": wrap_text(relevancy_feedback),
        "Relevancy Score": wrap_text(str(relevancy_score)),
        "Faith Response": faithfulness_passing,
        "Faith Feedback": wrap_text(faithfulness_feedback),
    }

    eval_df = pd.DataFrame([eval_data])

    print("\nEvaluation Result:")
    print(tabulate(eval_df, headers='keys', tablefmt='grid', showindex=False, stralign='left'))

query_engine = vector_index.as_query_engine(llm=llm)

total_questions = len(eval_questions_list)
for idx, question in enumerate(eval_questions_list, start=1):
    try:
        response_vector = query_engine.query(question)
        eval_result_relevancy = evaluator_relevancy.evaluate_response(
            query=question, response=response_vector
        )
        eval_result_faith = evaluator_faith.evaluate_response(
            response=response_vector
        )

        print(f"\nProcessing Question {idx} of {total_questions}:")
        display_eval_df(question, response_vector, eval_result_relevancy, eval_result_faith)
    except ReadTimeout as e:
        print(f"Request to OpenAI timed out while processing question {idx}.")
        traceback.print_exc()
        continue  
    except Exception as e:
        print(f"An error occurred while processing question {idx}: {e}")
        traceback.print_exc()
        continue