| from abc import ABC, abstractmethod |
| from enum import Enum |
| from typing import Optional |
|
|
| from pydantic import BaseModel |
|
|
| from core.extension.extensible import Extensible, ExtensionModule |
|
|
|
|
| class ModerationAction(Enum): |
| DIRECT_OUTPUT = "direct_output" |
| OVERRIDDEN = "overridden" |
|
|
|
|
| class ModerationInputsResult(BaseModel): |
| flagged: bool = False |
| action: ModerationAction |
| preset_response: str = "" |
| inputs: dict = {} |
| query: str = "" |
|
|
|
|
| class ModerationOutputsResult(BaseModel): |
| flagged: bool = False |
| action: ModerationAction |
| preset_response: str = "" |
| text: str = "" |
|
|
|
|
| class Moderation(Extensible, ABC): |
| """ |
| The base class of moderation. |
| """ |
|
|
| module: ExtensionModule = ExtensionModule.MODERATION |
|
|
| def __init__(self, app_id: str, tenant_id: str, config: Optional[dict] = None) -> None: |
| super().__init__(tenant_id, config) |
| self.app_id = app_id |
|
|
| @classmethod |
| @abstractmethod |
| def validate_config(cls, tenant_id: str, config: dict) -> None: |
| """ |
| Validate the incoming form config data. |
| |
| :param tenant_id: the id of workspace |
| :param config: the form config data |
| :return: |
| """ |
| raise NotImplementedError |
|
|
| @abstractmethod |
| def moderation_for_inputs(self, inputs: dict, query: str = "") -> ModerationInputsResult: |
| """ |
| Moderation for inputs. |
| After the user inputs, this method will be called to perform sensitive content review |
| on the user inputs and return the processed results. |
| |
| :param inputs: user inputs |
| :param query: query string (required in chat app) |
| :return: |
| """ |
| raise NotImplementedError |
|
|
| @abstractmethod |
| def moderation_for_outputs(self, text: str) -> ModerationOutputsResult: |
| """ |
| Moderation for outputs. |
| When LLM outputs content, the front end will pass the output content (may be segmented) |
| to this method for sensitive content review, and the output content will be shielded if the review fails. |
| |
| :param text: LLM output content |
| :return: |
| """ |
| raise NotImplementedError |
|
|
| @classmethod |
| def _validate_inputs_and_outputs_config(cls, config: dict, is_preset_response_required: bool) -> None: |
| |
| inputs_config = config.get("inputs_config") |
| if not isinstance(inputs_config, dict): |
| raise ValueError("inputs_config must be a dict") |
|
|
| |
| outputs_config = config.get("outputs_config") |
| if not isinstance(outputs_config, dict): |
| raise ValueError("outputs_config must be a dict") |
|
|
| inputs_config_enabled = inputs_config.get("enabled") |
| outputs_config_enabled = outputs_config.get("enabled") |
| if not inputs_config_enabled and not outputs_config_enabled: |
| raise ValueError("At least one of inputs_config or outputs_config must be enabled") |
|
|
| |
| if not is_preset_response_required: |
| return |
|
|
| if inputs_config_enabled: |
| if not inputs_config.get("preset_response"): |
| raise ValueError("inputs_config.preset_response is required") |
|
|
| if len(inputs_config.get("preset_response")) > 100: |
| raise ValueError("inputs_config.preset_response must be less than 100 characters") |
|
|
| if outputs_config_enabled: |
| if not outputs_config.get("preset_response"): |
| raise ValueError("outputs_config.preset_response is required") |
|
|
| if len(outputs_config.get("preset_response")) > 100: |
| raise ValueError("outputs_config.preset_response must be less than 100 characters") |
|
|
|
|
| class ModerationError(Exception): |
| pass |
|
|
