Underground-Digital/Workflow-Engine

1from abc import ABC, abstractmethod2from enum import Enum3from typing import Optional4 5from pydantic import BaseModel6 7from core.extension.extensible import Extensible, ExtensionModule8 9 10class ModerationAction(Enum):11    DIRECT_OUTPUT = "direct_output"12    OVERRIDDEN = "overridden"13 14 15class ModerationInputsResult(BaseModel):16    flagged: bool = False17    action: ModerationAction18    preset_response: str = ""19    inputs: dict = {}20    query: str = ""21 22 23class ModerationOutputsResult(BaseModel):24    flagged: bool = False25    action: ModerationAction26    preset_response: str = ""27    text: str = ""28 29 30class Moderation(Extensible, ABC):31    """32    The base class of moderation.33    """34 35    module: ExtensionModule = ExtensionModule.MODERATION36 37    def __init__(self, app_id: str, tenant_id: str, config: Optional[dict] = None) -> None:38        super().__init__(tenant_id, config)39        self.app_id = app_id40 41    @classmethod42    @abstractmethod43    def validate_config(cls, tenant_id: str, config: dict) -> None:44        """45        Validate the incoming form config data.46 47        :param tenant_id: the id of workspace48        :param config: the form config data49        :return:50        """51        raise NotImplementedError52 53    @abstractmethod54    def moderation_for_inputs(self, inputs: dict, query: str = "") -> ModerationInputsResult:55        """56        Moderation for inputs.57        After the user inputs, this method will be called to perform sensitive content review58        on the user inputs and return the processed results.59 60        :param inputs: user inputs61        :param query: query string (required in chat app)62        :return:63        """64        raise NotImplementedError65 66    @abstractmethod67    def moderation_for_outputs(self, text: str) -> ModerationOutputsResult:68        """69        Moderation for outputs.70        When LLM outputs content, the front end will pass the output content (may be segmented)71        to this method for sensitive content review, and the output content will be shielded if the review fails.72 73        :param text: LLM output content74        :return:75        """76        raise NotImplementedError77 78    @classmethod79    def _validate_inputs_and_outputs_config(cls, config: dict, is_preset_response_required: bool) -> None:80        # inputs_config81        inputs_config = config.get("inputs_config")82        if not isinstance(inputs_config, dict):83            raise ValueError("inputs_config must be a dict")84 85        # outputs_config86        outputs_config = config.get("outputs_config")87        if not isinstance(outputs_config, dict):88            raise ValueError("outputs_config must be a dict")89 90        inputs_config_enabled = inputs_config.get("enabled")91        outputs_config_enabled = outputs_config.get("enabled")92        if not inputs_config_enabled and not outputs_config_enabled:93            raise ValueError("At least one of inputs_config or outputs_config must be enabled")94 95        # preset_response96        if not is_preset_response_required:97            return98 99        if inputs_config_enabled:100            if not inputs_config.get("preset_response"):101                raise ValueError("inputs_config.preset_response is required")102 103            if len(inputs_config.get("preset_response")) > 100:104                raise ValueError("inputs_config.preset_response must be less than 100 characters")105 106        if outputs_config_enabled:107            if not outputs_config.get("preset_response"):108                raise ValueError("outputs_config.preset_response is required")109 110            if len(outputs_config.get("preset_response")) > 100:111                raise ValueError("outputs_config.preset_response must be less than 100 characters")112 113 114class ModerationError(Exception):115    pass116