【Dify(v1.x) 核心源码深入解析】moderation 模块

小爷毛毛（卓寿杰）

于 2025-05-19 13:48:04 发布

阅读量45

点赞数

分类专栏：大模型AIGC 系统架构与解决方案 Dify 文章标签：自然语言处理人工智能 python 语言模型深度学习

本文链接：https://blog.csdn.net/u011239443/article/details/148061750

版权

大模型AIGC 同时被 3 个专栏收录

84 篇文章 ¥29.90 ¥99.00

订阅专栏

超级会员免费看

系统架构与解决方案

29 篇文章

订阅专栏

Dify

14 篇文章

订阅专栏

重磅推荐专栏：
《大模型AIGC》
《课程大纲》
《知识星球》

本专栏致力于探索和讨论当今最前沿的技术趋势和应用领域，包括但不限于ChatGPT和Stable Diffusion等。我们将深入研究大型模型的开发和应用，以及与之相关的人工智能生成内容（AIGC）技术。通过深入的技术解析和实践经验分享，旨在帮助读者更好地理解和应用这些领域的最新进展

模块简介

Dify 的 moderation 模块是一套功能强大的内容审核系统，旨在对用户输入和模型输出进行敏感内容检测与处理，以确保内容的安全性和合规性。它提供了灵活多样的审核策略，可帮助开发者根据实际需求定制审核流程，广泛应用于各种需要内容管控的应用场景，如聊天机器人、内容生成平台等。

模块架构

整体架构图

类图

核心组件详解

ModerationFactory - 审核工厂类

这是 moderation 模块的核心入口，负责根据配置创建具体的审核实例。

代码片段及注释：

class ModerationFactory:
    __extension_instance: Moderation

    def __init__(self, name: str, app_id: str, tenant_id: str, config: dict) -> None:
        # 根据审核类型名称获取对应的审核类
        extension_class = code_based_extension.extension_class(ExtensionModule.MODERATION, name)
        # 创建具体审核实例
        self.__extension_instance = extension_class(app_id, tenant_id, config)

    @classmethod
    def validate_config(cls, name: str, tenant_id: str, config: dict) -> None:
        # 验证审核配置
        code_based_extension.validate_form_schema(ExtensionModule.MODERATION, name, config)
        extension_class = code_based_extension.extension_class(ExtensionModule.MODERATION, name)
        # 调用具体审核类的配置验证方法
        extension_class.validate_config(tenant_id, config)  # type: ignore

    def moderation_for_inputs(self, inputs: dict, query: str = "") -> ModerationInputsResult:
        # 执行输入内容审核
        return self.__extension_instance.moderation_for_inputs(inputs, query)

    def moderation_for_outputs(self, text: str) -> ModerationOutputsResult:
        # 执行输出内容审核
        return self.__extension_instance.moderation_for_outputs(text)

Moderation - 审核基类

定义了审核的基本规范和通用逻辑，所有具体审核类都需继承自该基类。

代码片段及注释：

class Moderation(Extensible, ABC):
    """
    审核基类，定义审核的基本规范和通用逻辑
    """

    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:
        """
        验证审核配置
        """
        raise NotImplementedError

    @abstractmethod
    def moderation_for_inputs(self, inputs: dict, query: str = "") -> ModerationInputsResult:
        """
        执行输入内容审核
        """
        raise NotImplementedError

    @abstractmethod
    def moderation_for_outputs(self, text: str) -> ModerationOutputsResult:
        """
        执行输出内容审核
        """
        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", 0)) > 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", 0)) > 100:
                raise ValueError("outputs_config.preset_response must be less than 100 characters")

OpenAIModeration - OpenAI 审核类

利用 OpenAI 的审核模型对内容进行敏感性检测。

代码片段及注释：

