Feature/content safety runnable

docs/docs/integrations/chains/azure_ai_content_safety.ipynb

@@ -0,0 +1,138 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# `AzureAIContentSafetyChain`"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "> [Azure AI Content Safety Chain](https://learn.microsoft.com/python/api/overview/azure/ai-contentsafety-readme?view=azure-python) is a wrapper around\n",
+    "> the Azure AI Content Safety service, implemented in LangChain using the LangChain \n",
+    "> [Runnables](https://python.langchain.com/docs/how_to/lcel_cheatsheet/) base class to allow use in a Runnables Sequence.\n",
+    "\n",
+    "The Class can be used to stop or filter content based on the Azure AI Content Safety policy."
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Example Usage"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "Get the required imports, here we will use a `ChatPromptTemplate` for convenience and the `AzureChatOpenAI`, however, any LangChain integrated model will work in a chain."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 2,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import os\n",
+    "\n",
+    "from langchain_community.chains.azure_content_safety_chain import (\n",
+    "    AzureOpenAIContentSafetyChain,\n",
+    ")\n",
+    "from langchain_core.prompts import ChatPromptTemplate\n",
+    "from langchain_openai import AzureChatOpenAI"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "moderate = AzureOpenAIContentSafetyChain()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "model = AzureChatOpenAI(\n",
+    "    openai_api_version=os.environ[\"OPENAI_API_VERSION\"],\n",
+    "    azure_deployment=os.environ[\"COMPLETIONS_MODEL\"],\n",
+    "    azure_endpoint=os.environ[\"AZURE_OPENAI_ENDPOINT\"],\n",
+    "    api_key=os.environ[\"AZURE_OPENAI_API_KEY\"],\n",
+    ")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "prompt = ChatPromptTemplate.from_messages([(\"system\", \"repeat after me: {input}\")])"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "Combine the objects to create a LangChain RunnablesSequence"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "moderated_chain = moderate | prompt | model"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "response = moderated_chain.invoke({\"input\": \"I hate you!\"})"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "response.content"
+   ]
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.11.4"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 2
+}

libs/community/extended_testing_deps.txt

@@ -4,6 +4,7 @@ anthropic>=0.3.11,<0.4
 arxiv>=1.4,<2
 assemblyai>=0.17.0,<0.18
 atlassian-python-api>=3.36.0,<4
+azure-ai-contentsafety>=1.0.0
 azure-ai-documentintelligence>=1.0.0b1,<2
 azure-identity>=1.15.0,<2
 azure-search-documents==11.4.0

libs/community/langchain_community/chains/azure_content_safety_chain.py

@@ -0,0 +1,163 @@
+"""Pass input through an azure content safety resource."""
+
+from typing import Any, Dict, List, Optional
+
+from langchain.chains.base import Chain
+from langchain_core.callbacks import (
+    CallbackManagerForChainRun,
+)
+from langchain_core.exceptions import LangChainException
+from langchain_core.utils import get_from_dict_or_env
+from pydantic import model_validator
+
+
+class AzureHarmfulContentError(LangChainException):
+    """Exception for handling harmful content detected
+    in input for a model or chain according to Azure's
+    content safety policy."""
+
+    def __init__(
+        self,
+        input: str,
+    ):
+        """Constructor
+
+        Args:
+            input (str): The input given by the user to the model.
+        """
+        self.input = input
+        self.message = "The input has breached Azure's Content Safety Policy"
+        super().__init__(self.message)
+
+
+class AzureAIContentSafetyChain(Chain):
+    """
+    A wrapper for the Azure AI Content Safety API in a Runnable form.
+    Allows for harmful content detection and filtering before input is
+    provided to a model.
+
+    **Note**:
+    This Service will filter input that shows any sign of harmful content,
+    this is non-configurable.
+
+    Attributes:
+        error (bool): Whether to raise an error if harmful content is detected.
+        content_safety_key (Optional[str]): API key for Azure Content Safety.
+        content_safety_endpoint (Optional[str]): Endpoint URL for Azure Content Safety.
+
+    Setup:
+        1. Follow the instructions here to deploy Azure AI Content Safety:
+            https://learn.microsoft.com/azure/ai-services/content-safety/overview
+
+        2. Install ``langchain`` ``langchain_community`` and set the following
+        environment variables:
+
+        .. code-block:: bash
+
+            pip install -U langchain langchain-community
+
+            export AZURE_CONTENT_SAFETY_KEY="your-api-key"
+            export AZURE_CONTENT_SAFETY_ENDPOINT="https://your-endpoint.azure.com/"
+
+
+    Example Usage:
+        .. code-block:: python
+
+            from langchain_community.chains import AzureAIContentSafetyChain
+            from langchain_openai import AzureChatOpenAI
+
+            moderate = AzureAIContentSafetyChain()
+            prompt = ChatPromptTemplate.from_messages([("system",
+                    "repeat after me: {input}")])
+            model = AzureChatOpenAI()
+
+            moderated_chain = moderate | prompt | model
+
+            moderated_chain.invoke({"input": "Hey, How are you?"})
+    """
+
+    client: Any = None  #: :meta private:
+    error: bool = False
+    """Whether or not to error if bad content was found."""
+    input_key: str = "input"  #: :meta private:
+    output_key: str = "output"  #: :meta private:
+    content_safety_key: Optional[str] = None
+    content_safety_endpoint: Optional[str] = None
+
+    @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]
+
+    @model_validator(mode="before")
+    @classmethod
+    def validate_environment(cls, values: Dict) -> Any:
+        """Validate that api key and python package exists in environment."""
+        content_safety_key = get_from_dict_or_env(
+            values, "content_safety_key", "CONTENT_SAFETY_API_KEY"
+        )
+        content_safety_endpoint = get_from_dict_or_env(
+            values, "content_safety_endpoint", "CONTENT_SAFETY_ENDPOINT"
+        )
+        try:
+            import azure.ai.contentsafety as sdk
+            from azure.core.credentials import AzureKeyCredential
+
+            values["client"] = sdk.ContentSafetyClient(
+                endpoint=content_safety_endpoint,
+                credential=AzureKeyCredential(content_safety_key),
+            )
+
+        except ImportError:
+            raise ImportError(
+                "azure-ai-contentsafety is not installed. "
+                "Run `pip install azure-ai-contentsafety` to install."
+            )
+        return values
+
+    def _detect_harmful_content(self, text: str, results: Any) -> str:
+        contains_harmful_content = False
+
+        for category in results:
+            if category["severity"] > 0:
+                contains_harmful_content = True
+
+        if contains_harmful_content:
+            error_str = (
+                "The input text contains harmful content "
+                "according to Azure OpenAI's content policy"
+            )
+            if self.error:
+                raise AzureHarmfulContentError(input=text)
+            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]
+
+        from azure.ai.contentsafety.models import AnalyzeTextOptions
+
+        request = AnalyzeTextOptions(text=text)
+        response = self.client.analyze_text(request)
+
+        result = response.categories_analysis
+        output = self._detect_harmful_content(text, result)
+
+        return {self.input_key: output, self.output_key: output}

