-
Notifications
You must be signed in to change notification settings - Fork 304
/
ContentSafetyChecker.py
69 lines (58 loc) · 3 KB
/
ContentSafetyChecker.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
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
import logging
from azure.ai.contentsafety import ContentSafetyClient
from azure.core.credentials import AzureKeyCredential
from azure.identity import DefaultAzureCredential
from azure.core.exceptions import HttpResponseError
from azure.ai.contentsafety.models import AnalyzeTextOptions
from ..helpers.EnvHelper import EnvHelper
from .AnswerProcessingBase import AnswerProcessingBase
from ..common.Answer import Answer
logger = logging.getLogger(__name__)
class ContentSafetyChecker(AnswerProcessingBase):
def __init__(self):
env_helper = EnvHelper()
if env_helper.AZURE_AUTH_TYPE == "rbac":
self.content_safety_client = ContentSafetyClient(
env_helper.AZURE_CONTENT_SAFETY_ENDPOINT,
DefaultAzureCredential(),
)
else:
self.content_safety_client = ContentSafetyClient(
env_helper.AZURE_CONTENT_SAFETY_ENDPOINT,
AzureKeyCredential(env_helper.AZURE_CONTENT_SAFETY_KEY),
)
def process_answer(self, answer: Answer, **kwargs: dict) -> Answer:
response_template = kwargs["response_template"]
answer.answer = self._filter_text_and_replace(answer.answer, response_template)
return answer
def validate_input_and_replace_if_harmful(self, text):
response_template = f'{"Unfortunately, I am not able to process your question, as I have detected sensitive content that I am not allowed to process. This might be a mistake, so please try rephrasing your question."}'
return self.process_answer(
Answer(question="", answer=text, source_documents=[]),
response_template=response_template,
).answer
def validate_output_and_replace_if_harmful(self, text):
response_template = f'{"Unfortunately, I have detected sensitive content in my answer, which I am not allowed to show you. This might be a mistake, so please try again and maybe rephrase your question."}'
return self.process_answer(
Answer(question="", answer=text, source_documents=[]),
response_template=response_template,
).answer
def _filter_text_and_replace(self, text, response_template):
request = AnalyzeTextOptions(text=text)
try:
response = self.content_safety_client.analyze_text(request)
except HttpResponseError as e:
if e.error:
logger.error(
f"Analyze text failed. Error code: {e.error.code}. Error message: {e.error.message}."
)
raise
logger.exception("Analyze text failed.")
raise
filtered_text = text
# if response.hate_result.severity > 0 or response.self_harm_result.severity > 0 or response.sexual_result.severity > 0 or response.violence_result.severity > 0:
# filtered_text = response_template
for result in response.categories_analysis:
if result.severity > 0:
filtered_text = response_template
return filtered_text