Skip to content

Azure Content Safety

any_guardrail.guardrails.azure_content_safety.azure_content_safety

AzureContentSafety

Bases: Guardrail

Guardrail implementation using Azure Content Safety service.

Source code in src/any_guardrail/guardrails/azure_content_safety/azure_content_safety.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
class AzureContentSafety(Guardrail):
    """Guardrail implementation using Azure Content Safety service."""

    SUPPORTED_MODELS: ClassVar = ["azure-content-safety"]

    def __init__(
        self,
        endpoint: str | None = None,
        api_key: str | None = None,
        threshold: int = 2,
        score_type: str = "max",
        blocklist_names: list[str] | None = None,
    ) -> None:
        """Initialize Azure Content Safety client.

        Args:
            endpoint (str): The endpoint URL for the Azure Content Safety service.
            api_key (str): The API key for authenticating with the service.
            threshold (int): The threshold for determining if content is unsafe.
            score_type (str): The type of score to use ("max" or "avg").
            blocklist_names (List[str] | None): List of blocklist names to use for content evaluation.

        """
        if api_key:
            credential = AzureKeyCredential(api_key)
        else:
            try:
                credential = AzureKeyCredential(os.environ["CONTENT_SAFETY_KEY"])
            except KeyError as e:
                msg = "CONTENT_SAFETY_KEY environment variable is not set. Either provide an api_key or set the environment variable."
                raise KeyError(msg) from e
        if not endpoint:
            try:
                endpoint = os.environ["CONTENT_SAFETY_ENDPOINT"]
            except KeyError as e:
                msg = "CONTENT_SAFETY_ENDPOINT environment variable is not set. Either provide an endpoint or set the environment variable."
                raise KeyError(msg) from e

        self.client = ContentSafetyClient(endpoint=endpoint, credential=credential)
        self.blocklist_client = BlocklistClient(endpoint=endpoint, credential=credential)
        self.threshold = threshold

        if score_type not in ["max", "avg"]:
            msg = "score_type must be either 'max' or 'avg'"
            raise ValueError(msg)
        self.score_type = score_type

        if blocklist_names:
            if not isinstance(blocklist_names, list):
                msg = "blocklist_names must be a list of strings"
                raise ValueError(msg)
            for name in blocklist_names:
                if not isinstance(name, str):
                    msg = "blocklist_names must be a list of strings"
                    raise ValueError(msg)
        self.blocklist_names = blocklist_names

    def validate(self, content: str) -> GuardrailOutput:
        """Validate content using Azure Content Safety.

        Args:
            content (str): The content to be evaluated.

        Returns:
            GuardrailOutput: The result of the guardrail evaluation.

        """
        model_inputs = self._pre_processing(content)
        model_outputs = self._inference(model_inputs)
        return self._post_processing(model_outputs)

    @error_message("Was unable to create or update blocklist.")  # type: ignore [untyped-decorator]
    def create_or_update_blocklist(self, blocklist_name: str, blocklist_description: str) -> None:
        """Create or update a blocklist in Azure Content Safety.

        Args:
            blocklist_name (str): The name of the blocklist.
            blocklist_description (str): The description of the blocklist.

        """
        self.blocklist_client.create_or_update_text_blocklist(
            blocklist_name=blocklist_name,
            options=TextBlocklist(blocklist_name=blocklist_name, description=blocklist_description),
        )

    @error_message("Was unable to add blocklist items.")  # type: ignore [untyped-decorator]
    def add_blocklist_items(self, blocklist_name: str, blocklist_terms: list[str]) -> None:
        """Add items to a blocklist.

        Args:
            blocklist_name (str): The name of the blocklist.
            blocklist_terms (List[str]): The terms to add to the blocklist.

        """
        blocklist_items = []
        for term in blocklist_terms:
            blocklist_items.append(TextBlocklistItem(text=term))

        self.blocklist_client.add_or_update_blocklist_items(
            blocklist_name=blocklist_name,
            options=AddOrUpdateTextBlocklistItemsOptions(blocklist_items=blocklist_items),
        )

    @error_message("Was unable to list blocklists.")  # type: ignore [untyped-decorator]
    def list_blocklists(self) -> list[dict[str, str | None]]:
        """List all blocklists in Azure Content Safety.

        Returns:
            List[Dict[str, str]]: A list of blocklist details.

        """
        blocklists = self.blocklist_client.list_text_blocklists()
        return [{"name": blocklist.blocklist_name, "description": blocklist.description} for blocklist in blocklists]

    @error_message("Was unable to list blocklist items.")  # type: ignore [untyped-decorator]
    def list_blocklist_items(self, blocklist_name: str) -> list[dict[str, str | None]]:
        """List items in a blocklist.

        Args:
            blocklist_name (str): The name of the blocklist.

        Returns:
            List[Dict[str, str]]: The list of blocklist items.

        """
        blocklist_items = self.blocklist_client.list_text_blocklist_items(blocklist_name=blocklist_name)
        return [
            {"id": item.blocklist_item_id, "text": item.text, "description": item.description}
            for item in blocklist_items
        ]

    @error_message("Was unable to get blocklist.")  # type: ignore [untyped-decorator]
    def get_blocklist(self, blocklist_name: str) -> dict[str, str | None]:
        """Get a blocklist by name.

        Args:
            blocklist_name (str): The name of the blocklist.

        Returns:
            dict[str, str]: The blocklist details.

        """
        blocklist = self.blocklist_client.get_text_blocklist(blocklist_name=blocklist_name)
        return {"name": blocklist.blocklist_name, "description": blocklist.description}

    @error_message("Was unable to get blocklist item.")  # type: ignore [untyped-decorator]
    def get_blocklist_item(self, blocklist_name: str, item_id: str) -> dict[str, str | None]:
        """Get a blocklist item by ID.

        Args:
            blocklist_name (str): The name of the blocklist.
            item_id (str): The ID of the blocklist item.

        Returns:
            dict[str, str]: The blocklist item details.

        """
        item = self.blocklist_client.get_text_blocklist_item(blocklist_name=blocklist_name, blocklist_item_id=item_id)
        return {"id": item.blocklist_item_id, "text": item.text, "description": item.description}

    @error_message("Was unable to delete blocklist.")  # type: ignore [untyped-decorator]
    def delete_blocklist(self, blocklist_name: str) -> None:
        """Delete a blocklist by name.

        Args:
            blocklist_name (str): The name of the blocklist.

        """
        self.blocklist_client.delete_text_blocklist(blocklist_name=blocklist_name)

    @error_message("Was unable to delete blocklist item.")  # type: ignore [untyped-decorator]
    def delete_blocklist_items(self, blocklist_name: str, item_ids: list[str]) -> None:
        """Delete a blocklist item by ID.

        Args:
            blocklist_name (str): The name of the blocklist.
            item_ids (List[str]): The IDs of the blocklist items.

        """
        self.blocklist_client.remove_blocklist_items(  # type: ignore [call-overload]
            blocklist_name=blocklist_name,
            blocklist_item_id=RemoveTextBlocklistItemsOptions(blocklist_item_ids=item_ids),
        )

    def _pre_processing(self, text: str) -> AnalyzeTextOptions | AnalyzeImageOptions:
        if self._is_existing_path(text):
            try:
                with open(text, "rb") as file:
                    return AnalyzeImageOptions(image=ImageData(content=file.read()))
            except ValueError as e:
                msg = "Must provide a file path to an image file."
                raise ValueError(msg) from e
        else:
            if self.blocklist_names:
                return AnalyzeTextOptions(text=text, blocklist_names=self.blocklist_names, halt_on_blocklist_hit=False)
            return AnalyzeTextOptions(text=text)

    @error_message("Was unable to analyze text or image.")  # type: ignore [untyped-decorator]
    def _inference(self, model_inputs: AnalyzeTextOptions | AnalyzeImageOptions) -> Any:
        if isinstance(model_inputs, AnalyzeTextOptions):
            response = self.client.analyze_text(model_inputs)
        else:
            response = self.client.analyze_image(model_inputs)  # type: ignore [assignment]
        return response

    def _post_processing(self, model_outputs: Any) -> GuardrailOutput:
        results_dict = {
            "hate": next(item for item in model_outputs.categories_analysis if item.category == TextCategory.HATE),
            "self_harm": next(
                item for item in model_outputs.categories_analysis if item.category == TextCategory.SELF_HARM
            ),
            "sexual": next(item for item in model_outputs.categories_analysis if item.category == TextCategory.SEXUAL),
            "violence": next(
                item for item in model_outputs.categories_analysis if item.category == TextCategory.VIOLENCE
            ),
        }

        explanation = {key: result.severity for key, result in results_dict.items() if result is not None}

        if self.score_type == "max":
            score = max(
                explanation_score for explanation_score in explanation.values() if explanation_score is not None
            )
        else:
            score = sum(
                explanation_score for explanation_score in explanation.values() if explanation_score is not None
            ) / sum(1 for explanation_score in explanation.values() if explanation_score is not None)

        explanation["blocklist"] = model_outputs.blocklists_match if self.blocklist_names else None

        valid = score < self.threshold
        if valid and explanation.get("blocklist"):
            valid = False

        return GuardrailOutput(valid=valid, explanation=explanation, score=score)

    def _is_existing_path(self, text: str) -> bool:
        return os.path.exists(text)
