LLM Evals for Text Summarization and RAG use-cases.
from .rag_evaluator import RagEvaluator
from ...loaders.rag_loader import RagLoader
from ...metrics.rag.answer_relevance_failure import AnswerRelevanceFailure
from ...publishers.publisher_log import PublisherLog
from ...llms.rag.answer_relevance import AnswerRelevance


class AnswerRelevanceEvaluator(RagEvaluator):
    """
    Evaluator for answer relevance in rag chatbot.

    Attributes:
        dataset: Dataset containing instances for evaluation.
        context_relevance_evaluator: Evaluator for context relevance.
        publisher_log: JSON publisher to save the evaluation logs.
        performance_report_filename: txt file to save the perfrormance of a batch
        metrics: List of metrics to evaluate.
        logs: List to accumulate evaluation results for each instance.
    """

    # Chamge metric
    metric_str_to_class = {"answer_relevance_failure": AnswerRelevanceFailure}

    def __init__(
        self,
        loader,
        log_filepath="data/logs/log_rag_ans_rel_eval.json",
        log_format="json",
        performance_filepath="data/logs/perf_rag_ans_rel_eval.txt",
        llm_model="gpt-3.5-turbo",
        metrics=["answer_relevance_failure"],
        open_ai_key=None,
    ):
        """
        Initialize the evaluator with given parameters.

        Args:
        - loader: An instance of SummarizationLoader.
        - log_filepath: Path to save the logs.
        - llm_model: Language model to be used.
        - metrics: List of metrics for evaluation.
        """
        if not isinstance(loader, RagLoader):
            raise TypeError("Loader must be an instance of RagLoader")
        # Load data
        self.dataset = loader.processed_dataset
        # Intialize LLMs
        self.llm_model = llm_model
        self.answer_relevance_evaluator = AnswerRelevance(llm_model, open_ai_key)
        # Initialize logging
        self.log_format = log_format
        if log_format is not None:
            self.publisher_log = PublisherLog(log_filepath, log_format)
        self.logs = []
        self.n_instances = 0
        # Intialize metrics
        self.performance_filepath = performance_filepath
        self.metrics = metrics
        self.label_counts = {}
        for metric in metrics:
            setattr(self, f"{metric}_scores", {})

    def _evaluate_element(self, instance):
        """Evaluate an instance for hallucination."""
        question = instance["question"]
        answer = instance["answer"]
        if "label" in instance:
            label = instance["label"]
        else:
            label = "overall"
        # Run LLM Evaluator for faithfulness
        anw_rel_eval = self.answer_relevance_evaluator.evaluate(question, answer)

        metric_results = {}
        # Compute metrics
        if anw_rel_eval is None:
            metric_results["evaluation"] = "undefined"
        else:
            for metric in self.metrics:
                metric_class = self.metric_str_to_class.get(metric)
                metric_result, explanation = metric_class.compute(anw_rel_eval)
                metric_results[metric] = metric_result
                metric_results["reason"] = explanation
                self.update_metric_aggr(metric, label, metric_result)
            self.n_instances = self.n_instances + 1
            self.label_counts[label] = self.label_counts.get(label, 0) + 1
        return {
            "question": question,
            "answer": answer,
            "label": label,
            **metric_results,
        }

    def update_metric_aggr(self, metric, label, aggr_score):
        """Update the aggregated score for a specific metric and label."""
        metric_aggr = getattr(self, f"{metric}_scores", {})
        current_score = metric_aggr.get(label, 0)
        metric_aggr[label] = current_score + aggr_score
        setattr(self, f"{metric}_scores", metric_aggr)

    def get_metric_aggr(self, metric, label):
        """Compute the average scores based on the provided score dictionary."""
        metric_aggr = getattr(self, f"{metric}_scores", {})
        return metric_aggr.get(label, None)

    def get_average_scores(self, score_dict):
        """Compute average scores for a metric"""
        avg_scores = {}
        sum_score = 0
        n_instances = 0
        for label_type, total_score in score_dict.items():
            avg_scores[label_type] = total_score / self.label_counts[label_type]
            sum_score = sum_score + total_score
            n_instances = n_instances + self.label_counts[label_type]
        avg_scores["overall"] = sum_score / n_instances
        return avg_scores

    def compute_average_scores(self):
        """Compute average scores for each metric."""
        avg_scores = {}
        for metric in self.metrics:
            scores = getattr(self, f"{metric}_scores")
            avg_score = self.get_average_scores(scores)
            avg_scores[metric] = avg_score
        return avg_scores

    def generate_performance_report(self, filename):
        """Generate a performance report and save it to the provided filename."""
        avg_scores = self.compute_average_scores()
        with open(filename, "w") as f:
            f.write(f"Model: {self.llm_model}\n")
            f.write("\nNumber of instances per Label:\n")
            for label, cnt in self.label_counts.items():
                f.write(f"{label}: {cnt}\n")
            f.write(f"total: {self.n_instances}\n")
            for metric in self.metrics:
                f.write(f"\nAverage {metric} per Label:\n")
                for label, avg in avg_scores[metric].items():
                    f.write(f"{label}: {avg}\n")

    def run(self):
        """Evaluate all instances in the dataset."""
        for instance in self.dataset:
            log = self._evaluate_element(instance)
            self.logs.append(log)
        self.generate_performance_report(self.performance_filepath)
        if self.log_format is not None:
            self.publisher_log.write(self.logs)
        return self.logs

ariadne_ai/evaluators/rag/answer_relevance_evaluator.py
athina-ai/ariadne
from .rag_evaluator import RagEvaluator
from ...loaders.rag_loader import RagLoader
from ...metrics.rag.context_relevance_failure import ContextRelevanceFailure
from ...publishers.publisher_log import PublisherLog
from ...llms.rag.context_relevance import ContextRelevance


class ContextRelevanceEvaluator(RagEvaluator):
    """
    Evaluator for context relevance in rag chatbot.

    Attributes:
        dataset: Dataset containing instances for evaluation.
        context_relevance_evaluator: Evaluator for context relevance.
        publisher_log: JSON publisher to save the evaluation logs.
        performance_report_filename: txt file to save the perfrormance of a batch
        metrics: List of metrics to evaluate.
        logs: List to accumulate evaluation results for each instance.
    """

    # Chamge metric
    metric_str_to_class = {"context_relevance_failure": ContextRelevanceFailure}

    def __init__(
        self,
        loader,
        log_filepath="data/logs/log_rag_cont_rel_eval.json",
        log_format="json",
        performance_filepath="data/logs/perf_rag_cont_rel_eval.txt",
        llm_model="gpt-3.5-turbo",
        metrics=["context_relevance_failure"],
        open_ai_key=None,
    ):
        """
        Initialize the evaluator with given parameters.

        Args:
        - loader: An instance of SummarizationLoader.
        - log_filepath: Path to save the logs.
        - llm_model: Language model to be used.
        - metrics: List of metrics for evaluation.
        """
        if not isinstance(loader, RagLoader):
            raise TypeError("Loader must be an instance of RagLoader")
        # Load data
        self.dataset = loader.processed_dataset
        # Intialize LLMs
        self.llm_model = llm_model
        self.context_relevance_evaluator = ContextRelevance(llm_model, open_ai_key)
        # Initialize logging
        self.log_format = log_format
        if log_format is not None:
            self.publisher_log = PublisherLog(log_filepath, log_format)
        self.logs = []
        self.n_instances = 0
        # Intialize metrics
        self.performance_filepath = performance_filepath
        self.metrics = metrics
        self.label_counts = {}
        for metric in metrics:
            setattr(self, f"{metric}_scores", {})

    def _evaluate_element(self, instance):
        """Evaluate an instance for hallucination."""
        question = instance["question"]
        context = instance["context"]
        if "label" in instance:
            label = instance["label"]
        else:
            label = "overall"
        # Run LLM Evaluator for faithfulness
        cont_rel_eval = self.context_relevance_evaluator.evaluate(question, context)

        metric_results = {}
        # Compute metrics
        if cont_rel_eval is None:
            metric_results["evaluation"] = "undefined"
        else:
            for metric in self.metrics:
                metric_class = self.metric_str_to_class.get(metric)
                metric_result, explanation = metric_class.compute(cont_rel_eval)
                metric_results[metric] = metric_result
                metric_results["reason"] = explanation
                self.update_metric_aggr(metric, label, metric_result)
            self.n_instances = self.n_instances + 1
            self.label_counts[label] = self.label_counts.get(label, 0) + 1
        return {
            "question": question,
            "context": context,
            "label": label,
            **metric_results,
        }

    def update_metric_aggr(self, metric, label, aggr_score):
        """Update the aggregated score for a specific metric and label."""
        metric_aggr = getattr(self, f"{metric}_scores", {})
        current_score = metric_aggr.get(label, 0)
        metric_aggr[label] = current_score + aggr_score
        setattr(self, f"{metric}_scores", metric_aggr)

    def get_metric_aggr(self, metric, label):
        """Compute the average scores based on the provided score dictionary."""
        metric_aggr = getattr(self, f"{metric}_scores", {})
        return metric_aggr.get(label, None)

    def get_average_scores(self, score_dict):
        """Compute average scores for a metric"""
        avg_scores = {}
        sum_score = 0
        n_instances = 0
        for label_type, total_score in score_dict.items():
            avg_scores[label_type] = total_score / self.label_counts[label_type]
            sum_score = sum_score + total_score
            n_instances = n_instances + self.label_counts[label_type]
        avg_scores["overall"] = sum_score / n_instances
        return avg_scores

    def compute_average_scores(self):
        """Compute average scores for each metric."""
        avg_scores = {}
        for metric in self.metrics:
            scores = getattr(self, f"{metric}_scores")
            avg_score = self.get_average_scores(scores)
            avg_scores[metric] = avg_score
        return avg_scores

    def generate_performance_report(self, filename):
        """Generate a performance report and save it to the provided filename."""
        avg_scores = self.compute_average_scores()
        with open(filename, "w") as f:
            f.write(f"Model: {self.llm_model}\n")
            f.write("\nNumber of instances per Label:\n")
            for label, cnt in self.label_counts.items():
                f.write(f"{label}: {cnt}\n")
            f.write(f"total: {self.n_instances}\n")
            for metric in self.metrics:
                f.write(f"\nAverage {metric} per Label:\n")
                for label, avg in avg_scores[metric].items():
                    f.write(f"{label}: {avg}\n")

    def run(self):
        """Evaluate all instances in the dataset."""
        for instance in self.dataset:
            log = self._evaluate_element(instance)
            self.logs.append(log)
        self.generate_performance_report(self.performance_filepath)
        if self.log_format is not None:
            self.publisher_log.write(self.logs)
        return self.logs

