deepset-ai--haystack/test/components/evaluators/test_llm_evaluator.py

# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0

from unittest.mock import AsyncMock, Mock

import pytest

from haystack import Pipeline
from haystack.components.evaluators import LLMEvaluator
from haystack.components.generators.chat.openai import OpenAIChatGenerator
from haystack.dataclasses.chat_message import ChatMessage


class TestLLMEvaluator:
    def test_init_default(self, monkeypatch):
        monkeypatch.setenv("OPENAI_API_KEY", "test-api-key")
        component = LLMEvaluator(
            instructions="test-instruction",
            inputs=[("predicted_answers", list[str])],
            outputs=["score"],
            examples=[
                {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}}
            ],
        )
        assert component.instructions == "test-instruction"
        assert component.inputs == [("predicted_answers", list[str])]
        assert component.outputs == ["score"]
        assert component.examples == [
            {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}}
        ]

        assert isinstance(component._chat_generator, OpenAIChatGenerator)
        assert component._chat_generator.generation_kwargs == {"response_format": {"type": "json_object"}, "seed": 42}

    def test_key_resolved_at_warm_up_not_init(self, monkeypatch):
        monkeypatch.delenv("OPENAI_API_KEY", raising=False)
        component = LLMEvaluator(
            instructions="test-instruction",
            inputs=[("predicted_answers", list[str])],
            outputs=["score"],
            examples=[
                {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}}
            ],
        )
        with pytest.raises(ValueError, match="None of the .* environment variables are set"):
            component.warm_up()

    def test_init_with_chat_generator(self, monkeypatch):
        monkeypatch.setenv("OPENAI_API_KEY", "test-api-key")
        chat_generator = OpenAIChatGenerator(generation_kwargs={"custom_key": "custom_value"})
        component = LLMEvaluator(
            instructions="test-instruction",
            chat_generator=chat_generator,
            inputs=[("predicted_answers", list[str])],
            outputs=["custom_score"],
            examples=[
                {"inputs": {"predicted_answers": "answer 1"}, "outputs": {"custom_score": 1}},
                {"inputs": {"predicted_answers": "answer 2"}, "outputs": {"custom_score": 0}},
            ],
        )

        assert component._chat_generator is chat_generator

    def test_init_with_invalid_parameters(self, monkeypatch):
        monkeypatch.setenv("OPENAI_API_KEY", "test-api-key")
        # Invalid inputs
        with pytest.raises(ValueError):
            LLMEvaluator(
                instructions="test-instruction",
                inputs={("predicted_answers", list[str])},
                outputs=["score"],
                examples=[
                    {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}}
                ],
            )
        with pytest.raises(ValueError):
            LLMEvaluator(
                instructions="test-instruction",
                inputs=[(list[str], "predicted_answers")],
                outputs=["score"],
                examples=[
                    {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}}
                ],
            )
        with pytest.raises(ValueError):
            LLMEvaluator(
                instructions="test-instruction",
                inputs=[list[str]],
                outputs=["score"],
                examples=[
                    {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}}
                ],
            )
        with pytest.raises(ValueError):
            LLMEvaluator(
                instructions="test-instruction",
                inputs={("predicted_answers", str)},
                outputs=["score"],
                examples=[
                    {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}}
                ],
            )

        # Invalid outputs
        with pytest.raises(ValueError):
            LLMEvaluator(
                instructions="test-instruction",
                inputs=[("predicted_answers", list[str])],
                outputs="score",
                examples=[
                    {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}}
                ],
            )
        with pytest.raises(ValueError):
            LLMEvaluator(
                instructions="test-instruction",
                inputs=[("predicted_answers", list[str])],
                outputs=[["score"]],
                examples=[
                    {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}}
                ],
            )

        # Invalid examples
        with pytest.raises(ValueError):
            LLMEvaluator(
                instructions="test-instruction",
                inputs=[("predicted_answers", list[str])],
                outputs=["score"],
                examples={
                    "inputs": {"predicted_answers": "Damn, this is straight outta hell!!!"},
                    "outputs": {"custom_score": 1},
                },
            )
        with pytest.raises(ValueError):
            LLMEvaluator(
                instructions="test-instruction",
                inputs=[("predicted_answers", list[str])],
                outputs=["score"],
                examples=[
                    [
                        {
                            "inputs": {"predicted_answers": "Damn, this is straight outta hell!!!"},
                            "outputs": {"custom_score": 1},
                        }
                    ]
                ],
            )
        with pytest.raises(ValueError):
            LLMEvaluator(
                instructions="test-instruction",
                inputs=[("predicted_answers", list[str])],
                outputs=["score"],
                examples=[
                    {
                        "wrong_key": {"predicted_answers": "Damn, this is straight outta hell!!!"},
                        "outputs": {"custom_score": 1},
                    }
                ],
            )
        with pytest.raises(ValueError):
            LLMEvaluator(
                instructions="test-instruction",
                inputs=[("predicted_answers", list[str])],
                outputs=["score"],
                examples=[
                    {
                        "inputs": [{"predicted_answers": "Damn, this is straight outta hell!!!"}],
                        "outputs": [{"custom_score": 1}],
                    }
                ],
            )
        with pytest.raises(ValueError):
            LLMEvaluator(
                instructions="test-instruction",
                inputs=[("predicted_answers", list[str])],
                outputs=["score"],
                examples=[{"inputs": {1: "Damn, this is straight outta hell!!!"}, "outputs": {2: 1}}],
            )

    def test_to_dict_default(self, monkeypatch):
        monkeypatch.setenv("OPENAI_API_KEY", "test-api-key")
        chat_generator = OpenAIChatGenerator(generation_kwargs={"response_format": {"type": "json_object"}, "seed": 42})

        component = LLMEvaluator(
            instructions="test-instruction",
            inputs=[("predicted_answers", list[str])],
            outputs=["score"],
            examples=[
                {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}}
            ],
        )
        data = component.to_dict()
        assert data == {
            "type": "haystack.components.evaluators.llm_evaluator.LLMEvaluator",
            "init_parameters": {
                "chat_generator": chat_generator.to_dict(),
                "instructions": "test-instruction",
                "inputs": [["predicted_answers", "list[str]"]],
                "outputs": ["score"],
                "raise_on_failure": True,
                "progress_bar": True,
                "examples": [
                    {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}}
                ],
            },
        }

    def test_to_dict_with_parameters(self, monkeypatch):
        monkeypatch.setenv("OPENAI_API_KEY", "test-api-key")
        chat_generator = OpenAIChatGenerator(generation_kwargs={"response_format": {"type": "json_object"}, "seed": 42})

        component = LLMEvaluator(
            instructions="test-instruction",
            inputs=[("predicted_answers", list[str])],
            outputs=["custom_score"],
            raise_on_failure=False,
            examples=[
                {
                    "inputs": {"predicted_answers": "Damn, this is straight outta hell!!!"},
                    "outputs": {"custom_score": 1},
                },
                {
                    "inputs": {"predicted_answers": "Football is the most popular sport."},
                    "outputs": {"custom_score": 0},
                },
            ],
        )
        data = component.to_dict()
        assert data == {
            "type": "haystack.components.evaluators.llm_evaluator.LLMEvaluator",
            "init_parameters": {
                "chat_generator": chat_generator.to_dict(),
                "instructions": "test-instruction",
                "inputs": [["predicted_answers", "list[str]"]],
                "outputs": ["custom_score"],
                "raise_on_failure": False,
                "progress_bar": True,
                "examples": [
                    {
                        "inputs": {"predicted_answers": "Damn, this is straight outta hell!!!"},
                        "outputs": {"custom_score": 1},
                    },
                    {
                        "inputs": {"predicted_answers": "Football is the most popular sport."},
                        "outputs": {"custom_score": 0},
                    },
                ],
            },
        }

    def test_from_dict(self, monkeypatch):
        monkeypatch.setenv("OPENAI_API_KEY", "test-api-key")
        chat_generator = OpenAIChatGenerator(generation_kwargs={"response_format": {"type": "json_object"}, "seed": 42})

        data = {
            "type": "haystack.components.evaluators.llm_evaluator.LLMEvaluator",
            "init_parameters": {
                "chat_generator": chat_generator.to_dict(),
                "instructions": "test-instruction",
                "inputs": [["predicted_answers", "list[str]"]],
                "outputs": ["score"],
                "examples": [
                    {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}}
                ],
            },
        }

        component = LLMEvaluator.from_dict(data)
        assert isinstance(component._chat_generator, OpenAIChatGenerator)
        assert component._chat_generator.generation_kwargs == {"response_format": {"type": "json_object"}, "seed": 42}
        assert component.instructions == "test-instruction"
        assert component.inputs == [("predicted_answers", list[str])]
        assert component.outputs == ["score"]
        assert component.examples == [
            {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}}
        ]

    def test_pipeline_serde(self, monkeypatch):
        monkeypatch.setenv("OPENAI_API_KEY", "test-api-key")
        pipeline = Pipeline()
        component = LLMEvaluator(
            instructions="test-instruction",
            inputs=[("questions", list[str]), ("predicted_answers", list[list[str]])],
            outputs=["score"],
            examples=[
                {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}}
            ],
        )
        pipeline.add_component("evaluator", component)
        serialized_pipeline = pipeline.dumps()
        deserialized_pipeline = Pipeline.loads(serialized_pipeline)
        assert deserialized_pipeline == pipeline

    def test_run_with_different_lengths(self, monkeypatch):
        monkeypatch.setenv("OPENAI_API_KEY", "test-api-key")
        component = LLMEvaluator(
            instructions="test-instruction",
            inputs=[("questions", list[str]), ("predicted_answers", list[list[str]])],
            outputs=["score"],
            examples=[
                {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}}
            ],
        )

        def chat_generator_run(self, *args, **kwargs):
            return {"replies": [ChatMessage.from_assistant('{"score": 0.5}')]}

        monkeypatch.setattr("haystack.components.evaluators.llm_evaluator.OpenAIChatGenerator.run", chat_generator_run)

        with pytest.raises(ValueError):
            component.run(questions=["What is the capital of Germany?"], predicted_answers=[["Berlin"], ["Paris"]])

        with pytest.raises(ValueError):
            component.run(
                questions=["What is the capital of Germany?", "What is the capital of France?"],
                predicted_answers=[["Berlin"]],
            )

    def test_run_returns_parsed_result(self, monkeypatch):
        monkeypatch.setenv("OPENAI_API_KEY", "test-api-key")
        component = LLMEvaluator(
            instructions="test-instruction",
            inputs=[("questions", list[str]), ("predicted_answers", list[list[str]])],
            outputs=["score"],
            examples=[
                {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}}
            ],
        )

        def chat_generator_run(self, *args, **kwargs):
            return {"replies": [ChatMessage.from_assistant('{"score": 0.5}')]}

        monkeypatch.setattr("haystack.components.evaluators.llm_evaluator.OpenAIChatGenerator.run", chat_generator_run)

        results = component.run(questions=["What is the capital of Germany?"], predicted_answers=["Berlin"])
        assert results == {"results": [{"score": 0.5}], "meta": None}

    def test_prepare_template(self, monkeypatch):
        monkeypatch.setenv("OPENAI_API_KEY", "test-api-key")
        component = LLMEvaluator(
            instructions="test-instruction",
            inputs=[("predicted_answers", list[str])],
            outputs=["score"],
            examples=[
                {"inputs": {"predicted_answers": "Damn, this is straight outta hell!!!"}, "outputs": {"score": 1}},
                {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}},
            ],
        )
        template = component.prepare_template()
        assert (
            template
            == "Instructions:\ntest-instruction\n\nGenerate the response in JSON format with the following keys:"
            '\n["score"]\nConsider the instructions and the examples below to determine those values.\n\n'
            'Examples:\nInputs:\n{"predicted_answers": "Damn, this is straight outta hell!!!"}\nOutputs:'
            '\n{"score": 1}\nInputs:\n{"predicted_answers": "Football is the most popular sport."}\nOutputs:'
            '\n{"score": 0}\n\nInputs:\n{"predicted_answers": {{ predicted_answers }}}\nOutputs:\n'
        )

    def test_invalid_input_parameters(self, monkeypatch):
        monkeypatch.setenv("OPENAI_API_KEY", "test-api-key")
        component = LLMEvaluator(
            instructions="test-instruction",
            inputs=[("predicted_answers", list[str])],
            outputs=["score"],
            examples=[
                {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}}
            ],
        )
        # None of the expected parameters are received
        with pytest.raises(ValueError):
            component.validate_input_parameters(
                expected={"predicted_answers": list[str]}, received={"questions": list[str]}
            )

        # Only one but not all the expected parameters are received
        with pytest.raises(ValueError):
            component.validate_input_parameters(
                expected={"predicted_answers": list[str], "questions": list[str]}, received={"questions": list[str]}
            )

        # Received inputs are not lists
        with pytest.raises(ValueError):
            component.validate_input_parameters(expected={"questions": list[str]}, received={"questions": str})

    def test_invalid_outputs(self, monkeypatch):
        monkeypatch.setenv("OPENAI_API_KEY", "test-api-key")
        component = LLMEvaluator(
            instructions="test-instruction",
            inputs=[("predicted_answers", list[str])],
            outputs=["score"],
            examples=[
                {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}}
            ],
        )

        def chat_generator_run(self, *args, **kwargs):
            return {"replies": [ChatMessage.from_assistant('{"score": 1.0}')]}

        monkeypatch.setattr("haystack.components.evaluators.llm_evaluator.OpenAIChatGenerator.run", chat_generator_run)

        # Test missing key "another_expected_output"
        component.outputs = ["score", "another_expected_output"]
        with pytest.raises(ValueError, match="Missing expected keys"):
            component.run(predicted_answers=["answer"])

        # Test wrong key
        def chat_generator_run_wrong_key(self, *args, **kwargs):
            return {"replies": [ChatMessage.from_assistant('{"wrong_name": 1.0}')]}

        monkeypatch.setattr(
            "haystack.components.evaluators.llm_evaluator.OpenAIChatGenerator.run", chat_generator_run_wrong_key
        )

        component.outputs = ["score"]
        with pytest.raises(ValueError, match="Missing expected keys"):
            component.run(predicted_answers=["answer"])

    def test_output_invalid_json_raise_on_failure_false(self, monkeypatch):
        monkeypatch.setenv("OPENAI_API_KEY", "test-api-key")
        component = LLMEvaluator(
            instructions="test-instruction",
            inputs=[("predicted_answers", list[str])],
            outputs=["score"],
            examples=[
                {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}}
            ],
            raise_on_failure=False,
        )

        def chat_generator_run(self, *args, **kwargs):
            return {"replies": [ChatMessage.from_assistant("some_invalid_json_output")]}

        monkeypatch.setattr("haystack.components.evaluators.llm_evaluator.OpenAIChatGenerator.run", chat_generator_run)

        result = component.run(predicted_answers=["answer"])
        assert result["results"] == [None]

    def test_output_invalid_json_raise_on_failure_true(self, monkeypatch):
        monkeypatch.setenv("OPENAI_API_KEY", "test-api-key")
        component = LLMEvaluator(
            instructions="test-instruction",
            inputs=[("predicted_answers", list[str])],
            outputs=["score"],
            examples=[
                {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}}
            ],
            raise_on_failure=True,
        )

        def chat_generator_run(self, *args, **kwargs):
            return {"replies": [ChatMessage.from_assistant("some_invalid_json_output")]}

        monkeypatch.setattr("haystack.components.evaluators.llm_evaluator.OpenAIChatGenerator.run", chat_generator_run)

        with pytest.raises(
            ValueError
        ):  # json_utils/LLMEvaluator might raise JSONDecodeError which inherits from ValueError or wrapped
            component.run(predicted_answers=["answer"])