__init__(endpoint=None, api_key=None, threshold=2, score_type='max', blocklist_names=None)

Initialize Azure Content Safety client.

Parameters:

Name Type Description Default
endpoint str

The endpoint URL for the Azure Content Safety service.

 None
api_key str

The API key for authenticating with the service.

 None
threshold int

The threshold for determining if content is unsafe.

 2
score_type str

The type of score to use ("max" or "avg").

 'max'
blocklist_names List[str] | None

List of blocklist names to use for content evaluation.

 None
Source code in src/any_guardrail/guardrails/azure_content_safety/azure_content_safety.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def __init__(
    self,
    endpoint: str | None = None,
    api_key: str | None = None,
    threshold: int = 2,
    score_type: str = "max",
    blocklist_names: list[str] | None = None,
) -> None:
    """Initialize Azure Content Safety client.

    Args:
        endpoint (str): The endpoint URL for the Azure Content Safety service.
        api_key (str): The API key for authenticating with the service.
        threshold (int): The threshold for determining if content is unsafe.
        score_type (str): The type of score to use ("max" or "avg").
        blocklist_names (List[str] | None): List of blocklist names to use for content evaluation.

    """
    if api_key:
        credential = AzureKeyCredential(api_key)
    else:
        try:
            credential = AzureKeyCredential(os.environ["CONTENT_SAFETY_KEY"])
        except KeyError as e:
            msg = "CONTENT_SAFETY_KEY environment variable is not set. Either provide an api_key or set the environment variable."
            raise KeyError(msg) from e
    if not endpoint:
        try:
            endpoint = os.environ["CONTENT_SAFETY_ENDPOINT"]
        except KeyError as e:
            msg = "CONTENT_SAFETY_ENDPOINT environment variable is not set. Either provide an endpoint or set the environment variable."
            raise KeyError(msg) from e

    self.client = ContentSafetyClient(endpoint=endpoint, credential=credential)
    self.blocklist_client = BlocklistClient(endpoint=endpoint, credential=credential)
    self.threshold = threshold

    if score_type not in ["max", "avg"]:
        msg = "score_type must be either 'max' or 'avg'"
        raise ValueError(msg)
    self.score_type = score_type

    if blocklist_names:
        if not isinstance(blocklist_names, list):
            msg = "blocklist_names must be a list of strings"
            raise ValueError(msg)
        for name in blocklist_names:
            if not isinstance(name, str):
                msg = "blocklist_names must be a list of strings"
                raise ValueError(msg)
    self.blocklist_names = blocklist_names