ariadne_ai/evaluators/rag/context_relevance_evaluator.py
athina-ai/ariadne
from .rag_evaluator import RagEvaluator
from ...loaders.rag_loader import RagLoader
from ...metrics.rag.faithfulness_failure import FaithfulnessFailure
from ...publishers.publisher_log import PublisherLog
from ...llms.rag.faithfulness import Faithfulness


class FaithfulnessEvaluator(RagEvaluator):
    """
    Evaluator for faithfulness in rag chatbot.

    Attributes:
        dataset: Dataset containing instances for evaluation.
        faithfulness_evaluator: Evaluator for faithfulness
        publisher_log: JSON publisher to save the evaluation logs.
        performance_report_filename: txt file to save the perfrormance of a batch
        metrics: List of metrics to evaluate.
        logs: List to accumulate evaluation results for each instance.
    """

    # Chamge metric
    metric_str_to_class = {"faithfulness_failure": FaithfulnessFailure}

    def __init__(
        self,
        loader,
        log_filepath="data/logs/log_rag_faith_eval.json",
        log_format="json",
        performance_filepath="data/logs/perf_rag_faith_eval.txt",
        llm_model="gpt-3.5-turbo",
        metrics=["faithfulness_failure"],
        open_ai_key=None,
    ):
        """
        Initialize the evaluator with given parameters.

        Args:
        - loader: An instance of SummarizationLoader.
        - log_filepath: Path to save the logs.
        - llm_model: Language model to be used.
        - metrics: List of metrics for evaluation.
        """
        if not isinstance(loader, RagLoader):
            raise TypeError("Loader must be an instance of RagLoader")
        # Load data
        self.dataset = loader.processed_dataset
        # Intialize LLMs
        self.llm_model = llm_model
        self.faithfulness_evaluator = Faithfulness(llm_model, open_ai_key)
        # Initialize logging
        self.log_format = log_format
        if log_format is not None:
            self.publisher_log = PublisherLog(log_filepath, log_format)
        self.logs = []
        self.n_instances = 0
        # Intialize metrics
        self.performance_filepath = performance_filepath
        self.metrics = metrics
        self.label_counts = {}
        for metric in metrics:
            setattr(self, f"{metric}_scores", {})

    def _evaluate_element(self, instance):
        """Evaluate an instance for hallucination."""
        context = instance["context"]
        answer = instance["answer"]
        if "label" in instance:
            label = instance["label"]
        else:
            label = "overall"
        # Run LLM Evaluator for faithfulness
        faith_eval = self.faithfulness_evaluator.evaluate(context, answer)

        metric_results = {}
        # Compute metrics
        if faith_eval is None:
            metric_results["evaluation"] = "undefined"
        else:
            for metric in self.metrics:
                metric_class = self.metric_str_to_class.get(metric)
                metric_result, explanation = metric_class.compute(faith_eval)
                metric_results[metric] = metric_result
                metric_results["reason"] = explanation
                self.update_metric_aggr(metric, label, metric_result)
            self.n_instances = self.n_instances + 1
            self.label_counts[label] = self.label_counts.get(label, 0) + 1
        log_instnace = {
            "context": context,
            "answer": answer,
            "label": label,
            **metric_results,
        }
        return log_instnace

    def update_metric_aggr(self, metric, label, aggr_score):
        """Update the aggregated score for a specific metric and label."""
        metric_aggr = getattr(self, f"{metric}_scores", {})
        current_score = metric_aggr.get(label, 0)
        metric_aggr[label] = current_score + aggr_score
        setattr(self, f"{metric}_scores", metric_aggr)

    def get_metric_aggr(self, metric, label):
        """Compute the average scores based on the provided score dictionary."""
        metric_aggr = getattr(self, f"{metric}_scores", {})
        return metric_aggr.get(label, None)

    def get_average_scores(self, score_dict):
        """Compute average scores for a metric"""
        avg_scores = {}
        sum_score = 0
        n_instances = 0
        for label_type, total_score in score_dict.items():
            avg_scores[label_type] = total_score / self.label_counts[label_type]
            sum_score = sum_score + total_score
            n_instances = n_instances + self.label_counts[label_type]
        avg_scores["overall"] = sum_score / n_instances
        return avg_scores

    def compute_average_scores(self):
        """Compute average scores for each metric."""
        avg_scores = {}
        for metric in self.metrics:
            scores = getattr(self, f"{metric}_scores")
            avg_score = self.get_average_scores(scores)
            avg_scores[metric] = avg_score
        return avg_scores

    def generate_performance_report(self, filename):
        """Generate a performance report and save it to the provided filename."""
        avg_scores = self.compute_average_scores()
        with open(filename, "w") as f:
            f.write(f"Model: {self.llm_model}\n")
            f.write("\nNumber of instances per Label:\n")
            for label, cnt in self.label_counts.items():
                f.write(f"{label}: {cnt}\n")
            f.write(f"total: {self.n_instances}\n")
            for metric in self.metrics:
                f.write(f"\nAverage {metric} per Label:\n")
                for label, avg in avg_scores[metric].items():
                    f.write(f"{label}: {avg}\n")

    def run(self):
        """Evaluate all instances in the dataset."""
        for instance in self.dataset:
            log = self._evaluate_element(instance)
            self.logs.append(log)
        self.generate_performance_report(self.performance_filepath)
        if self.log_format is not None:
            self.publisher_log.write(self.logs)
        return self.logs

ariadne_ai/evaluators/rag/faithfulness_evaluator.py
athina-ai/ariadne
from abc import ABC, abstractmethod
from ...loaders.rag_loader import RagLoader

class RagEvaluator(ABC):
    
    @abstractmethod
    def __init__(self, loader:RagLoader, **kwargs):
        pass

    @abstractmethod
    def run(self):
        pass


ariadne_ai/evaluators/rag/rag_evaluator.py
athina-ai/ariadne
from .summarization_evaluator import SummarizationEvaluator
from ...loaders.summarization_loader import SummarizationLoader
from ...metrics.text_summarization.aggreement_score import AgreementScore 
from ...metrics.text_summarization.contradiction_failure import ContradictionFailure
from ...metrics.text_summarization.hallucination_failure import HallucinationfFailure
from ...publishers.publisher_log import PublisherLog
from ...llms.text_summarization.question_generator import QuestionGenerator
from ...llms.text_summarization.question_answerer import QuestionAnswerer