class TestLLMEvaluatorAsync:
    @pytest.mark.asyncio
    async def test_run_async_returns_parsed_result(self, monkeypatch):
        monkeypatch.setenv("OPENAI_API_KEY", "test-api-key")
        component = LLMEvaluator(
            instructions="test-instruction",
            inputs=[("questions", list[str]), ("predicted_answers", list[str])],
            outputs=["score"],
            examples=[
                {
                    "inputs": {
                        "questions": "What is the value of any non-zero number raised to the power of zero?",
                        "predicted_answers": "Zero",
                    },
                    "outputs": {"score": 0},
                }
            ],
        )

        async def chat_generator_run_async(self, *args, **kwargs):
            return {"replies": [ChatMessage.from_assistant('{"score": 1}')]}

        monkeypatch.setattr(
            "haystack.components.evaluators.llm_evaluator.OpenAIChatGenerator.run_async", chat_generator_run_async
        )

        results = await component.run_async(
            questions=["What is the perimeter of a circle called?"], predicted_answers=["Circumference"]
        )
        assert results == {"results": [{"score": 1}], "meta": None}

    @pytest.mark.asyncio
    async def test_run_async_fallback_to_thread_with_sync_generator(self, monkeypatch):
        monkeypatch.setenv("OPENAI_API_KEY", "test-api-key")

        class SyncOnlyGenerator:
            def run(self, messages):
                return {"replies": [ChatMessage.from_assistant('{"score": 0}')]}

        component = LLMEvaluator(
            instructions="test-instruction",
            inputs=[("questions", list[str]), ("predicted_answers", list[str])],
            outputs=["score"],
            examples=[
                {
                    "inputs": {
                        "questions": "What is the top sport?",
                        "predicted_answers": "Football is the most popular sport.",
                    },
                    "outputs": {"score": 1},
                }
            ],
            chat_generator=SyncOnlyGenerator(),
        )

        results = await component.run_async(questions=["question"], predicted_answers=["answer"])
        assert results == {"results": [{"score": 0}], "meta": None}

    @pytest.mark.asyncio
    async def test_run_async_raise_on_failure_false(self, monkeypatch):
        monkeypatch.setenv("OPENAI_API_KEY", "test-api-key")
        component = LLMEvaluator(
            instructions="test-instruction",
            inputs=[("questions", list[str]), ("predicted_answers", list[str])],
            outputs=["score"],
            examples=[
                {
                    "inputs": {
                        "questions": "What is the value of any non-zero number raised to the power of zero?",
                        "predicted_answers": "One",
                    },
                    "outputs": {"score": 1},
                }
            ],
            raise_on_failure=False,
        )

        async def chat_generator_run_async(self, *args, **kwargs):
            raise Exception("API error")

        monkeypatch.setattr(
            "haystack.components.evaluators.llm_evaluator.OpenAIChatGenerator.run_async", chat_generator_run_async
        )

        result = await component.run_async(questions=["question"], predicted_answers=["answer"])
        assert result["results"] == [None]

    @pytest.mark.asyncio
    async def test_run_async_raise_on_failure_true(self, monkeypatch):
        monkeypatch.setenv("OPENAI_API_KEY", "test-api-key")
        component = LLMEvaluator(
            instructions="test-instruction",
            inputs=[("questions", list[str]), ("predicted_answers", list[str])],
            outputs=["score"],
            examples=[
                {
                    "inputs": {
                        "questions": "What is the smallest unit of data in a computer?",
                        "predicted_answers": "Bit",
                    },
                    "outputs": {"score": 1},
                }
            ],
        )

        async def chat_generator_run_async(self, *args, **kwargs):
            raise Exception("API error")

        monkeypatch.setattr(
            "haystack.components.evaluators.llm_evaluator.OpenAIChatGenerator.run_async", chat_generator_run_async
        )

        with pytest.raises(ValueError):
            await component.run_async(questions=["question"], predicted_answers=["answer"])


