/deepeval/simulator/conversation_simulator.py

from typing import Any, Optional, List, Type, Union, Callable
from rich.progress import Progress
from pydantic import BaseModel
import inspect
import asyncio
import uuid
import warnings

from deepeval.utils import (
    get_or_create_event_loop,
    update_pbar,
    add_pbar,
)
from deepeval.metrics.utils import (
    initialize_model,
    trimAndLoadJson,
)
from deepeval.test_case import ConversationalTestCase, Turn
from deepeval.simulator.template import (
    SimulationTemplate,
)
from deepeval.models import DeepEvalBaseLLM
from deepeval.metrics.utils import MULTIMODAL_SUPPORTED_MODELS
from deepeval.simulator.schema import (
    SimulatedInput,
)
from deepeval.simulator.controller.controller import (
    SimulationController,
    expected_outcome_controller,
)
from deepeval.simulator.simulation_graph import (
    SimulationNode,
    default_simulation_node,
)
from deepeval.simulator.simulation_graph.runner import (
    _SimulationGraphRunner,
    _GraphConversationState,
)
from deepeval.progress_context import conversation_simulator_progress_context
from deepeval.dataset import ConversationalGolden

_MISSING = object()


class ConversationSimulator:
    def __init__(
        self,
        model_callback: Callable[[str], str],
        simulation_graph: Optional[SimulationNode] = None,
        stopping_controller: Callable = expected_outcome_controller,
        simulator_model: Optional[Union[str, DeepEvalBaseLLM]] = None,
        max_concurrent: int = 5,
        async_mode: bool = True,
        language: str = "English",
        controller: Any = _MISSING,
    ):
        if controller is not _MISSING:
            if stopping_controller is not expected_outcome_controller:
                raise TypeError(
                    "Pass either `stopping_controller` or the deprecated "
                    "`controller`, not both."
                )
            warnings.warn(
                "`controller` is deprecated; use `stopping_controller` "
                "instead.",
                DeprecationWarning,
                stacklevel=2,
            )
            stopping_controller = controller

        self.model_callback = model_callback
        self.is_callback_async = inspect.iscoroutinefunction(
            self.model_callback
        )
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.async_mode = async_mode
        self.language = language
        self.simulated_conversations: List[ConversationalTestCase] = []
        self.simulator_model, self.using_native_model = initialize_model(
            simulator_model
        )
        # `None` is rewritten to the default node so the runtime path is
        # uniform: `_SimulationGraphRunner` always drives user-turn generation.
        # To customize the prompt template, pass
        # `simulation_graph=default_simulation_node(template=MyTemplate)`.
        self.simulation_graph = (
            simulation_graph
            if simulation_graph is not None
            else default_simulation_node()
        )
        self._graph_runner = _SimulationGraphRunner(root=self.simulation_graph)
        self.stopping_controller = SimulationController(
            controller=stopping_controller,
            generate_schema=self.generate_schema,
            a_generate_schema=self.a_generate_schema,
        )

    def simulate(
        self,
        conversational_goldens: List[ConversationalGolden],
        max_user_simulations: int = 10,
        on_simulation_complete: Optional[
            Callable[[ConversationalTestCase, int], None]
        ] = None,
    ) -> List[ConversationalTestCase]:
        self.simulation_cost = 0 if self.using_native_model else None

        with conversation_simulator_progress_context(
            simulator_model=self.simulator_model.get_model_name(),
            num_conversations=len(conversational_goldens),
            async_mode=self.async_mode,
        ) as (progress, pbar_id), progress:

            if self.async_mode:
                loop = get_or_create_event_loop()
                loop.run_until_complete(
                    self._a_simulate(
                        conversational_goldens=conversational_goldens,
                        max_user_simulations=max_user_simulations,
                        on_simulation_complete=on_simulation_complete,
                        progress=progress,
                        pbar_id=pbar_id,
                    )
                )
            else:
                multimodal = any(
                    [golden.multimodal for golden in conversational_goldens]
                )
                if multimodal:
                    if (
                        not self.simulator_model
                        or not self.simulator_model.supports_multimodal()
                    ):
                        if (
                            self.simulator_model
                            and type(self.simulator_model)
                            in MULTIMODAL_SUPPORTED_MODELS
                        ):
                            raise ValueError(
                                f"The evaluation model {self.simulator_model.name} does not support multimodal evaluations at the moment. Available multi-modal models for the {self.simulator_model.__class__.__name__} provider includes {', '.join(self.simulator_model.__class__.valid_multimodal_models)}."
                            )
                        else:
                            raise ValueError(
                                f"The evaluation model {self.simulator_model.name} does not support multimodal inputs, please use one of the following evaluation models: {', '.join([cls.__name__ for cls in MULTIMODAL_SUPPORTED_MODELS])}"
                            )
                conversational_test_cases: List[ConversationalTestCase] = []
                for conversation_index, golden in enumerate(
                    conversational_goldens
                ):
                    conversational_test_case = (
                        self._simulate_single_conversation(
                            golden=golden,
                            max_user_simulations=max_user_simulations,
                            index=conversation_index,
                            progress=progress,
                            pbar_id=pbar_id,
                            on_simulation_complete=on_simulation_complete,
                        )
                    )
                    conversational_test_cases.append(conversational_test_case)

                self.simulated_conversations = conversational_test_cases

        return self.simulated_conversations

    async def _a_simulate(
        self,
        conversational_goldens: List[ConversationalGolden],
        max_user_simulations: int,
        on_simulation_complete: Optional[
            Callable[[ConversationalTestCase, int], None]
        ] = None,
        progress: Optional[Progress] = None,
        pbar_id: Optional[int] = None,
    ) -> List[ConversationalTestCase]:

        multimodal = any(
            [golden.multimodal for golden in conversational_goldens]
        )
        if multimodal:
            if (
                not self.simulator_model
                or not self.simulator_model.supports_multimodal()
            ):
                if (
                    self.simulator_model
                    and type(self.simulator_model)
                    in MULTIMODAL_SUPPORTED_MODELS
                ):
                    raise ValueError(
                        f"The evaluation model {self.simulator_model.name} does not support multimodal evaluations at the moment. Available multi-modal models for the {self.simulator_model.__class__.__name__} provider includes {', '.join(self.simulator_model.__class__.valid_multimodal_models)}."
                    )
                else:
                    raise ValueError(
                        f"The evaluation model {self.simulator_model.name} does not support multimodal inputs, please use one of the following evaluation models: {', '.join([cls.__name__ for cls in MULTIMODAL_SUPPORTED_MODELS])}"
                    )

        self.simulation_cost = 0 if self.using_native_model else None

        async def simulate_conversations(
            golden: ConversationalGolden,
            conversation_index: int,
        ):
            async with self.semaphore:
                return await self._a_simulate_single_conversation(
                    golden=golden,
                    max_user_simulations=max_user_simulations,
                    index=conversation_index,
                    progress=progress,
                    pbar_id=pbar_id,
                    on_simulation_complete=on_simulation_complete,
                )

        tasks = [
            simulate_conversations(golden, i)
            for i, golden in enumerate(conversational_goldens)
        ]
        self.simulated_conversations = await asyncio.gather(*tasks)

    ############################################
    ### Simulate Single Conversation ###########
    ############################################

    def _simulate_single_conversation(
        self,
        golden: ConversationalGolden,
        max_user_simulations: int,
        index: int,
        progress: Optional[Progress] = None,
        pbar_id: Optional[int] = None,
        on_simulation_complete: Optional[
            Callable[[ConversationalTestCase, int], None]
        ] = None,
    ) -> ConversationalTestCase:
        simulation_counter = 0
        if max_user_simulations <= 0:
            raise ValueError("max_user_simulations must be greater than 0")

        # Define pbar
        pbar_max_user_simluations_id = add_pbar(
            progress,
            f"\t⚡ Test case #{index}",
            total=max_user_simulations + 1,
        )

        additional_metadata = {"User Description": golden.user_description}
        user_input = None
        thread_id = str(uuid.uuid4())
        turns: List[Turn] = []
        graph_state: _GraphConversationState = (
            self._graph_runner.new_conversation_state()
        )

        if golden.turns is not None:
            turns.extend(golden.turns)

        while True:
            if simulation_counter >= max_user_simulations:
                update_pbar(progress, pbar_max_user_simluations_id)
                break

            # Stop conversation if needed
            should_stop_simulation = self.stopping_controller.run(
                turns=turns,
                golden=golden,
                index=index,
                thread_id=thread_id,
                simulation_counter=simulation_counter,
                max_user_simulations=max_user_simulations,
                progress=progress,
                pbar_turns_id=pbar_max_user_simluations_id,
            )
            if should_stop_simulation:
                break

            # Generate turn from user (via simulation graph)
            emission_end = False
            if len(turns) > 0 and turns[-1].role == "user":
                user_input = turns[-1].content
            else:
                emission = self._graph_runner.run(
                    self,
                    graph_state,
                    turns,
                    golden,
                    thread_id,
                    self.language,
                )
                emission_end = emission.end
                if emission.turn is None:
                    # max_visits exhausted on entry; end without another turn.
                    update_pbar(progress, pbar_max_user_simluations_id)
                    break
                turns.append(emission.turn)
                user_input = emission.turn.content
                update_pbar(progress, pbar_max_user_simluations_id)
                simulation_counter += 1

            # Generate turn from assistant
            if self.is_callback_async:
                assistant_turn = asyncio.run(
                    self.a_generate_turn_from_callback(
                        user_input,
                        model_callback=self.model_callback,
                        turns=turns,
                        thread_id=thread_id,
                    )
                )
            else:
                assistant_turn = self.generate_turn_from_callback(
                    user_input,
                    model_callback=self.model_callback,
                    turns=turns,
                    thread_id=thread_id,
                )
            turns.append(assistant_turn)

            # Route to the next graph node based on the assistant reply.
            self._graph_runner.advance(
                self, graph_state, assistant_turn.content
            )

            if emission_end:
                break

        update_pbar(progress, pbar_id)
        conversational_test_case = ConversationalTestCase(
            turns=turns,
            scenario=golden.scenario,
            expected_outcome=golden.expected_outcome,
            user_description=golden.user_description,
            context=golden.context,
            name=golden.name,
            additional_metadata={
                **(golden.additional_metadata or {}),
                **additional_metadata,
            },
            comments=golden.comments,
            _dataset_rank=golden._dataset_rank,
            _dataset_alias=golden._dataset_alias,
            _dataset_id=golden._dataset_id,
        )
        if on_simulation_complete:
            on_simulation_complete(conversational_test_case, index)
        return conversational_test_case

    async def _a_simulate_single_conversation(
        self,
        golden: ConversationalGolden,
        max_user_simulations: int,
        index: Optional[int] = None,
        progress: Optional[Progress] = None,
        pbar_id: Optional[int] = None,
        on_simulation_complete: Optional[
            Callable[[ConversationalTestCase, int], None]
        ] = None,
    ) -> ConversationalTestCase:
        simulation_counter = 0
        if max_user_simulations <= 0:
            raise ValueError("max_user_simulations must be greater than 0")

        # Define pbar
        pbar_max_user_simluations_id = add_pbar(
            progress,
            f"\t⚡ Test case #{index}",
            total=max_user_simulations + 1,
        )

        additional_metadata = {"User Description": golden.user_description}
        user_input = None
        thread_id = str(uuid.uuid4())
        turns: List[Turn] = []
        graph_state: _GraphConversationState = (
            self._graph_runner.new_conversation_state()
        )

        if golden.turns is not None:
            turns.extend(golden.turns)

        while True:
            if simulation_counter >= max_user_simulations:
                update_pbar(progress, pbar_max_user_simluations_id)
                break

            # Stop conversation if needed
            should_stop_simulation = await self.stopping_controller.a_run(
                turns=turns,
                golden=golden,
                index=index if index is not None else 0,
                thread_id=thread_id,
                simulation_counter=simulation_counter,
                max_user_simulations=max_user_simulations,
                progress=progress,
                pbar_turns_id=pbar_max_user_simluations_id,
            )
            if should_stop_simulation:
                break

            # Generate turn from user (via simulation graph)
            emission_end = False
            if len(turns) > 0 and turns[-1].role == "user":
                user_input = turns[-1].content
            else:
                emission = await self._graph_runner.a_run(
                    self,
                    graph_state,
                    turns,
                    golden,
                    thread_id,
                    self.language,
                )
                emission_end = emission.end
                if emission.turn is None:
                    update_pbar(progress, pbar_max_user_simluations_id)
                    break
                turns.append(emission.turn)
                user_input = emission.turn.content
                update_pbar(progress, pbar_max_user_simluations_id)
                simulation_counter += 1

            # Generate turn from assistant
            if self.is_callback_async:
                assistant_turn = await self.a_generate_turn_from_callback(
                    user_input,
                    model_callback=self.model_callback,
                    turns=turns,
                    thread_id=thread_id,
                )
            else:
                assistant_turn = self.generate_turn_from_callback(
                    user_input,
                    model_callback=self.model_callback,
                    turns=turns,
                    thread_id=thread_id,
                )
            turns.append(assistant_turn)

            # Route to the next graph node based on the assistant reply.
            await self._graph_runner.a_advance(
                self, graph_state, assistant_turn.content
            )

            if emission_end:
                break

        update_pbar(progress, pbar_id)
        conversational_test_case = ConversationalTestCase(
            turns=turns,
            scenario=golden.scenario,
            expected_outcome=golden.expected_outcome,
            user_description=golden.user_description,
            context=golden.context,
            name=golden.name,
            additional_metadata={
                **(golden.additional_metadata or {}),
                **additional_metadata,
            },
            comments=golden.comments,
            _dataset_rank=golden._dataset_rank,
            _dataset_alias=golden._dataset_alias,
            _dataset_id=golden._dataset_id,
        )
        if on_simulation_complete:
            on_simulation_complete(conversational_test_case, index)
        return conversational_test_case

    ############################################
    ### Generate User Inputs ###################
    ############################################

    def generate_first_user_input(
        self,
        golden: ConversationalGolden,
        template: Optional[Type[SimulationTemplate]] = None,
    ):
        tmpl = template or SimulationTemplate
        prompt = tmpl.simulate_first_user_turn(golden, self.language)
        simulated_input: SimulatedInput = self.generate_schema(
            prompt, SimulatedInput
        )
        return simulated_input.simulated_input

    async def a_generate_first_user_input(
        self,
        golden: ConversationalGolden,
        template: Optional[Type[SimulationTemplate]] = None,
    ):
        tmpl = template or SimulationTemplate
        prompt = tmpl.simulate_first_user_turn(golden, self.language)
        simulated_input: SimulatedInput = await self.a_generate_schema(
            prompt, SimulatedInput
        )
        return simulated_input.simulated_input

    def generate_next_user_input(
        self,
        golden: ConversationalGolden,
        turns: List[Turn],
        template: Optional[Type[SimulationTemplate]] = None,
    ):
        tmpl = template or SimulationTemplate
        prompt = tmpl.simulate_user_turn(golden, turns, self.language)
        simulated_input: SimulatedInput = self.generate_schema(
            prompt, SimulatedInput
        )
        return simulated_input.simulated_input

    async def a_generate_next_user_input(
        self,
        golden: ConversationalGolden,
        turns: List[Turn],
        template: Optional[Type[SimulationTemplate]] = None,
    ):
        tmpl = template or SimulationTemplate
        prompt = tmpl.simulate_user_turn(golden, turns, self.language)
        simulated_input: SimulatedInput = await self.a_generate_schema(
            prompt, SimulatedInput
        )
        return simulated_input.simulated_input

    ############################################
    ### Generate Structured Response ###########
    ############################################

    def generate_schema(
        self,
        prompt: str,
        schema: BaseModel,
    ) -> BaseModel:
        if self.using_native_model:
            res, cost = self.simulator_model.generate(prompt, schema=schema)
            if cost is not None:
                self.simulation_cost += cost
            return res
        else:
            try:
                res = self.simulator_model.generate(prompt, schema=schema)
                return res
            except TypeError:
                res = self.simulator_model.generate(prompt)
                data = trimAndLoadJson(res)
                return schema(**data)

    async def a_generate_schema(
        self,
        prompt: str,
        schema: BaseModel,
    ) -> BaseModel:
        if self.using_native_model:
            res, cost = await self.simulator_model.a_generate(
                prompt, schema=schema
            )
            if cost is not None:
                self.simulation_cost += cost
            return res
        else:
            try:
                res = await self.simulator_model.a_generate(
                    prompt, schema=schema
                )
                return res
            except TypeError:
                res = await self.simulator_model.a_generate(prompt)
            data = trimAndLoadJson(res)
            return schema(**data)

    ############################################
    ### Invoke Model Callback ##################
    ############################################

    def generate_turn_from_callback(
        self,
        input: str,
        turns: List[Turn],
        thread_id: str,
        model_callback: Callable,
    ) -> Turn:
        callback_kwargs = {
            "input": input,
            "turns": turns,
            "thread_id": thread_id,
        }
        supported_args = set(
            inspect.signature(model_callback).parameters.keys()
        )
        return model_callback(
            **{k: v for k, v in callback_kwargs.items() if k in supported_args}
        )

    async def a_generate_turn_from_callback(
        self,
        input: str,
        model_callback: Callable,
        turns: List[Turn],
        thread_id: str,
    ) -> Turn:
        candidate_kwargs = {
            "input": input,
            "turns": turns,
            "thread_id": thread_id,
        }
        supported_args = set(
            inspect.signature(model_callback).parameters.keys()
        )
        return await model_callback(
            **{k: v for k, v in candidate_kwargs.items() if k in supported_args}
        )

    ############################################
    ### Invoke Model Callback ##################
    ############################################