class HallucinationEvaluator(SummarizationEvaluator):
    """
    Evaluator for hallucinations in text summarizations. 

    Attributes:
        dataset: Dataset containing instances for evaluation.
        question_generator: Question generator based on summaries.
        question_answerer: Question answerer for the questions based on documents/summaries.
        publisher_log: JSON publisher to save the evaluation logs.
        performance_report_filename: txt file to save the perfrormance of a batch
        metrics: List of metrics to evaluate.
        logs: List to accumulate evaluation results for each instance.
        n_questions: Number of questions to be generated for each summary.
    """

    metric_str_to_class = {
        'agreement_score': AgreementScore,
        'hallucination_failure': HallucinationfFailure,
        'contradiction_failure': ContradictionFailure
    }

    def __init__(self, loader, questions = None, log_filepath='data/logs/log_sum_hal_eval.json', log_format = 'json', performance_filepath = 'data/logs/perf_sum_hal_eval.txt',  n_questions=5, 
                 llm_model='gpt-3.5-turbo', metrics=[ 'hallucination_failure', 'contradiction_failure', 'agreement_score'], open_ai_key = None):
        """
        Initialize the evaluator with given parameters.

        Args:
        - loader: An instance of SummarizationLoader.
        - log_filepath: Path to save the logs.
        - n_questions: Number of questions to generate for summaries.
        - llm_model: Language model to be used.
        - metrics: List of metrics for evaluation.
        """
        if not isinstance(loader, SummarizationLoader):
            raise TypeError("Loader must be an instance of SummarizationLoader")
        # Load data
        self.dataset = loader.processed_dataset
        # Intialize LLMs
        self.llm_model = llm_model
        self.n_questions = n_questions
        self.questions_defined = None
        if questions is None:
            self.question_generator = QuestionGenerator(llm_model, n_questions,  open_ai_key)
        else:
            self.question_generator = None
            self.questions_defined  = questions
        self.question_answerer = QuestionAnswerer(llm_model, open_ai_key)
        # Initialize logging
        self.log_format = log_format
        if(log_format is not None):
            self.publisher_log = PublisherLog(log_filepath, log_format)
        self.logs = []
        self.n_instances = 0 
        # Intialize metrics
        self.performance_filepath = performance_filepath
        self.metrics = metrics
        self.label_counts ={}
        for metric in metrics:
            setattr(self, f'{metric}_scores', {})

    def _evaluate_element(self, instance):
        """Evaluate an instance for hallucination."""
        document = instance['document']
        summary = instance['summary']
        if 'label' in instance:
            label = instance['label']
        else:
            label = 'overall'
        
        # Generate questions based on summary
        if (self.questions_defined is None):
            questions = self.question_generator.generate(summary)
        # Or load the pre-defined questions:
        else:
            questions = self.questions_defined
        
        # Get answers from document and summary
        answers_doc = self.question_answerer.answer(questions, document)
        answers_sum = self.question_answerer.answer(questions, summary)
        metric_results = {}
        # Compute metrics
        if( answers_doc is None or answers_sum is None or questions is None):
            metric_results['evaluation'] = 'undefined'
        else:
            for metric in self.metrics:
                metric_class = self.metric_str_to_class.get(metric)
                metric_result, explanation, score = metric_class.compute(answers_doc, answers_sum, questions, self.n_questions)
                metric_results[metric] = metric_result
                metric_results[f'reason_{metric}'] = explanation
                self.update_metric_aggr(metric, label, metric_result)
            self.n_instances = self.n_instances +1
            self.label_counts[label] = self.label_counts.get(label, 0) + 1
        return {
            'document': document,
            'summary': summary,
            'questions': questions,
            'answers_doc': answers_doc,
            'answers_sum': answers_sum,
            'label': label,
            **metric_results
        }

    def update_metric_aggr(self, metric, label, aggr_score):
        """ Update the aggregated score for a specific metric and label."""
        metric_aggr = getattr(self, f'{metric}_scores', {})
        current_score = metric_aggr.get(label, 0)
        metric_aggr[label] = current_score + aggr_score
        setattr(self, f'{metric}_scores', metric_aggr)

    def get_metric_aggr(self, metric, label):
        """Compute the average scores based on the provided score dictionary."""
        metric_aggr = getattr(self, f'{metric}_scores', {})
        return metric_aggr.get(label, None)
    
    def get_average_scores(self, score_dict):
        """Compute average scores for a metric"""
        avg_scores = {}
        sum_score = 0 
        n_instances = 0
        for label_type, total_score in score_dict.items(): 
            avg_scores[label_type] = total_score / self.label_counts[label_type]
            sum_score = sum_score + total_score
            n_instances = n_instances + self.label_counts[label_type]
        avg_scores['overall'] = sum_score/n_instances
        return avg_scores
    
    def compute_average_scores(self):
        """Compute average scores for each metric."""
        avg_scores = {}
        for metric in self.metrics:
            scores = getattr(self, f"{metric}_scores")
            avg_score = self.get_average_scores(scores)
            avg_scores[metric] = avg_score
        return avg_scores
        
    def generate_performance_report(self, filename):
        """Generate a performance report and save it to the provided filename."""
        avg_scores = self.compute_average_scores()
        with open(filename, 'w') as f:
            f.write(f"Number of Questions: {self.n_questions}\n")
            f.write(f"Model: {self.llm_model}\n")
            f.write("\nNumber of instances per Label:\n")
            for label, cnt in self.label_counts.items():
                f.write(f"{label}: {cnt}\n")
            f.write(f"total: {self.n_instances}\n")
            for metric in self.metrics:
                f.write(f"\nAverage {metric} per Label:\n")
                for label, avg in avg_scores[metric].items():
                    f.write(f"{label}: {avg}\n")



    def run(self):
        """Evaluate all instances in the dataset."""
        for instance in self.dataset:
            log = self._evaluate_element(instance)
            self.logs.append(log)
        self.generate_performance_report(self.performance_filepath)
        if(self.log_format is not None):
            self.publisher_log.write(self.logs)
        return self.logs

    



    
ariadne_ai/evaluators/text_summarization/hallucination_evaluator.py
athina-ai/ariadne
from .summarization_evaluator import SummarizationEvaluator
from ...loaders.summarization_loader import SummarizationLoader
from ...metrics.text_summarization.aggreement_score import AgreementScore 
from ...metrics.text_summarization.informativeness_failure import InformativenessFailure
from ...publishers.publisher_log import PublisherLog
from ...llms.text_summarization.question_generator import QuestionGenerator
from ...llms.text_summarization.question_answerer import QuestionAnswerer

class InformativenessEvaluator(SummarizationEvaluator):
    """
    Evaluator for informativeness in text summarizations. 

    Attributes:
        dataset: Dataset containing instances for evaluation.
        question_generator: Question generator based on documents.
        question_answerer: Question answerer for the questions based on documents/summaries.
        publisher_log: JSON publisher to save the evaluation logs.
        performance_report_filename: txt file to save the perfrormance of a batch
        metrics: List of metrics to evaluate.
        logs: List to accumulate evaluation results for each instance.
        n_questions: Number of questions to be generated for each summary.
    """

    metric_str_to_class = {
        'agreement_score': AgreementScore,
        'informativeness_failure': InformativenessFailure
    }

    def __init__(self, loader, questions = None, log_filepath='data/logs/log_sum_hal_eval.json', log_format = 'json', performance_filepath = 'data/logs/perf_sum_hal_eval.txt',  n_questions=5, 
                 llm_model='gpt-3.5-turbo', metrics=['agreement_score', 'informativeness_failure'], open_ai_key = None):
        """
        Initialize the evaluator with given parameters.

        Args:
        - loader: An instance of SummarizationLoader.
        - log_filepath: Path to save the logs.
        - n_questions: Number of questions to generate for summaries.
        - llm_model: Language model to be used.
        - metrics: List of metrics for evaluation.
        """
        if not isinstance(loader, SummarizationLoader):
            raise TypeError("Loader must be an instance of SummarizationLoader")
        # Load data
        self.dataset = loader.processed_dataset
        # Intialize LLMs
        self.llm_model = llm_model
        self.n_questions = n_questions
        self.questions_defined = None
        if questions is None:
            self.question_generator = QuestionGenerator(llm_model, n_questions,  open_ai_key)
        else:
            self.question_generator = None
            self.questions_defined  = questions
        self.question_answerer = QuestionAnswerer(llm_model, open_ai_key)
        # Initialize logging
        self.publisher_log = PublisherLog(log_filepath, log_format)
        self.logs = []
        self.n_instances = 0 
        # Intialize metrics
        self.performance_filepath = performance_filepath
        self.metrics = metrics
        self.label_counts ={}
        for metric in metrics:
            setattr(self, f'{metric}_scores', {})

    def _evaluate_element(self, instance):
        """Evaluate an instance for non-informativeness."""
        document = instance['document']
        summary = instance['summary']
        if 'label' in instance:
            label = instance['label']
        else:
            label = 'overall'
        
        # Generate questions based on summary
        if (self.questions_defined is None):
            questions = self.question_generator.generate(summary)
        # Or load the pre-defined questions:
        else:
            questions = self.questions_defined
        
        # Get answers from document and summary
        answers_doc = self.question_answerer.answer(questions, document)
        answers_sum = self.question_answerer.answer(questions, summary)
        metric_results = {}
        # Compute metrics
        if( answers_doc is None or answers_sum is None or questions is None):
            metric_results['evaluation'] = 'undefined'
        else:
            for metric in self.metrics:
                metric_class = self.metric_str_to_class.get(metric)
                metric_result, explanation, score = metric_class.compute(answers_doc, answers_sum, questions, self.n_questions)
                metric_results[metric] = metric_result
                metric_results[f'reason_{metric}'] = explanation
                self.update_metric_aggr(metric, label, metric_result)
            self.n_instances = self.n_instances +1
            self.label_counts[label] = self.label_counts.get(label, 0) + 1
        return {
            'document': document,
            'summary': summary,
            'questions': questions,
            'answers_doc': answers_doc,
            'answers_sum': answers_sum,
            'label': label,
            **metric_results
        }

    def update_metric_aggr(self, metric, label, aggr_score):
        """ Update the aggregated score for a specific metric and label."""
        metric_aggr = getattr(self, f'{metric}_scores', {})
        current_score = metric_aggr.get(label, 0)
        metric_aggr[label] = current_score + aggr_score
        setattr(self, f'{metric}_scores', metric_aggr)

    def get_metric_aggr(self, metric, label):
        """Compute the average scores based on the provided score dictionary."""
        metric_aggr = getattr(self, f'{metric}_scores', {})
        return metric_aggr.get(label, None)
    
    def get_average_scores(self, score_dict):
        """Compute average scores for a metric"""
        avg_scores = {}
        sum_score = 0 
        n_instances = 0
        for label_type, total_score in score_dict.items(): 
            avg_scores[label_type] = total_score / self.label_counts[label_type]
            sum_score = sum_score + total_score
            n_instances = n_instances + self.label_counts[label_type]
        avg_scores['overall'] = sum_score/n_instances
        return avg_scores
    
    def compute_average_scores(self):
        """Compute average scores for each metric."""
        avg_scores = {}
        for metric in self.metrics:
            scores = getattr(self, f"{metric}_scores")
            avg_score = self.get_average_scores(scores)
            avg_scores[metric] = avg_score
        return avg_scores
        
    def generate_performance_report(self, filename):
        """Generate a performance report and save it to the provided filename."""
        avg_scores = self.compute_average_scores()
        with open(filename, 'w') as f:
            f.write(f"Number of Questions: {self.n_questions}\n")
            f.write(f"Model: {self.llm_model}\n")
            f.write("\nNumber of instances per Label:\n")
            for label, cnt in self.label_counts.items():
                f.write(f"{label}: {cnt}\n")
            f.write(f"total: {self.n_instances}\n")
            for metric in self.metrics:
                f.write(f"\nAverage {metric} per Label:\n")
                for label, avg in avg_scores[metric].items():
                    f.write(f"{label}: {avg}\n")



    def run(self):
        """Evaluate all instances in the dataset."""
        for instance in self.dataset:
            log = self._evaluate_element(instance)
            self.logs.append(log)
        self.generate_performance_report(self.performance_filepath)
        self.publisher_log.write(self.logs)

    



    