class TestComponentLifecycle:
    @staticmethod
    def _make_evaluator(chat_generator):
        return LLMEvaluator(
            instructions="test-instruction",
            inputs=[("predicted_answers", list[str])],
            outputs=["score"],
            examples=[
                {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}}
            ],
            chat_generator=chat_generator,
        )

    def test_warm_up_delegates_to_chat_generator(self):
        chat_generator = Mock(spec=["run", "warm_up"])
        evaluator = self._make_evaluator(chat_generator)
        evaluator.warm_up()
        chat_generator.warm_up.assert_called_once()

    async def test_warm_up_async_delegates_to_chat_generator(self):
        chat_generator = Mock(spec=["run", "warm_up_async"])
        chat_generator.warm_up_async = AsyncMock()
        evaluator = self._make_evaluator(chat_generator)
        await evaluator.warm_up_async()
        chat_generator.warm_up_async.assert_awaited_once()

    async def test_warm_up_async_falls_back_to_sync_warm_up(self):
        chat_generator = Mock(spec=["run", "warm_up"])
        evaluator = self._make_evaluator(chat_generator)
        await evaluator.warm_up_async()
        chat_generator.warm_up.assert_called_once()

    def test_close_delegates_to_chat_generator(self):
        chat_generator = Mock(spec=["run", "close"])
        evaluator = self._make_evaluator(chat_generator)
        evaluator.close()
        chat_generator.close.assert_called_once()

    async def test_close_async_delegates_to_chat_generator(self):
        chat_generator = Mock(spec=["run", "close_async"])
        chat_generator.close_async = AsyncMock()
        evaluator = self._make_evaluator(chat_generator)
        await evaluator.close_async()
        chat_generator.close_async.assert_awaited_once()

    async def test_close_async_falls_back_to_sync_close(self):
        chat_generator = Mock(spec=["run", "close"])
        evaluator = self._make_evaluator(chat_generator)
        await evaluator.close_async()
        chat_generator.close.assert_called_once()

    async def test_lifecycle_is_safe_when_chat_generator_lacks_methods(self):
        chat_generator = Mock(spec=["run"])
        evaluator = self._make_evaluator(chat_generator)
        evaluator.warm_up()
        await evaluator.warm_up_async()
        evaluator.close()
        await evaluator.close_async()