Source code for langchain_experimental.comprehend_moderation.amazon_comprehend_moderation

from typing import Any, Dict, List, Optional

from langchain.chains.base import Chain
from langchain_core.callbacks.manager import CallbackManagerForChainRun

from langchain_experimental.comprehend_moderation.base_moderation import BaseModeration
from langchain_experimental.comprehend_moderation.base_moderation_callbacks import (
    BaseModerationCallbackHandler,
)
from langchain_experimental.comprehend_moderation.base_moderation_config import (
    BaseModerationConfig,
)
from langchain_experimental.pydantic_v1 import root_validator


class AmazonComprehendModerationChain(Chain):
    """基于`Amazon Comprehend`服务的内容审核链。

    请访问 https://aws.amazon.com/comprehend/ 了解更多信息。"""

    output_key: str = "output"  #: :meta private:
    """用于在数据容器中获取/存储输出的键。默认为`output`"""

    input_key: str = "input"  #: :meta private:
    """用于在数据容器中获取/存储输入的键。默认为`input`"""

    moderation_config: BaseModerationConfig = BaseModerationConfig()
    """配置设置用于内容的审核，默认为具有默认值的BaseModerationConfig"""

    client: Optional[Any] = None
    """创建与Amazon Comprehend连接的boto3客户端对象"""

    region_name: Optional[str] = None
    """AWS区域，例如`us-west-2`。如果未在此处提供，则回退到AWS_DEFAULT_REGION环境变量或~/.aws/config中指定的区域。"""

    credentials_profile_name: Optional[str] = None
    """~/.aws/credentials 或 ~/.aws/config 文件中配置文件的名称，其中包含指定的访问密钥或角色信息。
如果未指定，则将使用默认的凭据配置文件，或者如果在EC2实例上，则将使用来自IMDS的凭据。
参见：https://boto3.amazonaws.com/v1/documentation/api/latest/guide/credentials.html"""

    moderation_callback: Optional[BaseModerationCallbackHandler] = None
    """用于审核的回调处理程序，这与常规回调不同，可以与其同时使用。"""

    unique_id: Optional[str] = None
    """一个可以用来识别或分组用户或会话的唯一标识符。"""

    @root_validator(pre=True)
    def create_client(cls, values: Dict[str, Any]) -> Dict[str, Any]:
        """创建一个Amazon Comprehend客户端。

参数：
    values (Dict[str, Any]): 包含配置数值的字典。

返回：
    Dict[str, Any]: 包含更新后的配置数值的字典，包括Amazon Comprehend客户端。

引发：
    ModuleNotFoundError: 如果未安装'boto3'包。
    ValueError: 如果导入'boto3'或加载AWS凭证时出现问题。

示例：
.. code-block:: python

    config = {
        "credentials_profile_name": "my-profile",
        "region_name": "us-west-2"
    }
    updated_config = create_client(config)
    comprehend_client = updated_config["client"]

"""

        if values.get("client") is not None:
            return values
        try:
            import boto3

            if values.get("credentials_profile_name"):
                session = boto3.Session(profile_name=values["credentials_profile_name"])
            else:
                # use default credentials
                session = boto3.Session()

            client_params = {}
            if values.get("region_name"):
                client_params["region_name"] = values["region_name"]

            values["client"] = session.client("comprehend", **client_params)

            return values
        except ImportError:
            raise ModuleNotFoundError(
                "Could not import boto3 python package. "
                "Please install it with `pip install boto3`."
            )
        except Exception as e:
            raise ValueError(
                "Could not load credentials to authenticate with AWS client. "
                "Please check that credentials in the specified "
                f"profile name are valid. {e}"
            ) from e

    @property
    def output_keys(self) -> List[str]:
        """返回一个输出键的列表。

该方法定义了用于访问由链或函数产生的输出值的输出键。它确保指定的键可用于访问输出。

返回：
    List[str]：一个输出键的列表。

注意：
    此方法被视为私有方法，可能不适合直接外部使用。
"""
        return [self.output_key]

    @property
    def input_keys(self) -> List[str]:
        """返回预期由提示框接收的输入键列表。

该方法定义了提示框期望的输入键，以便执行其处理。它确保指定的键可用于向提示框提供输入。

返回：
    List[str]：输入键的列表。

注意：
    此方法被视为私有方法，可能不适合直接外部使用。
"""
        return [self.input_key]

    def _call(
        self,
        inputs: Dict[str, Any],
        run_manager: Optional[CallbackManagerForChainRun] = None,
    ) -> Dict[str, str]:
        """执行输入文本上的审查过程，并返回处理后的输出。

这个内部方法在输入文本上执行审查过程。它将输入提示值转换为纯文本，应用指定的过滤器，然后将过滤后的输出转换回适当的提示值对象。此外，它提供了使用提供的“run_manager”记录运行信息的选项。

参数:
    inputs: 包含输入值的字典
    run_manager: 用于处理与运行相关事件的运行管理器。默认值为None

返回:
    Dict[str, str]: 包含审查过程处理后的输出的字典。

Raises:
    ValueError: 如果在审查过程中出现错误
"""

        if run_manager:
            run_manager.on_text("Running AmazonComprehendModerationChain...\n")

        moderation = BaseModeration(
            client=self.client,
            config=self.moderation_config,
            moderation_callback=self.moderation_callback,
            unique_id=self.unique_id,
            run_manager=run_manager,
        )
        response = moderation.moderate(prompt=inputs[self.input_keys[0]])

        return {self.output_key: response}