ariadne_ai/evaluators/text_summarization/informativeness_evaluator.py
athina-ai/ariadne
from abc import ABC, abstractmethod
from ...loaders.summarization_loader import SummarizationLoader

class SummarizationEvaluator(ABC):
    
    @abstractmethod
    def __init__(self, loader:SummarizationLoader, **kwargs):
        pass

    @abstractmethod
    def run(self):
        pass


ariadne_ai/evaluators/text_summarization/summarization_evaluator.py
athina-ai/ariadne
import openai
import time
from dotenv import load_dotenv

import os
import json

class OpenAICompletion:
    """
    A class to interact with OpenAI's ChatCompletion API.
    
    Attributes:
    - model (str): The model to use for completions, default is "gpt-3.5-turbo".
    - open_ai_key (str): The API key for OpenAI.
    - temperature (float): OpenAI temperature setting.
    - max_tokens (int): OpenAI maximum number of tokens setting. The token count of your prompt plus max_tokens cannot exceed the model's context.
    """
    def __init__(self, model, open_ai_key):
        """
        Initializes the OpenAICompletion with the provided settings.
        """
        # Setting instance attributes based on provided parameters or defaults
        load_dotenv()
        self.model = model
        if open_ai_key is None:
            self.open_ai_key = os.getenv("OPENAI_API_KEY")
        else:
            self.open_ai_key = open_ai_key
        
        # Setting the API key for OpenAI based on provided key
        openai.api_key = self.open_ai_key

    def get_completion_from_messages(self, messages, temperature=0, max_tokens = 2000):
        """
        Fetches a completion response from OpenAI's ChatCompletion API based on the provided messages.
        """
        try:
            # Attempting to fetch a response from OpenAI
            response = openai.ChatCompletion.create(
                model=self.model,
                messages=messages,
                temperature=temperature,
                max_tokens=max_tokens,
            )
        except openai.error.RateLimitError:
            # In case of a rate limit error, wait for 15 seconds and retry
            time.sleep(15)
            print('RateLimitError')
            return self.get_completion_from_messages(messages)
        except openai.error.AuthenticationError:
            raise openai.error.AuthenticationError("Please pass a valid OpenAi key.")
        except openai.error.Timeout:
            # In case of a rate limit error, wait for 15 seconds and retry
            time.sleep(15)
            print('Timeout')
            return self.get_completion_from_messages(messages)
        except openai.error.InvalidRequestError:
            print('InvalidRequestError')
            return None
        except openai.error.APIConnectionError:
            # In case of a api connection error, wait for 60 seconds and retry
            time.sleep(60)
            print('APIConnectionError')
            return self.get_completion_from_messages(messages)
        return response.choices[0].message["content"]

    @staticmethod
    def _extract_json(data_string: str) -> str:
        """
        Extracts a JSON string from a larger string. 
        Assumes the JSON content starts with '{' and continues to the end of the input string.
        """
        try:
            start_index = data_string.index("{")
            json_string = data_string[start_index:]
        except:
            json_string = data_string
        return json_string
    
    @staticmethod
    def _load_json_from_text(text):
        """
        Extracts and loads a JSON string from a given text.
        """
        try:
            data = json.loads(text)
        except json.decoder.JSONDecodeError:
            data = None
        return data
    
    @staticmethod
    def extract_json_from_response(response):
        # In case you cannot handle an error, return None
        if response is None:
            return None
        response_json_format = OpenAICompletion._extract_json(response)
        response_json = OpenAICompletion._load_json_from_text(response_json_format)
        return(response_json)
ariadne_ai/llms/open_ai_completion.py
athina-ai/ariadne
from ..open_ai_completion import OpenAICompletion


class FewShotExampleAnswertRelevance:
    """
    Class represting an example of the evaluation that could be used for few-shot prompting.
    """

    # User's question
    query: str
    # User's question
    response: str
    # Name of the evaluation function
    eval_function: str
    # Evaluation result
    eval_result: str
    # LLM's reason for evaluation
    eval_reason: str

    def __init__(
        self,
        query: str,
        response: str,
        eval_function: str,
        eval_result: str,
        eval_reason: str,
    ):
        """
        Initialize a new instance of FewShotExample.
        """
        self.query = query
        self.response = response
        self.eval_function = eval_function
        self.eval_result = eval_result
        self.eval_reason = eval_reason

    def __str__(self):
        """
        Return a string representation of the FewShotExample.
        """
        return (
            f"Query: {self.query}\n"
            f"Response: {self.response}\n"
            f"{self.eval_function}: {self.eval_result}\n"
            f"Reason:{self.eval_reason}"
        )


class AnswerRelevance:
    """
    This class determines whether the chatbot's response answers specifically what the user is asking about, and covers all aspects of the user's query

    Attributes:
        openAIcompletion (OpenAICompletion): Instance for interactions with OpenAI's API.
        examples (list[FewShotExampleFaithfulness]): List of few-shot examples used for evaluation.
    """

    SYSTEM_MESSAGE = """ 
        You are an expert at evaluating whether a response answers a user's query sufficiently.
    """

    USER_MESSAGE_TEMPLATE = """
        Let's think step by step.
        1. Consider the following: 
        user's query: {}.
        response:{}.
        2. Determine if the response answers specifically what the user is asking about, and covers all aspects of the user's query.
        3. Provide a brief explanation of why the response does or does not answer the user's query sufficiently, labeled as 'explanation', leading up to a verdict (Yes/No) labeled as 'verdict'.
        4. Return a JSON object in the following format: "verdict": 'verdict', "explanation": 'explanation'.

        Here's are some examples: 
        {}
    """

    def __init__(self, model, open_ai_key):
        """
        Initialize the QuestionAnswerer class.
        """
        self.openAIcompletion = OpenAICompletion(model, open_ai_key)
        self.examples = self.get_few_shot_examples()

    def evaluate(self, query: str, response: str):
        """
        Evaluation for is response faithful to context
        """
        user_message = self.USER_MESSAGE_TEMPLATE.format(query, response, self.examples)
        system_message = self.SYSTEM_MESSAGE
        message = [
            {"role": "system", "content": system_message},
            {"role": "user", "content": user_message},
        ]

        openai_response = self.openAIcompletion.get_completion_from_messages(message)
        openai_response_json = self.openAIcompletion.extract_json_from_response(
            openai_response
        )
        return openai_response_json

    # Few shot examples
    @staticmethod
    def get_few_shot_examples():
        """
        Returns the few-shot examples.
        """
        # Creating instances of the FewShotExampleCcei class for each example
        example1 = FewShotExampleAnswertRelevance(
            query="How much does Y Combinator invest in startups",
            response="125,000",
            eval_function="does_response_answer_query",
            eval_result="Yes",
            eval_reason="The response is a reasonable answer to the query",
        )
        example2 = FewShotExampleAnswertRelevance(
            query="What was the name of the spaceship to first land on the moon",
            response="Neil Armstrong was the first astronaut on the moon",
            eval_function="does_response_answer_query",
            eval_result="No",
            eval_reason="The response does not answer the query asking about the name of the spaceship.",
        )
        example3 = FewShotExampleAnswertRelevance(
            query="Will alicia keys be at the festival",
            response="Neil Armstrong was the first astronaut on the moon",
            eval_function="does_response_answer_query",
            eval_result="Yes",
            eval_reason="The response is a reasonable answer to the query.",
        )
        # Joining the string representations of the instances
        examples = "\n\n".join([str(example1), str(example2), str(example3)])
        return examples

ariadne_ai/llms/rag/answer_relevance.py
athina-ai/ariadne
from ..open_ai_completion import OpenAICompletion


class FewShotExampleContextRelevance:
    """
    Class represting an example of the evaluation that could be used for few-shot prompting.
    """

    # Retrieved context
    context: str
    # User's question
    query: str
    # Name of the evaluation function
    eval_function: str
    # Evaluation result
    eval_result: str
    # LLM's reason for evaluation
    eval_reason: str

    def __init__(
        self,
        context: str,
        query: str,
        eval_function: str,
        eval_result: str,
        eval_reason: str,
    ):
        """
        Initialize a new instance of FewShotExample.
        """
        self.context = context
        self.query = query
        self.eval_function = eval_function
        self.eval_result = eval_result
        self.eval_reason = eval_reason


