| """Pass input through a moderation endpoint.""" |
|
|
| from typing import Any, Dict, List, Optional |
|
|
| from langchain_core.callbacks import ( |
| AsyncCallbackManagerForChainRun, |
| CallbackManagerForChainRun, |
| ) |
| from langchain_core.utils import check_package_version, get_from_dict_or_env |
| from pydantic import Field, model_validator |
|
|
| from langchain.chains.base import Chain |
|
|
|
|
| class OpenAIModerationChain(Chain): |
| """Pass input through a moderation endpoint. |
| |
| To use, you should have the ``openai`` python package installed, and the |
| environment variable ``OPENAI_API_KEY`` set with your API key. |
| |
| Any parameters that are valid to be passed to the openai.create call can be passed |
| in, even if not explicitly saved on this class. |
| |
| Example: |
| .. code-block:: python |
| |
| from langchain.chains import OpenAIModerationChain |
| moderation = OpenAIModerationChain() |
| """ |
|
|
| client: Any = None |
| async_client: Any = None |
| model_name: Optional[str] = None |
| """Moderation model name to use.""" |
| error: bool = False |
| """Whether or not to error if bad content was found.""" |
| input_key: str = "input" |
| output_key: str = "output" |
| openai_api_key: Optional[str] = None |
| openai_organization: Optional[str] = None |
| openai_pre_1_0: bool = Field(default=False) |
|
|
| @model_validator(mode="before") |
| @classmethod |
| def validate_environment(cls, values: Dict) -> Any: |
| """Validate that api key and python package exists in environment.""" |
| openai_api_key = get_from_dict_or_env( |
| values, "openai_api_key", "OPENAI_API_KEY" |
| ) |
| openai_organization = get_from_dict_or_env( |
| values, |
| "openai_organization", |
| "OPENAI_ORGANIZATION", |
| default="", |
| ) |
| try: |
| import openai |
|
|
| openai.api_key = openai_api_key |
| if openai_organization: |
| openai.organization = openai_organization |
| values["openai_pre_1_0"] = False |
| try: |
| check_package_version("openai", gte_version="1.0") |
| except ValueError: |
| values["openai_pre_1_0"] = True |
| if values["openai_pre_1_0"]: |
| values["client"] = openai.Moderation |
| else: |
| values["client"] = openai.OpenAI(api_key=openai_api_key) |
| values["async_client"] = openai.AsyncOpenAI(api_key=openai_api_key) |
|
|
| except ImportError: |
| raise ImportError( |
| "Could not import openai python package. " |
| "Please install it with `pip install openai`." |
| ) |
| return values |
|
|
| @property |
| def input_keys(self) -> List[str]: |
| """Expect input key. |
| |
| :meta private: |
| """ |
| return [self.input_key] |
|
|
| @property |
| def output_keys(self) -> List[str]: |
| """Return output key. |
| |
| :meta private: |
| """ |
| return [self.output_key] |
|
|
| def _moderate(self, text: str, results: Any) -> str: |
| if self.openai_pre_1_0: |
| condition = results["flagged"] |
| else: |
| condition = results.flagged |
| if condition: |
| error_str = "Text was found that violates OpenAI's content policy." |
| if self.error: |
| raise ValueError(error_str) |
| else: |
| return error_str |
| return text |
|
|
| def _call( |
| self, |
| inputs: Dict[str, Any], |
| run_manager: Optional[CallbackManagerForChainRun] = None, |
| ) -> Dict[str, Any]: |
| text = inputs[self.input_key] |
| if self.openai_pre_1_0: |
| results = self.client.create(text) |
| output = self._moderate(text, results["results"][0]) |
| else: |
| results = self.client.moderations.create(input=text) |
| output = self._moderate(text, results.results[0]) |
| return {self.output_key: output} |
|
|
| async def _acall( |
| self, |
| inputs: Dict[str, Any], |
| run_manager: Optional[AsyncCallbackManagerForChainRun] = None, |
| ) -> Dict[str, Any]: |
| if self.openai_pre_1_0: |
| return await super()._acall(inputs, run_manager=run_manager) |
| text = inputs[self.input_key] |
| results = await self.async_client.moderations.create(input=text) |
| output = self._moderate(text, results.results[0]) |
| return {self.output_key: output} |
|
|