class OpenAIModeration(Moderation):
    name: str = "openai_moderation"

    @classmethod
    def validate_config(cls, tenant_id: str, config: dict) -> None:
        # 验证 OpenAI 审核配置
        cls._validate_inputs_and_outputs_config(config, True)

    def moderation_for_inputs(self, inputs: dict, query: str = "") -> ModerationInputsResult:
        # 执行输入内容审核
        flagged = False
        preset_response = ""
        if self.config is None:
            raise ValueError("The config is not set.")

        if self.config["inputs_config"]["enabled"]:
            preset_response = self.config["inputs_config"]["preset_response"]

            if query:
                inputs["query__"] = query
            flagged = self._is_violated(inputs)

        return ModerationInputsResult(
            flagged=flagged, action=ModerationAction.DIRECT_OUTPUT, preset_response=preset_response
        )

    def moderation_for_outputs(self, text: str) -> ModerationOutputsResult:
        # 执行输出内容审核
        flagged = False
        preset_response = ""
        if self.config is None:
            raise ValueError("The config is not set.")

        if self.config["outputs_config"]["enabled"]:
            flagged = self._is_violated({"text": text})
            preset_response = self.config["outputs_config"]["preset_response"]

        return ModerationOutputsResult(
            flagged=flagged, action=ModerationAction.DIRECT_OUTPUT, preset_response=preset_response
        )

    def _is_violated(self, inputs: dict):
        # 判断内容是否违规
        text = "\n".join(str(inputs.values()))
        model_manager = ModelManager()
        # 获取 OpenAI 审核模型实例
        model_instance = model_manager.get_model_instance(
            tenant_id=self.tenant_id, provider="openai", model_type=ModelType.MODERATION, model="text-moderation-stable"
        )

        # 调用 OpenAI 审核模型进行检测
        openai_moderation = model_instance.invoke_moderation(text=text)

        return openai_moderation

KeywordsModeration - 关键词审核类

通过匹配预设的敏感关键词来检测内容是否违规。

代码片段及注释：

class KeywordsModeration(Moderation):
    name: str = "keywords"

    @classmethod
    def validate_config(cls, tenant_id: str, config: dict) -> None:
        # 验证关键词审核配置
        cls._validate_inputs_and_outputs_config(config, True)

        if not config.get("keywords"):
            raise ValueError("keywords is required")

        if len(config.get("keywords", [])) > 10000:
            raise ValueError("keywords length must be less than 10000")

        keywords_row_len = config["keywords"].split("\n")
        if len(keywords_row_len) > 100:
            raise ValueError("the number of rows for the keywords must be less than 100")

    def moderation_for_inputs(self, inputs: dict, query: str = "") -> ModerationInputsResult:
        # 执行输入内容审核
        flagged = False
        preset_response = ""
        if self.config is None:
            raise ValueError("The config is not set.")

        if self.config["inputs_config"]["enabled"]:
            preset_response = self.config["inputs_config"]["preset_response"]

            if query:
                inputs["query__"] = query

            # 获取过滤后的关键词列表
            keywords_list = [keyword for keyword in self.config["keywords"].split("\n") if keyword]

            flagged = self._is_violated(inputs, keywords_list)

        return ModerationInputsResult(
            flagged=flagged, action=ModerationAction.DIRECT_OUTPUT, preset_response=preset_response
        )

    def moderation_for_outputs(self, text: str) -> ModerationOutputsResult:
        # 执行输出内容审核
        flagged = False
        preset_response = ""
        if self.config is None:
            raise ValueError("The config is not set.")

        if self.config["outputs_config"]["enabled"]:
            # 获取过滤后的关键词列表
            keywords_list = [keyword for keyword in self.config["keywords"].split("\n") if keyword]

            flagged = self._is_violated({"text": text}, keywords_list)
            preset_response = self.config["outputs_config"]["preset_response"]

        return ModerationOutputsResult(
            flagged=flagged, action=ModerationAction.DIRECT_OUTPUT, preset_response=preset_response
        )

    def _is_violated(self, inputs: dict, keywords_list: list) -> bool:
        # 判断内容是否包含敏感关键词
        return any(self._check_keywords_in_value(keywords_list, value) for value in inputs.values())

    def _check_keywords_in_value(self, keywords_list: Sequence[str], value: Any) -> bool:
        # 检查单个值是否包含敏感关键词
        return any(keyword.lower() in str(value).lower() for keyword in keywords_list)

ApiModeration - API 审核类

通过调用外部 API 进行内容审核。

代码片段及注释：