class ContextRelevance:
    """
    This class determines whether the chatbot's response can be inferred using only the information provided as context.

    Attributes:
        openAIcompletion (OpenAICompletion): Instance for interactions with OpenAI's API.
        examples (list[FewShotExampleFaithfulness]): List of few-shot examples used for evaluation.
    """

    SYSTEM_MESSAGE = """ 
        You are an expert at evaluating whether a chatbot can answer a user's query using ONLY the information provided to you as context.
    """

    USER_MESSAGE_TEMPLATE = """
        Let's think step by step.
        1. Consider the following: 
        user's query: {}.
        context:{}.
        2. Determine if the chatbot can answer the user's query with nothing but the "context" information provided to you.
        3. Provide a brief explanation of why the context does or does not contain sufficient information, labeled as 'explanation', leading up to a verdict (Yes/No) labeled as 'verdict'.
        4. Return a JSON object in the following format: "verdict": 'verdict', "explanation": 'explanation'.

        Here's are some examples: 
        {}
    """

    def __init__(self, model, open_ai_key):
        """
        Initialize the QuestionAnswerer class.
        """
        self.openAIcompletion = OpenAICompletion(model, open_ai_key)
        self.examples = self.get_few_shot_examples()

    def evaluate(self, query: str, context: str):
        """
        Evaluation for is response faithful to context
        """
        user_message = self.USER_MESSAGE_TEMPLATE.format(query, context, self.examples)
        system_message = self.SYSTEM_MESSAGE
        message = [
            {"role": "system", "content": system_message},
            {"role": "user", "content": user_message},
        ]

        openai_response = self.openAIcompletion.get_completion_from_messages(message)
        openai_response_json = self.openAIcompletion.extract_json_from_response(
            openai_response
        )
        return openai_response_json

    # Few shot examples
    @staticmethod
    def get_few_shot_examples():
        """
        Returns the few-shot examples.
        """
        # Creating instances of the FewShotExampleContextRelevance class for each example
        example1 = FewShotExampleContextRelevance(
            context="bjarne stroustrup invented C++",
            query="Who invented the linux os",
            eval_function="does_context_contain_sufficient_information",
            eval_result="No",
            eval_reason="The context does not provide any relevant information about the Linux OS or its inventor.",
        )
        example2 = FewShotExampleContextRelevance(
            context="In 1969, Neil Armstrong became the first person to walk on the moon.",
            query="What was the name of the spaceship used for the moon landing in 1969?",
            eval_function="does_context_contain_sufficient_information",
            eval_result="No",
            eval_reason="The context provided does not include any information about the name of the spaceship used for the moon landing. The query specifically asks for the name of the spaceship, which is not present in the context.",
        )
        # Joining the string representations of the instances
        examples = "\n\n".join([str(example1), str(example2)])
        return examples

ariadne_ai/llms/rag/context_relevance.py
athina-ai/ariadne
from ..open_ai_completion import OpenAICompletion


class FewShotExampleFaithfulness:
    """
    Class represting an example of the evaluation that could be used for few-shot prompting.
    """

    # User's question
    context: str
    # User's question
    response: str
    # Name of the evaluation function
    eval_function: str
    # Evaluation result
    eval_result: str
    # LLM's reason for evaluation
    eval_reason: str

    def __init__(
        self,
        context: str,
        response: str,
        eval_function: str,
        eval_result: str,
        eval_reason: str,
    ):
        """
        Initialize a new instance of FewShotExample.
        """
        self.context = context
        self.response = response
        self.eval_function = eval_function
        self.eval_result = eval_result
        self.eval_reason = eval_reason


class Faithfulness:
    """
    This class determines whether the chatbot's answer hether the response can be inferred using only the information provided as context.

    Attributes:
        openAIcompletion (OpenAICompletion): Instance for interactions with OpenAI's API.
        examples (list[FewShotExampleFaithfulness]): List of few-shot examples used for evaluation.
    """

    # Pre-defined prompts for OpenAI's GPT model
    SYSTEM_MESSAGE = """ 
        You are an expert at evaluating whether the response can be inferred using ONLY the information provided as context.
    """

    USER_MESSAGE_TEMPLATE = """
        Let's think step by step.
        1. Consider the following: 
        context: {}.
        response:{}.
        2. Determine if the response can be inferred purely from the context provided.
        3. Provide a brief explanation of what information the response contained that was not provided to it in the context, labeled as 'explanation', leading up to a verdict (Yes/No) labeled as 'verdict'.
        4. Return a JSON object in the following format: "verdict": 'verdict', "explanation": 'explanation'.

        Here's are some examples: 
        {}
    """

    def __init__(self, model, open_ai_key):
        """
        Initialize the QuestionAnswerer class.
        """
        self.openAIcompletion = OpenAICompletion(model, open_ai_key)
        self.examples = self.get_few_shot_examples()

    def evaluate(self, context: str, response: str):
        """
        Evaluation for is response faithful to context
        """
        user_message = self.USER_MESSAGE_TEMPLATE.format(
            context, response, self.examples
        )
        system_message = self.SYSTEM_MESSAGE
        message = [
            {"role": "system", "content": system_message},
            {"role": "user", "content": user_message},
        ]

        openai_response = self.openAIcompletion.get_completion_from_messages(message)
        openai_response_json = self.openAIcompletion.extract_json_from_response(
            openai_response
        )
        return openai_response_json

    @staticmethod
    def get_few_shot_examples():
        """
        Returns the few-shot examples.
        """
        # Creating instances of the FewShotExampleCcei class for each example
        example1 = FewShotExampleFaithfulness(
            context="Y Combinator is a startup accelerator launched in March 2005. It has been used to launch more than 4,000 companies",
            response="125,000",
            eval_function="is_response_faithful_to_context",
            eval_result="No",
            eval_reason="The context does not contain any information to substantiate the response.",
        )

        # Joining the string representations of the instances
        examples = "\n\n".join([str(example1)])
        return examples

ariadne_ai/llms/rag/faithfulness.py
athina-ai/ariadne
from ..open_ai_completion import OpenAICompletion

class QuestionAnswerer:
    """
    This class determines whether the chatbot's answer was correct based on 
    the given content and user's question. 

    Attributes:
        openAIcompletion (OpenAICompletion): Instance for interactions with OpenAI's API.
    """

    # Pre-defined prompts for OpenAI's GPT model
    SYSTEM_MESSAGE = """ 
        You are an expert at responding to closed-ended (Yes/No) questions using ONLY the provided context.
    """

    USER_MESSAGE_TEMPLATE = """
        Let's think step by step.
        1. Consider the following: 
           Questions: {}.
           Context: {}.
        2. Respond to each question from the provided 'questions', using either 
           'Yes', 'No', or 'Unknown', based on the given context.
        3. Return a JSON object in the following format: "question1": "answer1", "question2": "answer2",...
    """

    def __init__(self, model, open_ai_key):
        """
        Initialize the QuestionAnswerer class.
        """
        self.openAIcompletion = OpenAICompletion(model, open_ai_key)

    def answer(self, questions: str, context: str) -> dict:
        """
        Respond to each question from the provided 'questions' given the context.

        Args:
            questions (str): A set of questions posed to the chatbot.
            context (str): Context used to inform the chatbot's answers.

        Returns:
            dict: Evaluation results formatted as a dictionary with questions as keys and 
                  'Yes', 'No', or 'Unknown' as values.
        """
        
        user_message = self.USER_MESSAGE_TEMPLATE.format(questions, context)
        message = [
            {'role': 'system', 'content': self.SYSTEM_MESSAGE}, 
            {'role': 'user', 'content': user_message}
        ]
        
        openai_response = self.openAIcompletion.get_completion_from_messages(message)
        openai_response_json = self.openAIcompletion.extract_json_from_response(openai_response)
        
        return openai_response_json

ariadne_ai/llms/text_summarization/question_answerer.py
athina-ai/ariadne
from ..open_ai_completion import OpenAICompletion

class QuestionGenerator:
    """
    Generates closed-ended (Yes/No) questions given a  text.
    
    Attributes:
        n_questions (int): Number of questions to generate.
        openAIcompletion (OpenAICompletion): Instance for interactions with OpenAI's API.
    """

    # Pre-defined prompts for OpenAI's GPT model
    SYSTEM_MESSAGE = """ 
        You are an expert at generating closed-ended (Yes/No) questions given the content of a text.
    """

    USER_MESSAGE_TEMPLATE = """
        Let's think step by step.
        1. Consider the text: {}.
        2. Generate {} closed-ended (Yes/No) questions based on the content.
        3. Return a JSON object in the following format: "question 1": 'Your question', "question 2": 'Your next question', ...
    """

    def __init__(self, model: str, n_questions: int, open_ai_key:str):
        """
        Initialize the QuestionGenerator.
        """
        self.n_questions = n_questions
        self.openAIcompletion = OpenAICompletion(model, open_ai_key)

    def generate(self, text: str) -> dict:
        """
        Generate a set of closed-ended questions based on the provided text.

        Args:
            text (str): The reference content used to generate questions.

        Returns:
            dict: A dictionary of generated questions with keys indicating the question order and values being the questions themselves.
        """
        user_message = self.USER_MESSAGE_TEMPLATE.format(text, self.n_questions)
        message = [
            {'role': 'system', 'content': self.SYSTEM_MESSAGE}, 
            {'role': 'user', 'content': user_message}
        ]

        openai_response = self.openAIcompletion.get_completion_from_messages(message)
        openai_response_json = self.openAIcompletion.extract_json_from_response(openai_response)

        return openai_response_json

ariadne_ai/llms/text_summarization/question_generator.py
athina-ai/ariadne
from abc import ABC, abstractmethod
import json


class Loader(ABC):
    """Abstract base class for data loaders."""

    @abstractmethod
    def load(self, filename: str) -> None:
        """Load data in the specified format."""
        pass

    @abstractmethod
    def process(self) -> None:
        """Prepare dataset to be consumed by evaluators."""
        pass

ariadne_ai/loaders/loader.py
athina-ai/ariadne
from .loader import Loader
import json