add_blocklist_items(blocklist_name, blocklist_terms)

Add items to a blocklist.

Parameters:

Name Type Description Default
blocklist_name str

The name of the blocklist.

 required
blocklist_terms List[str]

The terms to add to the blocklist.

 required
Source code in src/any_guardrail/guardrails/azure_content_safety/azure_content_safety.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
@error_message("Was unable to add blocklist items.")  # type: ignore [untyped-decorator]
def add_blocklist_items(self, blocklist_name: str, blocklist_terms: list[str]) -> None:
    """Add items to a blocklist.

    Args:
        blocklist_name (str): The name of the blocklist.
        blocklist_terms (List[str]): The terms to add to the blocklist.

    """
    blocklist_items = []
    for term in blocklist_terms:
        blocklist_items.append(TextBlocklistItem(text=term))

    self.blocklist_client.add_or_update_blocklist_items(
        blocklist_name=blocklist_name,
        options=AddOrUpdateTextBlocklistItemsOptions(blocklist_items=blocklist_items),
    )
create_or_update_blocklist(blocklist_name, blocklist_description)

Create or update a blocklist in Azure Content Safety.

Parameters:

Name Type Description Default
blocklist_name str

The name of the blocklist.

 required
blocklist_description str

The description of the blocklist.

 required
