Rag evaluator

RagEvaluatorPack

Bases: BaseLlamaPack

一个用于使用自己的RAG pipeline 进行评估的包。

Parameters:

Name	Type	Description	Default
query_engine	BaseQueryEngine	要评估的RAG pipeline。	required
rag_dataset	BaseLlamaDataset	要进行评估的BaseLlamaDataset。	required
judge_llm	Optional[LLM]	用作评估器的LLM。	None

Source code in llama_index/packs/rag_evaluator/base.py

class RagEvaluatorPack(BaseLlamaPack):
    """一个用于使用自己的RAG pipeline 进行评估的包。

    Args:
        query_engine: 要评估的RAG pipeline。
        rag_dataset: 要进行评估的BaseLlamaDataset。
        judge_llm: 用作评估器的LLM。"""

    def __init__(
        self,
        query_engine: BaseQueryEngine,
        rag_dataset: BaseLlamaDataset,
        judge_llm: Optional[LLM] = None,
        show_progress: bool = True,
    ):
        self.query_engine = query_engine
        self.rag_dataset = rag_dataset
        self._num_examples = len(self.rag_dataset.examples)
        if judge_llm is None:
            self.judge_llm = OpenAI(temperature=0, model="gpt-4-1106-preview")
        else:
            assert isinstance(judge_llm, LLM)
            self.judge_llm = judge_llm
        self.show_progress = show_progress
        self.evals = {
            "correctness": [],
            "relevancy": [],
            "faithfulness": [],
            "context_similarity": [],
        }
        self.eval_queue = deque(range(len(rag_dataset.examples)))
        self.prediction_dataset = None

    async def _amake_predictions(
        self,
        batch_size: int = 20,
        sleep_time_in_seconds: int = 1,
    ):
        """使用查询引擎进行异步预测。"""
        self.prediction_dataset: BaseLlamaPredictionDataset = (
            await self.rag_dataset.amake_predictions_with(
                self.query_engine,
                show_progress=self.show_progress,
                batch_size=batch_size,
                sleep_time_in_seconds=sleep_time_in_seconds,
            )
        )

    def _make_predictions(
        self,
        batch_size: int = 20,
        sleep_time_in_seconds: int = 1,
    ):
        """使用查询引擎同步进行预测。"""
        self.prediction_dataset: BaseLlamaPredictionDataset = (
            self.rag_dataset.make_predictions_with(
                self.query_engine,
                show_progress=self.show_progress,
                batch_size=batch_size,
                sleep_time_in_seconds=sleep_time_in_seconds,
            )
        )

    def _prepare_judges(self):
        """构建评估器。"""
        judges = {}
        judges["correctness"] = CorrectnessEvaluator(
            llm=self.judge_llm,
        )
        judges["relevancy"] = RelevancyEvaluator(
            llm=self.judge_llm,
        )
        judges["faithfulness"] = FaithfulnessEvaluator(
            llm=self.judge_llm,
        )
        judges["semantic_similarity"] = SemanticSimilarityEvaluator(
            embed_model=OpenAIEmbedding()
        )
        return judges

    async def _areturn_null_eval_result(self, query) -> EvaluationResult:
        """一个返回None的虚拟异步方法。

注意：这用于处理创建用于评估预测的异步任务时上下文不存在的情况。
"""
        return EvaluationResult(
            query=query,
        )

    def _return_null_eval_result(self, query) -> EvaluationResult:
        """一个返回None的虚拟异步方法。

注意：这用于处理创建用于评估预测的异步任务时上下文不存在的情况。
"""
        return EvaluationResult(
            query=query,
        )

    def _create_async_evaluate_example_prediction_tasks(
        self, judges, example, prediction, sleep_time_in_seconds
    ):
        """收集协程。"""
        correctness_task = judges["correctness"].aevaluate(
            query=example.query,
            response=prediction.response,
            reference=example.reference_answer,
            sleep_time_in_seconds=sleep_time_in_seconds,
        )

        relevancy_task = judges["relevancy"].aevaluate(
            query=example.query,
            response=prediction.response,
            contexts=prediction.contexts,
            sleep_time_in_seconds=sleep_time_in_seconds,
        )

        faithfulness_task = judges["faithfulness"].aevaluate(
            query=example.query,
            response=prediction.response,
            contexts=prediction.contexts,
            sleep_time_in_seconds=sleep_time_in_seconds,
        )

        if example.reference_contexts and prediction.contexts:
            semantic_similarity_task = judges["semantic_similarity"].aevaluate(
                query=example.query,
                response="\n".join(prediction.contexts),
                reference="\n".join(example.reference_contexts),
            )
        else:
            semantic_similarity_task = self._areturn_null_eval_result(
                query=example.query
            )

        return (
            correctness_task,
            relevancy_task,
            faithfulness_task,
            semantic_similarity_task,
        )

    def _evaluate_example_prediction(self, judges, example, prediction):
        """收集协程。"""
        correctness_result = judges["correctness"].evaluate(
            query=example.query,
            response=prediction.response,
            reference=example.reference_answer,
        )

        relevancy_result = judges["relevancy"].evaluate(
            query=example.query,
            response=prediction.response,
            contexts=prediction.contexts,
        )

        faithfulness_result = judges["faithfulness"].evaluate(
            query=example.query,
            response=prediction.response,
            contexts=prediction.contexts,
        )

        if example.reference_contexts and prediction.contexts:
            semantic_similarity_result = judges["semantic_similarity"].evaluate(
                query=example.query,
                response="\n".join(prediction.contexts),
                reference="\n".join(example.reference_contexts),
            )
        else:
            semantic_similarity_result = self._return_null_eval_result(
                query=example.query
            )

        return (
            correctness_result,
            relevancy_result,
            faithfulness_result,
            semantic_similarity_result,
        )

    def _save_evaluations(self):
        """保存评估的JSON对象。"""
        # saving evaluations
        evaluations_objects = {
            "context_similarity": [e.dict() for e in self.evals["context_similarity"]],
            "correctness": [e.dict() for e in self.evals["correctness"]],
            "faithfulness": [e.dict() for e in self.evals["faithfulness"]],
            "relevancy": [e.dict() for e in self.evals["relevancy"]],
        }

        with open("_evaluations.json", "w") as json_file:
            json.dump(evaluations_objects, json_file)

    def _prepare_and_save_benchmark_results(self):
        """获取所有评估示例预测的平均分。"""
        _, mean_correctness_df = get_eval_results_df(
            ["base_rag"] * len(self.evals["correctness"]),
            self.evals["correctness"],
            metric="correctness",
        )
        _, mean_relevancy_df = get_eval_results_df(
            ["base_rag"] * len(self.evals["relevancy"]),
            self.evals["relevancy"],
            metric="relevancy",
        )
        _, mean_faithfulness_df = get_eval_results_df(
            ["base_rag"] * len(self.evals["faithfulness"]),
            self.evals["faithfulness"],
            metric="faithfulness",
        )
        _, mean_context_similarity_df = get_eval_results_df(
            ["base_rag"] * len(self.evals["context_similarity"]),
            self.evals["context_similarity"],
            metric="context_similarity",
        )

        mean_scores_df = pd.concat(
            [
                mean_correctness_df.reset_index(),
                mean_relevancy_df.reset_index(),
                mean_faithfulness_df.reset_index(),
                mean_context_similarity_df.reset_index(),
            ],
            axis=0,
            ignore_index=True,
        )
        mean_scores_df = mean_scores_df.set_index("index")
        mean_scores_df.index = mean_scores_df.index.set_names(["metrics"])

        # save mean_scores_df
        mean_scores_df.to_csv("benchmark.csv")
        return mean_scores_df

    def _make_evaluations(
        self,
        batch_size,
        sleep_time_in_seconds,
    ):
        """同步进行评估。"""
        judges = self._prepare_judges()

        start_ix = self.eval_queue[0]
        for batch in self._batch_examples_and_preds(
            self.rag_dataset.examples,
            self.prediction_dataset.predictions,
            batch_size=batch_size,
            start_position=start_ix,
        ):
            examples, predictions = batch
            for example, prediction in tqdm.tqdm(zip(examples, predictions)):
                (
                    correctness_result,
                    relevancy_result,
                    faithfulness_result,
                    semantic_similarity_result,
                ) = self._evaluate_example_prediction(
                    judges=judges, example=example, prediction=prediction
                )

                self.evals["correctness"].append(correctness_result)
                self.evals["relevancy"].append(relevancy_result)
                self.evals["faithfulness"].append(faithfulness_result)
                self.evals["context_similarity"].append(semantic_similarity_result)
            time.sleep(sleep_time_in_seconds)

        self._save_evaluations()
        return self._prepare_and_save_benchmark_results()

    def _batch_examples_and_preds(
        self,
        examples: List[Any],
        predictions: List[Any],
        batch_size: int = 10,
        start_position: int = 0,
    ):
        """使用给定的batch_size对示例和预测进行分批处理。"""
        assert self._num_examples == len(predictions)
        for ndx in range(start_position, self._num_examples, batch_size):
            yield examples[
                ndx : min(ndx + batch_size, self._num_examples)
            ], predictions[ndx : min(ndx + batch_size, self._num_examples)]

    async def _amake_evaluations(self, batch_size, sleep_time_in_seconds):
        """异步进行评估。"""
        judges = self._prepare_judges()

        ix = self.eval_queue[0]
        batch_iterator = self._batch_examples_and_preds(
            self.rag_dataset.examples,
            self.prediction_dataset.predictions,
            batch_size=batch_size,
            start_position=ix,
        )
        total_batches = (self._num_examples - ix + 1) / batch_size + (
            (self._num_examples - ix + 1) % batch_size != 0
        )
        if self.show_progress:
            batch_iterator = tqdm_asyncio(
                batch_iterator,
                desc="Batch processing of evaluations",
                total=total_batches,
            )

        for batch in batch_iterator:
            examples, predictions = batch
            tasks = []
            for example, prediction in zip(examples, predictions):
                (
                    correctness_task,
                    relevancy_task,
                    faithfulness_task,
                    semantic_similarity_task,
                ) = self._create_async_evaluate_example_prediction_tasks(
                    judges=judges,
                    example=example,
                    prediction=prediction,
                    sleep_time_in_seconds=sleep_time_in_seconds,
                )

                tasks += [
                    correctness_task,
                    relevancy_task,
                    faithfulness_task,
                    semantic_similarity_task,
                ]

            # do this in batches to avoid RateLimitError
            try:
                eval_results: List[EvaluationResult] = await asyncio.gather(*tasks)
            except RateLimitError as err:
                if self.show_progress:
                    batch_iterator.close()
                raise ValueError(
                    "You've hit rate limits on your OpenAI subscription. This"
                    " `RagEvaluatorPack` maintains state of evaluations. Simply"
                    " re-invoke .arun() in order to continue from where you left"
                    " off."
                ) from err
            # store in memory
            # since final result of eval_results respects order of inputs
            # just take appropriate slices
            self.evals["correctness"] += eval_results[::4]
            self.evals["relevancy"] += eval_results[1::4]
            self.evals["faithfulness"] += eval_results[2::4]
            self.evals["context_similarity"] += eval_results[3::4]
            # update queue
            for _ in range(batch_size):
                if self.eval_queue:
                    self.eval_queue.popleft()
            ix += 1
            if self.show_progress:
                batch_iterator.update()
                batch_iterator.refresh()

        self._save_evaluations()
        return self._prepare_and_save_benchmark_results()

    def run(self, batch_size: int = 10, sleep_time_in_seconds: int = 1):
        if batch_size > 10:
            warnings.warn(
                "You've set a large batch_size (>10). If using OpenAI GPT-4 as "
                " `judge_llm` (which is the default judge_llm),"
                " you may experience a RateLimitError. Previous successful eval "
                " responses are cached per batch. So hitting a RateLimitError"
                " would mean you'd lose all of the current batches successful "
                " GPT-4 calls."
            )
        if self.prediction_dataset is None:
            self._make_predictions(batch_size, sleep_time_in_seconds)

        # evaluate predictions
        eval_sleep_time_in_seconds = (
            sleep_time_in_seconds * 2
        )  # since we make 3 evaluator llm calls
        eval_batch_size = int(max(batch_size / 4, 1))
        return self._make_evaluations(
            batch_size=eval_batch_size, sleep_time_in_seconds=eval_sleep_time_in_seconds
        )

    async def arun(
        self,
        batch_size: int = 10,
        sleep_time_in_seconds: int = 1,
    ):
        if batch_size > 10:
            warnings.warn(
                "You've set a large batch_size (>10). If using OpenAI GPT-4 as "
                " `judge_llm` (which is the default judge_llm),"
                " you may experience a RateLimitError. Previous successful eval "
                " responses are cached per batch. So hitting a RateLimitError"
                " would mean you'd lose all of the current batches successful "
                " GPT-4 calls."
            )

        # make predictions
        if self.prediction_dataset is None:
            await self._amake_predictions(batch_size, sleep_time_in_seconds)

        # evaluate predictions
        eval_sleep_time_in_seconds = (
            sleep_time_in_seconds * 2
        )  # since we make 3 evaluator llm calls and default is gpt-4
        # which is heavily rate-limited
        eval_batch_size = int(max(batch_size / 4, 1))
        return await self._amake_evaluations(
            batch_size=eval_batch_size, sleep_time_in_seconds=eval_sleep_time_in_seconds
        )