Spaces:
Runtime error
Runtime error
File size: 5,151 Bytes
cd05673 | 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 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 | import json
from openai import OpenAI
from app.core.config import get_settings
from typing import Dict, Any, List, Tuple
from transformers import pipeline
class ModerationService:
def __init__(self):
settings = get_settings()
# Initialize Hugging Face moderation pipeline (or OpenAI moderation)
self.hf_moderator = pipeline(
"text-classification", model=settings.hf_moderation_model
)
# Groq client for dynamic rewrites
self.groq_client = OpenAI(
base_url="https://api.groq.com/openai/v1", api_key=settings.groq_api_key
)
self.hf_model_name = settings.hf_moderation_model
self.chat_model = settings.chat_model
self.threshold = settings.moderation_threshold
def get_moderation_scores(self, text: str) -> Dict[str, Any]:
try:
# 1. Moderate
hf_result = self.hf_moderator(text)[0]
label = hf_result["label"]
score = float(hf_result["score"])
is_flagged = (
label == "LABEL_1" or label.lower() == "toxic"
) and score >= self.threshold
top_category = "toxic" if is_flagged else "safe"
# 2. Generate dynamic meaning and alternatives
identified_meaning = ""
alternatives = []
if is_flagged:
identified_meaning, alternatives = self._analyze_and_reframe(
text, top_category
)
return {
"is_flagged": is_flagged,
"model_used": self.hf_model_name,
"top_category": top_category,
"identified_meaning": identified_meaning,
"suggested_alternatives": alternatives,
"raw_scores": {label: score},
}
except Exception as e:
raise RuntimeError(f"Moderation service failed: {str(e)}")
def _analyze_and_reframe(
self, original_text: str, category: str
) -> Tuple[str, List[str]]:
prompt = f"""
The following text was flagged for toxicity:
ORIGINAL TEXT: "{original_text}"
Your task:
1. Identify the core underlying meaning or emotion of this text. (e.g., "The user is frustrated because a deadline was missed.")
2. Reframe this exact meaning into exactly 8 safe, polite, and constructive alternatives.
You MUST respond with a JSON object containing exactly two keys: "identified_meaning" (a string) and "alternatives" (an array of 8 strings).
Example:
{{
"identified_meaning": "The user is frustrated that their colleague is being slow.",
"alternatives": [
"I feel like we are moving a bit slowly on this.",
"Could we pick up the pace a little?",
"I am concerned about the timeline.",
"I would appreciate it if we could speed this up.",
"I think we need to be more efficient.",
"Let's try to get this done faster.",
"I'm feeling a bit impatient with the progress.",
"Can we focus and get this finished?"
]
}}
"""
try:
response = self.groq_client.chat.completions.create(
model=self.chat_model,
messages=[
{
"role": "system",
"content": "You are a helpful assistant that analyzes toxic text and outputs valid JSON.",
},
{"role": "user", "content": prompt},
],
temperature=0.7,
response_format={"type": "json_object"},
)
content = response.choices[0].message.content
parsed_json = json.loads(content)
meaning = parsed_json.get(
"identified_meaning", "Could not identify meaning."
)
# Robustly find the list of alternatives
alts = []
for key, value in parsed_json.items():
if isinstance(value, list) and len(value) > 0:
alts = value
break
return meaning, alts
except Exception as e:
print(f"🚨 GROQ FAILED: {str(e)}")
return (
"Error analyzing meaning.",
[
"Please ensure your message is respectful.",
"I disagree with you, but let's keep it civil.",
"Can we discuss this without insults?",
"I'd prefer to focus on the facts.",
"Let's take a moment to cool down.",
"I think we should agree to disagree.",
"Please rephrase your message politely.",
"Let's try to find common ground.",
],
)
moderation_service = ModerationService()
|