Source code in src/any_guardrail/guardrails/azure_content_safety/azure_content_safety.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
@error_message("Was unable to create or update blocklist.")  # type: ignore [untyped-decorator]
def create_or_update_blocklist(self, blocklist_name: str, blocklist_description: str) -> None:
    """Create or update a blocklist in Azure Content Safety.

    Args:
        blocklist_name (str): The name of the blocklist.
        blocklist_description (str): The description of the blocklist.

    """
    self.blocklist_client.create_or_update_text_blocklist(
        blocklist_name=blocklist_name,
        options=TextBlocklist(blocklist_name=blocklist_name, description=blocklist_description),
    )
delete_blocklist(blocklist_name)

Delete a blocklist by name.

Parameters:

Name Type Description Default
blocklist_name str

The name of the blocklist.

 required
Source code in src/any_guardrail/guardrails/azure_content_safety/azure_content_safety.py 
201
202
203
204
205
206
207
208
209
@error_message("Was unable to delete blocklist.")  # type: ignore [untyped-decorator]
def delete_blocklist(self, blocklist_name: str) -> None:
    """Delete a blocklist by name.

    Args:
        blocklist_name (str): The name of the blocklist.

    """
    self.blocklist_client.delete_text_blocklist(blocklist_name=blocklist_name)
delete_blocklist_items(blocklist_name, item_ids)

Delete a blocklist item by ID.

Parameters:

Name Type Description Default
blocklist_name str

The name of the blocklist.

 required
item_ids List[str]

The IDs of the blocklist items.

 required
Source code in src/any_guardrail/guardrails/azure_content_safety/azure_content_safety.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
@error_message("Was unable to delete blocklist item.")  # type: ignore [untyped-decorator]
def delete_blocklist_items(self, blocklist_name: str, item_ids: list[str]) -> None:
    """Delete a blocklist item by ID.

    Args:
        blocklist_name (str): The name of the blocklist.
        item_ids (List[str]): The IDs of the blocklist items.

    """
    self.blocklist_client.remove_blocklist_items(  # type: ignore [call-overload]
        blocklist_name=blocklist_name,
        blocklist_item_id=RemoveTextBlocklistItemsOptions(blocklist_item_ids=item_ids),
    )
get_blocklist(blocklist_name)

Get a blocklist by name.

Parameters:

Name Type Description Default
blocklist_name str

The name of the blocklist.

 required

Returns:

Type Description
dict[str, str | None]

dict[str, str]: The blocklist details.
Source code in src/any_guardrail/guardrails/azure_content_safety/azure_content_safety.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
@error_message("Was unable to get blocklist.")  # type: ignore [untyped-decorator]
def get_blocklist(self, blocklist_name: str) -> dict[str, str | None]:
    """Get a blocklist by name.

    Args:
        blocklist_name (str): The name of the blocklist.

    Returns:
        dict[str, str]: The blocklist details.

    """
    blocklist = self.blocklist_client.get_text_blocklist(blocklist_name=blocklist_name)
    return {"name": blocklist.blocklist_name, "description": blocklist.description}