class ApiModeration(Moderation):
    name: str = "api"

    @classmethod
    def validate_config(cls, tenant_id: str, config: dict) -> None:
        # 验证 API 审核配置
        cls._validate_inputs_and_outputs_config(config, False)

        api_based_extension_id = config.get("api_based_extension_id")
        if not api_based_extension_id:
            raise ValueError("api_based_extension_id is required")

        extension = cls._get_api_based_extension(tenant_id, api_based_extension_id)
        if not extension:
            raise ValueError("API-based Extension not found. Please check it again.")

    def moderation_for_inputs(self, inputs: dict, query: str = "") -> ModerationInputsResult:
        # 执行输入内容审核
        flagged = False
        preset_response = ""
        if self.config is None:
            raise ValueError("The config is not set.")

        if self.config["inputs_config"]["enabled"]:
            params = ModerationInputParams(app_id=self.app_id, inputs=inputs, query=query)

            # 调用外部 API 进行审核
            result = self._get_config_by_requestor(APIBasedExtensionPoint.APP_MODERATION_INPUT, params.model_dump())
            return ModerationInputsResult(**result)

        return ModerationInputsResult(
            flagged=flagged, action=ModerationAction.DIRECT_OUTPUT, preset_response=preset_response
        )

    def moderation_for_outputs(self, text: str) -> ModerationOutputsResult:
        # 执行输出内容审核
        flagged = False
        preset_response = ""
        if self.config is None:
            raise ValueError("The config is not set.")

        if self.config["outputs_config"]["enabled"]:
            params = ModerationOutputParams(app_id=self.app_id, text=text)

            # 调用外部 API 进行审核
            result = self._get_config_by_requestor(APIBasedExtensionPoint.APP_MODERATION_OUTPUT, params.model_dump())
            return ModerationOutputsResult(**result)

        return ModerationOutputsResult(
            flagged=flagged, action=ModerationAction.DIRECT_OUTPUT, preset_response=preset_response
        )

    def _get_config_by_requestor(self, extension_point: APIBasedExtensionPoint, params: dict) -> dict:
        # 发起对外部 API 的请求
        if self.config is None:
            raise ValueError("The config is not set.")
        extension = self._get_api_based_extension(self.tenant_id, self.config.get("api_based_extension_id", ""))
        if not extension:
            raise ValueError("API-based Extension not found. Please check it again.")
        requestor = APIBasedExtensionRequestor(extension.api_endpoint, decrypt_token(self.tenant_id, extension.api_key))

        result = requestor.request(extension_point, params)
        return result

    @staticmethod
    def _get_api_based_extension(tenant_id: str, api_based_extension_id: str) -> Optional[APIBasedExtension]:
        # 获取 API-based 扩展
        extension = (
            db.session.query(APIBasedExtension)
            .filter(APIBasedExtension.tenant_id == tenant_id, APIBasedExtension.id == api_based_extension_id)
            .first()
        )

        return extension

审核流程详解

输入审核流程

流程图

详细步骤

调用者通过 ModerationFactory 创建具体的审核实例，传入审核类型、应用 ID、租户 ID 和配置等参数。
根据审核类型，ModerationFactory 会创建对应的审核类实例（如 OpenAIModeration、KeywordsModeration 或 ApiModeration）。
调用审核实例的 moderation_for_inputs 方法，传入用户输入内容和查询字符串（可选）。
在审核类中，根据配置判断是否启用输入审核。如果启用，则执行相应的审核逻辑：
- OpenAIModeration ：将输入内容转换为文本，调用 OpenAI 审核模型进行检测，判断内容是否违规。
- KeywordsModeration ：将输入内容与预设的敏感关键词列表进行匹配，判断是否包含敏感关键词。
- ApiModeration ：将输入内容封装为请求参数，调用外部 API 进行审核，获取审核结果。
根据审核结果，返回 ModerationInputsResult 对象，包含是否违规、审核动作（如直接输出或覆盖）和预设响应等内容。

输出审核流程

流程图

详细步骤

调用者通过 ModerationFactory 创建具体的审核实例，传入审核类型、应用 ID、租户 ID 和配置等参数。
根据审核类型，ModerationFactory 会创建对应的审核类实例（如 OpenAIModeration、KeywordsModeration 或 ApiModeration）。
调用审核实例的 moderation_for_outputs 方法，传入模型输出的文本内容。
在审核类中，根据配置判断是否启用输出审核。如果启用，则执行相应的审核逻辑：
- OpenAIModeration ：将输出内容转换为文本，调用 OpenAI 审核模型进行检测，判断内容是否违规。
- KeywordsModeration ：将输出内容与预设的敏感关键词列表进行匹配，判断是否包含敏感关键词。
- ApiModeration ：将输出内容封装为请求参数，调用外部 API 进行审核，获取审核结果。
根据审核结果，返回 ModerationOutputsResult 对象，包含是否违规、审核动作（如直接输出或覆盖）和预设响应等内容。

审核结果处理

审核结果主要通过 ModerationInputsResult 和 ModerationOutputsResult 两个类来表示，它们包含了审核的关键信息，如下表所示：

属性名称	含义	类型
flagged	是否标记为违规内容	bool
action	审核动作（DIRECT_OUTPUT 或 OVERRIDDEN）	ModerationAction
preset_response	预设响应内容	str
inputs	审核后的输入内容（仅 ModerationInputsResult 有）	dict
query	审核后的查询字符串（仅 ModerationInputsResult 有）	str
text	审核后的输出文本（仅 ModerationOutputsResult 有）	str