libs/community/tests/unit_tests/chains/test_azure_content_safety.py

@@ -0,0 +1,72 @@
+"""Tests for the Azure AI Content Safety Chain."""
+
+from typing import Any
+
+import pytest
+
+from langchain_community.chains.azure_content_safety_chain import (
+    AzureAIContentSafetyChain,
+    AzureHarmfulContentError,
+)
+
+
+@pytest.mark.requires("azure.ai.contentsafety")
+def test_content_safety(mocker: Any) -> None:
+    mocker.patch("azure.ai.contentsafety.ContentSafetyClient", autospec=True)
+    mocker.patch("azure.core.credentials.AzureKeyCredential", autospec=True)
+
+    key = "key"
+    endpoint = "endpoint"
+
+    chain = AzureAIContentSafetyChain(
+        content_safety_key=key, content_safety_endpoint=endpoint
+    )
+    assert chain.content_safety_key == key
+    assert chain.content_safety_endpoint == endpoint
+
+
+@pytest.mark.requires("azure.ai.contentsafety")
+def test_raise_error_when_harmful_content_detected(mocker: Any) -> None:
+    key = "key"
+    endpoint = "endpoint"
+
+    mocker.patch("azure.core.credentials.AzureKeyCredential", autospec=True)
+    mocker.patch("azure.ai.contentsafety.ContentSafetyClient", autospec=True)
+    chain = AzureAIContentSafetyChain(
+        content_safety_key=key, content_safety_endpoint=endpoint, error=True
+    )
+
+    mock_content_client = mocker.Mock()
+    mock_content_client.analyze_text.return_value.categories_analysis = [
+        {"Category": "Harm", "severity": 1}
+    ]
+
+    chain.client = mock_content_client
+
+    text = "This text contains harmful content"
+    with pytest.raises(AzureHarmfulContentError):
+        chain._call({chain.input_key: text})
+
+
+@pytest.mark.requires("azure.ai.contentsafety")
+def test_no_harmful_content_detected(mocker: Any) -> None:
+    key = "key"
+    endpoint = "endpoint"
+
+    mocker.patch("azure.core.credentials.AzureKeyCredential", autospec=True)
+    mocker.patch("azure.ai.contentsafety.ContentSafetyClient", autospec=True)
+    chain = AzureAIContentSafetyChain(
+        content_safety_key=key, content_safety_endpoint=endpoint, error=True
+    )
+
+    mock_content_client = mocker.Mock()
+    mock_content_client.analyze_text.return_value.categories_analysis = [
+        {"Category": "Harm", "severity": 0}
+    ]
+
+    chain.client = mock_content_client
+
+    text = "This text contains no harmful content"
+    output = chain._call({chain.input_key: text})
+
+    assert output[chain.output_key] == text