get_blocklist_item(blocklist_name, item_id)

Get a blocklist item by ID.

Parameters:

Name Type Description Default
blocklist_name str

The name of the blocklist.

 required
item_id str

The ID of the blocklist item.

 required

Returns:

Type Description
dict[str, str | None]

dict[str, str]: The blocklist item details.
Source code in src/any_guardrail/guardrails/azure_content_safety/azure_content_safety.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
@error_message("Was unable to get blocklist item.")  # type: ignore [untyped-decorator]
def get_blocklist_item(self, blocklist_name: str, item_id: str) -> dict[str, str | None]:
    """Get a blocklist item by ID.

    Args:
        blocklist_name (str): The name of the blocklist.
        item_id (str): The ID of the blocklist item.

    Returns:
        dict[str, str]: The blocklist item details.

    """
    item = self.blocklist_client.get_text_blocklist_item(blocklist_name=blocklist_name, blocklist_item_id=item_id)
    return {"id": item.blocklist_item_id, "text": item.text, "description": item.description}
list_blocklist_items(blocklist_name)

List items in a blocklist.

Parameters:

Name Type Description Default
blocklist_name str

The name of the blocklist.

 required

Returns:

Type Description
list[dict[str, str | None]]

List[Dict[str, str]]: The list of blocklist items.
Source code in src/any_guardrail/guardrails/azure_content_safety/azure_content_safety.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
@error_message("Was unable to list blocklist items.")  # type: ignore [untyped-decorator]
def list_blocklist_items(self, blocklist_name: str) -> list[dict[str, str | None]]:
    """List items in a blocklist.

    Args:
        blocklist_name (str): The name of the blocklist.

    Returns:
        List[Dict[str, str]]: The list of blocklist items.

    """
    blocklist_items = self.blocklist_client.list_text_blocklist_items(blocklist_name=blocklist_name)
    return [
        {"id": item.blocklist_item_id, "text": item.text, "description": item.description}
        for item in blocklist_items
    ]
list_blocklists()

List all blocklists in Azure Content Safety.

Returns:

Type Description
list[dict[str, str | None]]

List[Dict[str, str]]: A list of blocklist details.
Source code in src/any_guardrail/guardrails/azure_content_safety/azure_content_safety.py 
144
145
146
147
148
149
150
151
152
153
@error_message("Was unable to list blocklists.")  # type: ignore [untyped-decorator]
def list_blocklists(self) -> list[dict[str, str | None]]:
    """List all blocklists in Azure Content Safety.

    Returns:
        List[Dict[str, str]]: A list of blocklist details.

    """
    blocklists = self.blocklist_client.list_text_blocklists()
    return [{"name": blocklist.blocklist_name, "description": blocklist.description} for blocklist in blocklists]
validate(content)

Validate content using Azure Content Safety.

Parameters:

Name Type Description Default
content str

The content to be evaluated.

 required

Returns:

Name Type Description
GuardrailOutput GuardrailOutput

The result of the guardrail evaluation.
Source code in src/any_guardrail/guardrails/azure_content_safety/azure_content_safety.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def validate(self, content: str) -> GuardrailOutput:
    """Validate content using Azure Content Safety.

    Args:
        content (str): The content to be evaluated.

    Returns:
        GuardrailOutput: The result of the guardrail evaluation.

    """
    model_inputs = self._pre_processing(content)
    model_outputs = self._inference(model_inputs)
    return self._post_processing(model_outputs)

error_message(message)

Handle exceptions for Azure Content Safety operations.

Source code in src/any_guardrail/guardrails/azure_content_safety/azure_content_safety.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def error_message(message: str) -> Any:
    """Handle exceptions for Azure Content Safety operations."""

    def error_handler_decorator(func: Callable) -> Any:  # type: ignore [type-arg]
        """Handle exceptions for the wrapped function."""

        @functools.wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            try:
                return func(*args, **kwargs)
            except HttpResponseError as e:
                raise RuntimeError(message + f" Details: {e!s}") from e

        return wrapper

    return error_handler_decorator