| # import re | |
| # from typing import List, Optional, Dict, Any | |
| # from transformers import pipeline | |
| # import torch | |
| # from pydantic import PrivateAttr | |
| # from crewai.tools import BaseTool | |
| # # --- BaseTool for Response Validation --- | |
| # class ValidateResponseTool(BaseTool): | |
| # name: str = "validate_response" | |
| # description: str = "Validates safety and helpfulness of an AI response" | |
| # model_config = {"arbitrary_types_allowed": True} | |
| # _prohibited_patterns: dict = PrivateAttr() | |
| # _supportive_elements: dict = PrivateAttr() | |
| # _crisis_indicators: List[str] = PrivateAttr() | |
| # _negative_tone_words: List[str] = PrivateAttr() | |
| # _dismissive_phrases: List[str] = PrivateAttr() | |
| # _sentiment_analyzer: object = PrivateAttr() | |
| # def __init__(self, config=None, **data): | |
| # super().__init__(**data) | |
| # # === Paste your initialization data here as PrivateAttr === | |
| # self._sentiment_analyzer = pipeline( | |
| # "sentiment-analysis", | |
| # model="nlptown/bert-base-multilingual-uncased-sentiment", | |
| # device=0 if torch.cuda.is_available() else -1 | |
| # ) | |
| # self._prohibited_patterns = { | |
| # 'medical': [ | |
| # r'\b(?:diagnos|prescrib|medicat|cure|treat|therap)\w*\b', | |
| # r'\b(?:disease|illness|disorder|syndrome)\s+(?:is|are|can be)\b', | |
| # r'\b(?:take|consume|dose|dosage)\s+\d+\s*(?:mg|ml|pill|tablet)', | |
| # r'\b(?:medical|clinical|physician|doctor)\s+(?:advice|consultation|opinion)', | |
| # ], | |
| # 'legal': [ | |
| # r'\b(?:legal advice|lawsuit|sue|court|litigation)\b', | |
| # r'\b(?:illegal|unlawful|crime|criminal|prosecut)\w*\b', | |
| # r'\b(?:you should|must|have to)\s+(?:sign|agree|consent|contract)', | |
| # r'\b(?:rights|obligations|liability|damages)\s+(?:are|include)\b', | |
| # ], | |
| # 'financial': [ | |
| # r'\b(?:invest|buy|sell|trade)\s+(?:stock|crypto|bitcoin|forex)\b', | |
| # r'\b(?:guaranteed|promise)\s+(?:return|profit|income|earnings)\b', | |
| # r'\b(?:financial advisor|investment advice|trading strategy)\b', | |
| # r'\b(?:tax|accounting|financial planning)\s+(?:advice|consultation)', | |
| # ], | |
| # 'harmful': [ | |
| # r'\b(?:suicide|suicidal|kill\s+(?:your|my)self|end\s+(?:it|life))\b', | |
| # r'\b(?:self[\-\s]?harm|hurt\s+(?:your|my)self|cutting)\b', | |
| # r'\b(?:violence|violent|weapon|attack|assault)\b', | |
| # r'\b(?:hate|discriminat|racist|sexist|homophobic)\b', | |
| # ], | |
| # 'absolute': [ | |
| # r'\b(?:always|never|every|all|none|no one|everyone)\s+(?:will|must|should|is|are)\b', | |
| # r'\b(?:definitely|certainly|guaranteed|assured|promise)\b', | |
| # r'\b(?:only way|only solution|must do|have to)\b', | |
| # ] | |
| # } | |
| # self._supportive_elements = { | |
| # 'empathy': [ | |
| # 'understand', 'hear', 'feel', 'acknowledge', 'recognize', | |
| # 'appreciate', 'empathize', 'relate', 'comprehend' | |
| # ], | |
| # 'validation': [ | |
| # 'valid', 'normal', 'understandable', 'natural', 'okay', | |
| # 'reasonable', 'makes sense', 'legitimate' | |
| # ], | |
| # 'support': [ | |
| # 'support', 'help', 'here for you', 'together', 'alongside', | |
| # 'assist', 'guide', 'accompany', 'with you' | |
| # ], | |
| # 'hope': [ | |
| # 'can', 'possible', 'able', 'capable', 'potential', | |
| # 'opportunity', 'growth', 'improve', 'better', 'progress' | |
| # ], | |
| # 'empowerment': [ | |
| # 'choice', 'decide', 'control', 'power', 'strength', | |
| # 'agency', 'capable', 'resource', 'ability' | |
| # ] | |
| # } | |
| # self._crisis_indicators = [ | |
| # r'\b(?:want|going|plan)\s+to\s+(?:die|kill|end)\b', | |
| # r'\b(?:no reason|point|hope)\s+(?:to|in)\s+(?:live|living|life)\b', | |
| # r'\b(?:better off|world)\s+without\s+me\b', | |
| # r'\bsuicide\s+(?:plan|method|attempt)\b', | |
| # r'\b(?:final|last)\s+(?:goodbye|letter|message)\b' | |
| # ] | |
| # self._negative_tone_words = [ | |
| # 'stupid', 'idiot', 'dumb', 'pathetic', 'worthless', | |
| # 'loser', 'failure', 'weak', 'incompetent', 'useless' | |
| # ] | |
| # self._dismissive_phrases = [ | |
| # 'just get over it', 'stop complaining', 'not a big deal', | |
| # 'being dramatic', 'overreacting', 'too sensitive' | |
| # ] | |
| # def _run(self, response: str, context: Optional[dict] = None): | |
| # """ | |
| # Pydantic and CrewAI-compatible single-tool version. | |
| # Returns a dictionary directly. | |
| # """ | |
| # # Issues, warnings, suggestions collections | |
| # issues = [] | |
| # warnings = [] | |
| # suggestions = [] | |
| # # --- Prohibited Content --- | |
| # for category, patterns in self._prohibited_patterns.items(): | |
| # for pattern in patterns: | |
| # if re.search(pattern, response, re.IGNORECASE): | |
| # issues.append(f"Contains {category} advice/content") | |
| # if category == "medical": | |
| # suggestions.append("Replace with: 'Consider speaking with a healthcare professional'") | |
| # elif category == "legal": | |
| # suggestions.append("For legal matters, consult with a qualified attorney") | |
| # elif category == "financial": | |
| # suggestions.append("For financial decisions, consider consulting a financial advisor") | |
| # elif category == "harmful": | |
| # suggestions.append("Include crisis resources and express immediate concern for safety") | |
| # elif category == "absolute": | |
| # suggestions.append("Use qualifying language like 'often', 'might', 'could' instead of absolutes") | |
| # break | |
| # # --- Sentiment/Tone --- | |
| # try: | |
| # sentiment_result = self._sentiment_analyzer(response[:512])[0] | |
| # sentiment_label = sentiment_result['label'] | |
| # if '1' in sentiment_label or '2' in sentiment_label: | |
| # warnings.append("Response tone is too negative") | |
| # suggestions.append("Add more supportive and hopeful language") | |
| # except Exception: | |
| # pass | |
| # # --- Negative words --- | |
| # found_negative = [word for word in self._negative_tone_words if word in response.lower()] | |
| # if found_negative: | |
| # warnings.append(f"Contains negative/judgmental language: {', '.join(found_negative)}") | |
| # suggestions.append("Replace judgmental terms with supportive language") | |
| # # --- Dismissive --- | |
| # found_dismissive = [phrase for phrase in self._dismissive_phrases if phrase in response.lower()] | |
| # if found_dismissive: | |
| # warnings.append("Contains dismissive language") | |
| # suggestions.append("Acknowledge and validate the person's feelings instead") | |
| # # --- Supportive Elements --- | |
| # text_lower = response.lower() | |
| # missing_elements = [] | |
| # for element, keywords in self._supportive_elements.items(): | |
| # if not any(keyword in text_lower for keyword in keywords): | |
| # missing_elements.append(element) | |
| # if missing_elements: | |
| # warnings.append(f"Missing supportive elements: {', '.join(missing_elements)}") | |
| # for miss in missing_elements: | |
| # if miss == 'empathy': | |
| # suggestions.append("Add empathetic language like 'I understand how difficult this must be'") | |
| # elif miss == 'validation': | |
| # suggestions.append("Validate their feelings with phrases like 'Your feelings are completely valid'") | |
| # elif miss == 'support': | |
| # suggestions.append("Express support with 'I'm here to support you through this'") | |
| # elif miss == 'hope': | |
| # suggestions.append("Include hopeful elements about growth and positive change") | |
| # elif miss == 'empowerment': | |
| # suggestions.append("Emphasize their agency and ability to make choices") | |
| # # --- Crisis detection from context --- | |
| # if context and context.get("user_input"): | |
| # for pattern in self._crisis_indicators: | |
| # if re.search(pattern, context["user_input"], re.IGNORECASE): | |
| # if "crisis" not in response.lower(): | |
| # warnings.append("User may be in crisis but response doesn't address this") | |
| # suggestions.append("Include crisis resources and immediate support options") | |
| # # --- Confidence --- | |
| # confidence = 1.0 | |
| # if issues: | |
| # confidence = 0.3 - (0.1 * len(issues)) | |
| # confidence = max(0.0, confidence - 0.1 * len(warnings)) | |
| # class ValidationTools: | |
| # def __init__(self, config=None): | |
| # self.validate_response_tool = ValidateResponseTool(config) | |
| import re | |
| from typing import List, Optional, Dict, Any | |
| from transformers import pipeline | |
| import torch | |
| from pydantic import PrivateAttr | |
| from crewai.tools import BaseTool | |
| class ValidateResponseTool(BaseTool): | |
| name: str = "validate_response" | |
| description: str = "Validates safety and helpfulness of an AI response" | |
| model_config = {"arbitrary_types_allowed": True} | |
| _prohibited_patterns: dict = PrivateAttr() | |
| _supportive_elements: dict = PrivateAttr() | |
| _crisis_indicators: List[str] = PrivateAttr() | |
| _negative_tone_words: List[str] = PrivateAttr() | |
| _dismissive_phrases: List[str] = PrivateAttr() | |
| _sentiment_analyzer: object = PrivateAttr() | |
| def __init__(self, config=None, **data): | |
| super().__init__(**data) | |
| self._sentiment_analyzer = pipeline( | |
| "sentiment-analysis", | |
| model="nlptown/bert-base-multilingual-uncased-sentiment", | |
| device=0 if torch.cuda.is_available() else -1 | |
| ) | |
| # ... (rest of pattern/class setup exactly as you wrote) ... | |
| self._prohibited_patterns = {...} | |
| self._supportive_elements = {...} | |
| self._crisis_indicators = [...] | |
| self._negative_tone_words = [...] | |
| self._dismissive_phrases = [...] | |
| def _run(self, response: str, context: Optional[dict] = None): | |
| # ... All your logic from the previous snippet. ... | |
| issues = [] | |
| warnings = [] | |
| suggestions = [] | |
| # ... checks (same as previous) ... | |
| return { | |
| "is_valid": len(issues) == 0, | |
| "issues": issues, | |
| "warnings": warnings, | |
| "suggestions": suggestions, | |
| } | |
| class ValidationTools: | |
| def __init__(self, config=None): | |
| self.validate_response_tool = ValidateResponseTool(config) |