Spaces:
Runtime error
Runtime error
| 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() | |