class RagLoader(Loader):
    """
    This class is a data loader for retrieval augmented generation (RAG) datasets.

    Attributes:
        col_question (str): The column name corresponding to the user's question.
        col_context (str): The column name corresponding to the retrieved context.
        col_answer (str): The column name corresponding to the answer.
        col_label (str, optional): The column name corresponding to labels, if any.
        col_comment (str, optional): The column name corresponding to additional comments, if any.
        raw_dataset (dict): The raw dataset as loaded from the source.
        processed_dataset (list): The processed dataset with queries, context, response and other attributes if present.
    """
    
    def __init__(self, col_question ='question',  col_context='context', col_answer = 'answer',
                 col_label=None, col_comment=None, format = 'json'):
        """ 
        Initializes the loader with specified or default column names.
        """
        self.col_question = col_question
        self.col_context = col_context
        self.col_answer = col_answer
        self.col_label = col_label
        self.col_comment = col_comment
        self._raw_dataset = {}
        self._processed_dataset = []
        self.format = format

    @property
    def processed_dataset(self):
        """ 
        Returns the processed dataset.
        """
        return self._processed_dataset
    
    @property
    def raw_dataset(self):
        """ 
        Returns the raw dataset.
        """
        return self._raw_dataset

    def process(self) -> None:
        """
        Transforms the raw data into a structured format. Processes each entry from the raw dataset, and extracts attributes.
        
        Raises:
            KeyError: If mandatory columns (question, context or response) are missing in the raw dataset.
            KeyError: If optional columns (label or comment) are missing in the raw dataset if defined.
        """
        for raw_instance in self._raw_dataset:
            # Check for mandatory columns in raw_instance
            if self.col_question not in raw_instance:
                raise KeyError(f"'{self.col_question}' not found in provided data.")
            if self.col_context not in raw_instance:
                raise KeyError(f"'{self.col_context}' not found in provided data.")
            if self.col_answer not in raw_instance:
                raise KeyError(f"'{self.col_answer}' not found in provided data.")
            # Create a processed instance with mandatory fields
            processed_instance = {
                'question': raw_instance[self.col_question],
                'context': raw_instance[self.col_context],
                'answer': raw_instance[self.col_answer]
            }

           # Add optional attributes if they exist
            if self.col_label is not None and self.col_label in raw_instance:
                processed_instance['label'] = raw_instance[self.col_label]
            if self.col_comment is not None and self.col_comment in raw_instance:
                processed_instance['comment'] = raw_instance[self.col_comment]

            # Store the results
            self._processed_dataset.append(processed_instance)

    def load_json(self, filename: str) -> None:
        """
        Loads and processes data from a JSON file.
        
        Raises:
            FileNotFoundError: If the specified JSON file is not found.
            json.JSONDecodeError: If there's an issue decoding the JSON.
        """
        try:
            with open(filename, 'r') as f:
                self._raw_dataset = json.load(f)
                self.process()
        except (FileNotFoundError, json.JSONDecodeError) as e:
            print(f"Error loading JSON: {e}")
    
    def load_dict(self, data: list) -> None:
        """
        Loads and processes data from a list of dictionaries.
        """
        self._raw_dataset = data
        self.process()

    def load_response(self, question, context, answer, comment = None, label = None) -> None:
        """
        Loads and processes a response of text summarization.
        """
        raw_instance = {'question':question, 'context': context, 'answer': answer}
        if comment is not None:
            raw_instance['comment'] =comment
        if label is not None:
            raw_instance['label'] =comment
        self._raw_dataset = [raw_instance]
        self.process()

    def load(self, data: list) -> None:
        """
        Loads data based on the format specified.
        """
        if self.format == 'json':
            self.load_json(data)
        elif self.format == 'dict':
            self.load_dict(data)
        else:
            raise NotImplementedError("This file format has not been supported yet.")
    
        
    def load_csv(self) -> None:
        """
        Placeholder for loading data from a CSV file.

        Raises:
            NotImplementedError: This method has not been implemented yet.
        """
        raise NotImplementedError("This method has not been implemented yet.")

    def load_pandas(self) -> None:
        """
        Placeholder for loading data from a pandas DataFrame.

        Raises:
            NotImplementedError: This method has not been implemented yet.
        """
        raise NotImplementedError("This method has not been implemented yet.")

   


ariadne_ai/loaders/rag_loader.py
athina-ai/ariadne
from .loader import Loader
import json

class SummarizationLoader(Loader):
    """
    A data loader for text summarization datasets.


    Attributes:
        col_document (str): Column name corresponding to the main document text.
        col_summary (str): Column name corresponding to the summary of the document.
        col_label (str, optional): Column name corresponding to labels, if any.
        col_comment (str, optional): Column name corresponding to additional comments, if any.
        raw_dataset (dict): Raw dataset as loaded from the source.
        processed_dataset (list): Processed dataset with documents, summaries, and other attributes if present.
    """
    
    def __init__(self, col_document='document', col_summary='summary', 
                 col_label=None, col_comment=None, format = 'json'):
        """ Initializes the loader with specified or default column names."""
        self.col_document = col_document
        self.col_summary = col_summary
        self.col_label = col_label
        self.col_comment = col_comment
        self._raw_dataset = {}
        self._processed_dataset = []
        self.format = format

    @property
    def processed_dataset(self):
        """ Returns the processed dataset."""
        return self._processed_dataset
    
    @property
    def raw_dataset(self):
        """ Returns the raw dataset."""
        return self._raw_dataset

    def process(self) -> None:
        """
        Transforms the raw data into a structured format. Processes each entry from the raw dataset, and extracts attributes
        
        Raises:
            KeyError: If mandatory columns (document or summary) are missing in the raw dataset.
            KeyError: If optional columns (label or comment) are missing in the raw dataset if defined.
        """
        for raw_instance in self._raw_dataset:
            # Check for mandatory columns in raw_instance
            if self.col_document not in raw_instance:
                raise KeyError(f"'{self.col_document}' not found in provided data.")
            if self.col_summary not in raw_instance:
                raise KeyError(f"'{self.col_summary}' not found in provided data.")
            # Create a processed instance with mandatory fields
            processed_instance = {
                'document': raw_instance[self.col_document],
                'summary': raw_instance[self.col_summary]
            }

           # Add optional attributes if they exist
            if self.col_label is not None and self.col_label in raw_instance:
                processed_instance['label'] = raw_instance[self.col_label]
            if self.col_comment is not None and self.col_comment in raw_instance:
                processed_instance['comment'] = raw_instance[self.col_comment]

            # Store the results
            self._processed_dataset.append(processed_instance)

    def load_json(self, filename: str) -> None:
        """
        Loads and processes data from a JSON file.
        
        Raises:
            FileNotFoundError: If the specified JSON file is not found.
            json.JSONDecodeError: If there's an issue decoding the JSON.
        """
        try:
            with open(filename, 'r') as f:
                self._raw_dataset = json.load(f)
                self.process()
        except (FileNotFoundError, json.JSONDecodeError) as e:
            print(f"Error loading JSON: {e}")
    
    def load_dict(self, data: list) -> None:
        """Loads and processes data from a list of dictionaries."""
        self._raw_dataset = data
        self.process()

    def load_response(self, document, summary, comment = None, label = None) -> None:
        """Loads and processes a response of text summarization"""
        raw_instance = {'document': document, 'summary': summary}
        if comment is not None:
            raw_instance['comment'] =comment
        if label is not None:
            raw_instance['label'] =comment
        self._raw_dataset = [raw_instance]
        self.process()

    def load(self, data: list) -> None:
        """" """
        if self.format == 'json':
            self.load_json(data)
        elif self.format == 'dict':
            self.load_dict(data)
        else:
            raise NotImplementedError("This file format has not been supported yet.")
    
        
    def load_csv(self) -> None:
        """
        Placeholder for loading data from a CSV file.

        Raises:
            NotImplementedError: This method has not been implemented yet.
        """
        raise NotImplementedError("This method has not been implemented yet.")

    def load_pandas(self) -> None:
        """
        Placeholder for loading data from a pandas DataFrame.

        Raises:
            NotImplementedError: This method has not been implemented yet.
        """
        raise NotImplementedError("This method has not been implemented yet.")

   

ariadne_ai/loaders/summarization_loader.py
athina-ai/ariadne
from abc import ABC, abstractmethod

class Metric(ABC):
    """
    Abstract base class for evaluation metrics.
    """

    @abstractmethod
    def compute(self):
        """
        Computes the metric.
        """
        pass
ariadne_ai/metrics/metric.py
athina-ai/ariadne
from ..metric import Metric

class AnswerRelevanceFailure(Metric):
    """
    The AnswerRelevanceFailure class is a metric that determines whether a response answers a user's query sufficiently.
    """

    @staticmethod
    def verdict_to_int(verdict: str) -> int:
        """
        Converts the verdict to an integer score. 'yes' verdict is considered non-failure, while 'no' verdict is considered failure.
        """
        verdict = verdict.lower()
        score = 1 if verdict == 'no' else 0 if verdict == 'yes' else None
        return score
    
    @staticmethod
    def compute(answer_relevance_eval):
        """
        Computes the context relevance metric.
        """
        is_answer_relevance_failure = AnswerRelevanceFailure.verdict_to_int(answer_relevance_eval['verdict'])
        explanation = answer_relevance_eval['explanation']
        return is_answer_relevance_failure, explanation
ariadne_ai/metrics/rag/answer_relevance_failure.py
athina-ai/ariadne
from ..metric import Metric

class ContextRelevanceFailure(Metric):
    """
    The ContextRelevanceFailure class is a metric that determines if  a chatbot can answer a user's query using ONLY the information provided to you as context.
    """

    @staticmethod
    def verdict_to_int(verdict: str) -> int:
        """
        Converts the verdict to an integer score. 'yes' verdict is considered non-failure, while 'no' verdict is considered failure.
        """
        verdict = verdict.lower()
        score = 1 if verdict == 'no' else 0 if verdict == 'yes' else None
        return score
    
    @staticmethod
    def compute(context_relevance_eval):
        """
        Computes the context relevance metric.
        """
        is_context_relevance_failure = ContextRelevanceFailure.verdict_to_int(context_relevance_eval['verdict'])
        explanation = context_relevance_eval['explanation']
        return is_context_relevance_failure, explanation
ariadne_ai/metrics/rag/context_relevance_failure.py
athina-ai/ariadne
from ..metric import Metric


class FaithfulnessFailure(Metric):
    """
    The Faithfulness class is a metric that determines if the response can be inferred purely from the context provided.
    """

    @staticmethod
    def verdict_to_int(verdict: str) -> int:
        """
        Converts the verdict to an integer score. 'yes' verdict is considered non-failure, while 'no' verdict is considered failure.
        """
        verdict = verdict.lower()
        score = 1 if verdict == "no" else 0 if verdict == "yes" else None
        return score

    @staticmethod
    def compute(faith_eval):
        """
        Computes the faithfulness metric.
        """
        is_faithfulness_failure = FaithfulnessFailure.verdict_to_int(
            faith_eval["verdict"]
        )
        explanation = faith_eval["explanation"]
        return is_faithfulness_failure, explanation

ariadne_ai/metrics/rag/faithfulness_failure.py
athina-ai/ariadne
from ..metric import Metric

class AgreementScore(Metric):
    """
    Calculates agreement score between two sets of answers.

    AgreementScore computes the proportion of questions that received 
    consistent answers between a source (e.g., document) and a summary.
    """

    @staticmethod
    def _compute_metric(answers_src, answers_sum, questions):
        """
        Computes the number of matches between the answers from source and summary.

        Args:
            answers_src (dict): Answers derived from the source.
            answers_sum (dict): Answers derived from the summary.

        Returns:
            int: Number of questions with consistent answers.
        """
        answers_src_ls = list(answers_src.values())
        answers_sum_ls = list(answers_sum.values())
        n_matches = 0
        aggr_questions = []
        for idx, (ans_src, ans_sum) in enumerate(zip(answers_src_ls, answers_sum_ls)):
            if ans_src.strip().lower() == ans_sum.strip().lower():
                n_matches += 1
                aggr_question = questions[f"question {idx+1}"]
                aggr_questions.append(f'{aggr_question}')
        return n_matches,aggr_questions

    @staticmethod
    def compute(answers_src, answers_sum, questions, n_questions):
        """
        Computes the agreement score.

        Args:
            answers_src (dict): Answers derived from the source.
            answers_sum (dict): Answers derived from the summary.
            n_questions (int): Total number of questions.

        Returns:
            float: Agreement score.
        """
        n_matches, aggr_questions = AgreementScore._compute_metric(answers_src, answers_sum, questions) 
        explanation = aggr_questions
        aggr_score = n_matches / n_questions
        return aggr_score, explanation, None
    


   
ariadne_ai/metrics/text_summarization/aggreement_score.py
athina-ai/ariadne
from ..metric import Metric

class ContradictionFailure(Metric):
    """
    Metric to evaluate the degree of contradiction between the answers obtained from 
    a summary and the original document. It captures the percentage of questions that 
    received contradictory answers between the summary and the document, with neither 
    being 'Unknown'. A high score suggests the summary might be contradicting the 
    original document's content.
    
    Attributes:
        answers_src (dict): Answers derived from the original document.
        answers_sum (dict): Answers derived from the summary.
        n_questions (int): Number of questions posed.
    """
    
    @staticmethod
    def _compute_metric(answers_src, answers_sum, questions):
        """
        Compute the number of contradictions between answers derived from the document 
        and the summary.
        
        Args:
            answers_src (dict): Answers based on the original document.
            answers_sum (dict): Answers based on the summary.
        
        Returns:
            int: Number of contradictions.
        """
        answers_src_ls = list(answers_src.values())
        answers_sum_ls = list(answers_sum.values())

        n_contradiction = 0
        cont_questions = []

        for idx, (ans_src, ans_sum) in enumerate(zip(answers_src_ls, answers_sum_ls)):
            if ans_src.strip().lower() in ['yes', 'no'] and  ans_src.strip().lower() != ans_sum.strip().lower():
                n_contradiction += 1
                cont_question = questions[f"question {idx+1}"]
                cont_questions.append(f'{cont_question}')

        return n_contradiction,cont_questions


    @staticmethod
    def compute(answers_src, answers_sum, questions, n_questions):
        """
        Compute the contradiction score by normalizing the number of contradictions by 
        the total number of questions.
        
        Args:
            answers_src (dict): Answers based on the original document.
            answers_sum (dict): Answers based on the summary.
            n_questions (int): Total number of questions.
        
        Returns:
            float: Contradiction score.
        """
        n_contradiction, cont_questions = ContradictionFailure._compute_metric(answers_src, answers_sum, questions)
        is_contradiction_failure = 1 if n_contradiction > 0 else 0 
        explanation = cont_questions
        cont_score = n_contradiction / n_questions
        return is_contradiction_failure, explanation,cont_score



ariadne_ai/metrics/text_summarization/contradiction_failure.py
athina-ai/ariadne
from ..metric import Metric

class HallucinationfFailure(Metric):
    """
    Calculates the hallucination score between two sets of answers.

    HallucinationScore computes the proportion of summaries where a question generated 
    from the summary receives a 'Yes/No' answer from the summary, but an 'Unknown' answer 
    from the source document. A high score indicates potential content in the summary 
    that is absent from the source document.
    """

    @staticmethod
    def _compute_metric(answers_src, answers_sum, questions):
        """
        Computes the number of hallucinations between the answers from source and summary.

        Args:
            answers_src (dict): Answers derived from the source.
            answers_sum (dict): Answers derived from the summary.

        Returns:
            int: Number of questions indicating hallucinations.
        """
        answers_src_ls = list(answers_src.values())
        answers_sum_ls = list(answers_sum.values())

        halu_questions =[]
        n_hallucination = 0

        for idx, (ans_src, ans_sum) in enumerate(zip(answers_src_ls, answers_sum_ls)):
            if ans_src.strip().lower() == 'unknown' and ans_sum.strip().lower() in ['yes', 'no']:
                n_hallucination += 1
                halu_question = questions[f"question {idx+1}"]
                halu_questions.append(f'{halu_question}')

        return n_hallucination,halu_questions

    @staticmethod
    def compute(answers_src, answers_sum, questions, n_questions):
        """
        Computes the hallucination score.

        Args:
            answers_src (dict): Answers derived from the source.
            answers_sum (dict): Answers derived from the summary.
            questions (dict): Questions generated from the summary.
            n_questions (int): Total number of questions.

        Returns:
            float: Hallucination score.
        """
        n_hallucination, halu_questions = HallucinationfFailure._compute_metric(answers_src, answers_sum, questions)
        is_hallucination_failure = 1 if n_hallucination > 0 else 0 
        halu_score = n_hallucination / n_questions
        explanation = halu_questions
        return is_hallucination_failure, explanation, halu_score

ariadne_ai/metrics/text_summarization/hallucination_failure.py
athina-ai/ariadne
from ..metric import Metric

class InformativenessFailure(Metric):
    """
    Calculates noninformativeness score between two sets of answers.

    Informativeness computes the proportion of questions that received 
    received a 'Unknown' summary-based answer and an 'Yes/No' document-based answer.
    """

    @staticmethod
    def _compute_metric(answers_src, answers_sum, questions):
        """
        Computes the number of questions with 'Unknown' summary-based answer and an 'Yes/No' document-based answer.

        Args:
            answers_src (dict): Answers derived from the source.
            answers_sum (dict): Answers derived from the summary.

        Returns:
            int: Number of questions with questions with 'Unknown' summary-based answer and an 'Yes/No' document-based answer.
        """
        # get answers
        answers_src_ls = list(answers_src.values())
        answers_sum_ls = list(answers_sum.values())
        # init 
        non_info_questions =[]
        n_non_informativeness = 0 

        for idx, (ans_src, ans_sum) in enumerate(zip(answers_src_ls, answers_sum_ls)):
            if ans_sum.strip().lower() == 'unknown' and ans_src.strip().lower() in ['yes', 'no']:
                n_non_informativeness += 1
                non_info_question = questions[f"question {idx+1}"]
                non_info_questions.append(f'{non_info_question}')

        return n_non_informativeness, non_info_questions

    @staticmethod
    def compute(answers_src, answers_sum, questions, n_questions):
        """
        Computes the non-informativeness score.

        Args:
            answers_src (dict): Answers derived from the source.
            answers_sum (dict): Answers derived from the summary.
            n_questions (int): Total number of questions.

        Returns:
            float: Non-informativeness score.
        """
        n_non_informativeness, non_info_questions = InformativenessFailure._compute_metric(answers_src, answers_sum, questions)
        is_informativeness_failure = 1 if n_non_informativeness > 0 else 0 
        non_info_score = n_non_informativeness / n_questions
        explanation = non_info_questions
        return is_informativeness_failure, explanation, non_info_score
    



ariadne_ai/metrics/text_summarization/informativeness_failure.py
athina-ai/ariadne
from abc import ABC, abstractmethod

class Publisher(ABC):
   
   @abstractmethod
   def write(self):
        pass

ariadne_ai/publishers/publisher.py
athina-ai/ariadne
from .publisher import Publisher
import json
import os


class PublisherLog(Publisher):
    """
    A class to publish log data in various formats.

    Attributes:
        filename (str): The output filename for the publisher.
        format (str): The format in which the data will be published, e.g., 'json', 'csv'.
    """

    def __init__(self, filename: str, format: str):
        """Initializes the PublisherLog with an output filename and format."""
        self.filename = filename
        self.format = format
        directory_path = os.path.dirname(self.filename)
        if not os.path.exists(directory_path):
            os.makedirs(directory_path)

    def write(self, data: dict):
        """Writes data in the specified format."""
        if self.format == "json":
            self.write_json(data)
        elif self.format == "csv":
            self.write_csv(data)
        else:
            raise NotImplementedError(
                f"The '{self.format}' format has not been implemented yet."
            )

    def write_json(self, data: dict):
        """Writes data to a JSON file."""
        with open(self.filename, "w") as f:
            json.dump(data, f, indent=4)

    def write_csv(self, data: dict):
        """Writes data to a CSV file."""
        pass

ariadne_ai/publishers/publisher_log.py
athina-ai/ariadne
from ariadne_ai.loaders.summarization_loader import SummarizationLoader
from ariadne_ai.evaluators.summarization_hallucination_evaluator import SummarizationHallucinationEvaluator

# Constants and Configurations

# Path to the input file containing raw data.
INPUT_FILEPATH = 'data/text_summarization/xsum_sample.json'

# OpenAI API key (should be kept confidential).
OPEN_AI_KEY = None

# Each configuration specifies the log, performance file paths, model, and number of questions.
# Configurations for different runs
RUN_CONFIGS = [
    {
        'log_filepath': 'data/logs/log_sum_hal_eval_gpt_35_questions_5.json',
        'perf_filepath': 'data/logs/perf_sum_hal_eval_gpt_35_questions_5.txt',
        'llm_model': 'gpt-3.5-turbo-16k',
        'n_questions': 5
    },
    {
        'log_filepath': 'data/logs/log_sum_hal_eval_gpt_35_questions_2.json',
        'perf_filepath': 'data/logs/perf_sum_hal_eval_gpt_35_questions_2.txt',
        'llm_model': 'gpt-3.5-turbo-16k',
        'n_questions': 2
    },
    {
        'log_filepath': 'data/logs/log_sum_hal_eval_gpt_4_questions_5.json',
        'perf_filepath': 'data/logs/perf_sum_hal_eval_gpt_4_questions_5.txt',
        'llm_model': 'gpt-4',
        'n_questions': 5
    }
]

def initialize_loader() -> SummarizationLoader:
    """
    Initialize the summarization loader, load the data, and return the loader object.
    """
    loader = SummarizationLoader(
        col_document='document', 
        col_summary='summary', 
        col_label='hallucination_type',
        col_comment='hallucinated_span'
    )
    loader.load(INPUT_FILEPATH)
    return loader

def run_evaluation(loader: SummarizationLoader, config: dict) -> None:
    """
    Given a loader and configuration, initialize an evaluator and run the evaluation.
    """
    evaluator = SummarizationHallucinationEvaluator(
        loader,
        log_filepath=config['log_filepath'],
        llm_model=config['llm_model'],
        performance_filepath=config['perf_filepath'],
        open_ai_key=OPEN_AI_KEY,
        n_questions=config.get('n_questions')
    )
    evaluator.run()

def main():
    """
    Main execution function. Initializes the loader and runs evaluations for each configuration.
    """
    loader = initialize_loader()
    for config in RUN_CONFIGS:
        run_evaluation(loader, config)

# Ensure that the main execution only occurs if this script is run directly (not imported).
if __name__ == "__main__":
    main()

run_experiment_example.py
athina-ai/ariadne
from ariadne_ai.loaders.rag_loader import RagLoader
from ariadne_ai.evaluators.rag.faithfulness_evaluator import FaithfulnessEvaluator
from ariadne_ai.evaluators.rag.context_relevance_evaluator import (
    ContextRelevanceEvaluator,
)
from ariadne_ai.evaluators.rag.answer_relevance_evaluator import (
    AnswerRelevanceEvaluator,
)

# Constants and Configurations

# Path to the input file containing raw data.
INPUT_FILEPATH = "data/rag/faithfulness_sample.json"
LOG_FILEPATH = "data/logs/log_rag_faith_eval.json"
PERF_FILEPATH = "data/logs/perf_rag_faith_eval.txt"

# OpenAI API key (should be kept confidential).
OPEN_AI_KEY = None

loader = RagLoader(
    col_question="question",
    col_context="context",
    col_answer="answer",
    col_label="label",
)

# Faithfulness
loader.load(INPUT_FILEPATH)
evaluator = FaithfulnessEvaluator(loader)
evaluator.run()

# Context Relevance
INPUT_FILEPATH = "data/rag/context_relevance_sample.json"
loader = RagLoader(
    col_question="question",
    col_context="context",
    col_answer="answer",
    col_label="label",
)
loader.load(INPUT_FILEPATH)
evaluator = ContextRelevanceEvaluator(loader=loader)
evaluator.run()

# Answer Relevance

INPUT_FILEPATH = "data/rag/answer_relevance_sample.json"
loader = RagLoader(
    col_question="question",
    col_context="context",
    col_answer="answer",
    col_label="label",
)
loader.load(INPUT_FILEPATH)
evaluator = AnswerRelevanceEvaluator(loader)
evaluator.run()

run_experiment_rag.py
athina-ai/ariadne
from ariadne_ai.loaders.summarization_loader import SummarizationLoader
from ariadne_ai.evaluators.text_summarization.hallucination_evaluator import HallucinationEvaluator
from ariadne_ai.evaluators.text_summarization.informativeness_evaluator import InformativenessEvaluator
# Constants and Configurations

# Path to the input file containing raw data.
INPUT_FILEPATH = 'data/text_summarization/xsum_sample.json'

# OpenAI API key (should be kept confidential).
OPEN_AI_KEY = None

# Each configuration specifies the log, performance file paths, model, and number of questions.
# Configurations for different runs
RUN_CONFIGS = [
    {
        'log_filepath': 'data/logs/log_sum_hal_eval_gpt_35_questions_5.json',
        'perf_filepath': 'data/logs/perf_sum_hal_eval_gpt_35_questions_5.txt',
        'llm_model': 'gpt-3.5-turbo-16k',
        'n_questions': 5
    },
    {
        'log_filepath': 'data/logs/log_sum_hal_eval_gpt_35_questions_2.json',
        'perf_filepath': 'data/logs/perf_sum_hal_eval_gpt_35_questions_2.txt',
        'llm_model': 'gpt-3.5-turbo-16k',
        'n_questions': 2
    }
]

def initialize_loader() -> SummarizationLoader:
    """
    Initialize the summarization loader, load the data, and return the loader object.
    """
    loader = SummarizationLoader(
        col_document='document', 
        col_summary='summary', 
        col_label='hallucination_type',
        col_comment='hallucinated_span'
    )
    loader.load(INPUT_FILEPATH)
    return loader

def run_evaluation_hallucination(loader: SummarizationLoader, config: dict) -> None:
    """
    Given a loader and configuration, initialize an evaluator and run the hallucination evaluation.
    """
    evaluator = HallucinationEvaluator(
        loader,
        log_filepath=config['log_filepath'],
        llm_model=config['llm_model'],
        performance_filepath=config['perf_filepath'],
        open_ai_key=OPEN_AI_KEY,
        n_questions=config.get('n_questions')
    )
    evaluator.run()

def run_evaluation_informativeness(loader: SummarizationLoader, config: dict) -> None:
    """
    Given a loader and configuration, initialize an evaluator and run the non-informativeness evaluation.
    """
    evaluator = InformativenessEvaluator(
        loader,
        log_filepath=config['log_filepath'],
        llm_model=config['llm_model'],
        performance_filepath=config['perf_filepath'],
        open_ai_key=OPEN_AI_KEY,
        n_questions=config.get('n_questions')
    )
    evaluator.run()

def main():
    """
    Main execution function. Initializes the loader and runs evaluations for each configuration.
    """
    loader = initialize_loader()
    for config in RUN_CONFIGS:
        run_evaluation_informativeness(loader, config)
        run_evaluation_hallucination(loader, config)

# Ensure that the main execution only occurs if this script is run directly (not imported).
if __name__ == "__main__":
    main()


run_experiment_summarization.py
athina-ai/ariadne
import pytest
from ariadne_ai.loaders.summarization_loader import SummarizationLoader

def test_keyerror_on_missing_column():
    """ Missing "summary" key on purpose to trigger KeyError """
    test_data = [
        {"document": "doc1", "summary": "summary1"},
        {"document": "doc2"}  
    ]
    # Check if KeyError is raised 
    with pytest.raises(KeyError):
        SummarizationLoader(format ='dict').load(test_data)


def test_keyerror_on_missing_defining_columns():
    """ Missing defining document and summary columns in initialization on purpose to trigger KeyError """
    test_data = [{"doc": "doc1", "sum": "summary1"}]    
    text_summarization_loader = SummarizationLoader(format ='dict')
    # Check if KeyError is raised 
    with pytest.raises(KeyError):
        text_summarization_loader.load_dict(test_data)
    


def test_loader_renames_documnet_summary_columns():
    """ Checking if the loader renames the document and summary column correctly"""
    test_data = [{"doc": "doc1", "sum": "summary1"}]
    text_summarization_loader = SummarizationLoader(col_document='doc', col_summary='sum', format ='dict')
    text_summarization_loader.load(test_data)
    processed_data = text_summarization_loader.processed_dataset
    assert(processed_data[0]["document"] == test_data[0]["doc"] )
    assert(processed_data[0]["summary"] == test_data[0]["sum"] )



def test_loader_instance():
    text_summarization_loader = SummarizationLoader()
    text_summarization_loader.load_response(document= 'doc', summary='sum')
    processed_data = text_summarization_loader.processed_dataset
    assert(processed_data[0]["document"] == 'doc' )
    assert(processed_data[0]["summary"] == 'sum' )
test/test_loader.py
athina-ai/ariadne