Spaces:
Sleeping
Sleeping
| import re | |
| import logging | |
| from typing import Tuple, List, Dict | |
| logger = logging.getLogger(__name__) | |
| # Policy rule definitions | |
| POLICY_RULES = { | |
| "return_window": { | |
| "correct_value": "30", | |
| "wrong_values": ["60", "45", "90", "14", "7", "15"], | |
| "keywords": ["return", "returns", "return policy", "return period"], | |
| "unit": "days" | |
| }, | |
| "refund_time": { | |
| "correct_value_min": "5", | |
| "correct_value_max": "7", | |
| "wrong_values": ["1", "2", "3", "10", "14", "20", "24", "instantly", "immediately", "same day"], | |
| "keywords": ["refund", "refunds", "money back", "refund time", "refund duration"], | |
| "unit": "business days" | |
| }, | |
| "standard_shipping": { | |
| "correct_value_min": "5", | |
| "correct_value_max": "7", | |
| "wrong_values": ["1", "2", "3", "10", "14"], | |
| "keywords": ["standard shipping", "regular shipping"], | |
| "unit": "business days" | |
| }, | |
| "express_shipping": { | |
| "correct_value_min": "2", | |
| "correct_value_max": "3", | |
| "keywords": ["express shipping", "express delivery"], | |
| "unit": "business days" | |
| }, | |
| "cancellation_window": { | |
| "correct_value": "1", | |
| "wrong_values": ["24", "48", "2", "3"], | |
| "keywords": ["cancel", "cancellation", "cancel order"], | |
| "unit": "hour" | |
| }, | |
| "cancellation_refund": { | |
| "correct_value_min": "3", | |
| "correct_value_max": "5", | |
| "keywords": ["cancelled order refund", "cancel refund"], | |
| "unit": "business days" | |
| }, | |
| "electronics_warranty": { | |
| "correct_value": "1", | |
| "wrong_values": ["2", "3", "lifetime", "90", "6"], | |
| "keywords": ["electronics warranty", "phone warranty", "laptop warranty"], | |
| "unit": "year" | |
| }, | |
| "clothing_warranty": { | |
| "correct_value": "90", | |
| "wrong_values": ["30", "180", "365", "1 year"], | |
| "keywords": ["clothing warranty", "apparel warranty"], | |
| "unit": "day" | |
| }, | |
| "damaged_report_window": { | |
| "correct_value": "48", | |
| "wrong_values": ["24", "72", "7", "30"], | |
| "keywords": ["damaged", "broken", "report damage"], | |
| "unit": "hours" | |
| }, | |
| "price_match_window": { | |
| "correct_value": "7", | |
| "wrong_values": ["14", "30", "3"], | |
| "keywords": ["price match", "price matching"], | |
| "unit": "days" | |
| }, | |
| "free_shipping_threshold": { | |
| "correct_value": "50", | |
| "wrong_values": ["25", "75", "100", "30"], | |
| "keywords": ["free shipping"], | |
| "unit": "dollars" | |
| }, | |
| } | |
| NEGATIVE_SIGNALS = ["not", "no", "cannot", "can't", "not eligible", "not accepted", | |
| "not possible", "unable", "not allowed", "ineligible", "prohibited", | |
| "unfortunately", "not qualify", "does not qualify"] | |
| POSITIVE_SIGNALS = ["can", "yes", "allowed", "eligible", "possible", "accepted", | |
| "qualify", "certainly", "absolutely", "of course"] | |
| def extract_numbers(text: str) -> List[str]: | |
| return re.findall(r'\b\d+(?:\.\d+)?\b', text) | |
| def check_return_policy(response: str) -> Tuple[str, str]: | |
| r = response.lower() | |
| details = [] | |
| if any(k in r for k in ["opened", "open product", "open item"]): | |
| if any(sig in r for sig in NEGATIVE_SIGNALS): | |
| details.append("Correctly states opened items cannot be returned") | |
| return "PASS", "; ".join(details) | |
| else: | |
| details.append("FAIL: Should state opened items cannot be returned") | |
| return "FAIL", "; ".join(details) | |
| if "30" in r and any(k in r for k in ["days", "day"]): | |
| details.append("Correctly mentions 30-day return window") | |
| return "PASS", "; ".join(details) | |
| if any(wrong in r for wrong in ["60 day", "45 day", "90 day", "60-day", "45-day", "90-day"]): | |
| details.append("FAIL: Incorrect return window stated") | |
| return "FAIL", "; ".join(details) | |
| return "PASS", "Return policy check passed" | |
| def check_refund_policy(response: str) -> Tuple[str, str]: | |
| r = response.lower() | |
| numbers = extract_numbers(r) | |
| for num in numbers: | |
| n = float(num) | |
| if n in [1, 2, 3, 10, 14, 20, 24, 48]: | |
| if "refund" in r and ("day" in r or "hour" in r or "business" in r): | |
| return "FAIL", f"FAIL: Incorrect refund timeframe mentioned: {num}" | |
| if "5" in numbers or "7" in numbers: | |
| if "refund" in r and "business" in r: | |
| return "PASS", "Correctly states 5-7 business day refund time" | |
| if any(word in r for word in ["immediately", "instant", "same day", "right away", "within 24"]): | |
| if "refund" in r: | |
| return "FAIL", "FAIL: Refund is NOT immediate; takes 5-7 business days" | |
| return "PASS", "Refund policy check passed" | |
| def check_shipping_policy(response: str) -> Tuple[str, str]: | |
| r = response.lower() | |
| if "standard" in r: | |
| numbers = extract_numbers(r) | |
| for num in numbers: | |
| n = float(num) | |
| if n in [1, 2, 3, 4, 10, 14]: | |
| if "day" in r: | |
| return "FAIL", f"FAIL: Standard shipping is 5-7 days, not {num} days" | |
| if "5" in numbers and "7" in numbers: | |
| return "PASS", "Correctly states standard shipping 5-7 days" | |
| if "express" in r: | |
| numbers = extract_numbers(r) | |
| for num in numbers: | |
| n = float(num) | |
| if n in [1, 5, 7, 10]: | |
| if "day" in r: | |
| return "FAIL", f"FAIL: Express shipping is 2-3 days, not {num} days" | |
| if "free" in r and "shipping" in r: | |
| if "50" not in r and "$50" not in r: | |
| if any(wrong in r for wrong in ["$25", "$75", "$100", "$30", "25 dollar", "75 dollar"]): | |
| return "FAIL", "FAIL: Free shipping threshold is $50, not the stated amount" | |
| return "PASS", "Shipping policy check passed" | |
| def check_cancellation_policy(response: str) -> Tuple[str, str]: | |
| r = response.lower() | |
| if "shipped" in r or "processing" in r: | |
| if any(sig in r for sig in NEGATIVE_SIGNALS): | |
| return "PASS", "Correctly states shipped orders cannot be cancelled" | |
| numbers = extract_numbers(r) | |
| for num in numbers: | |
| n = float(num) | |
| if "cancel" in r and n in [24, 48, 72]: | |
| if "hour" in r: | |
| return "FAIL", f"FAIL: Cancellation window is 1 hour, not {num} hours" | |
| if "1 hour" in r or "one hour" in r: | |
| return "PASS", "Correctly states 1-hour cancellation window" | |
| return "PASS", "Cancellation policy check passed" | |
| def check_warranty_policy(response: str) -> Tuple[str, str]: | |
| r = response.lower() | |
| if "electronic" in r or "phone" in r or "laptop" in r or "device" in r: | |
| if "lifetime" in r: | |
| return "FAIL", "FAIL: Electronics warranty is 1 year, NOT lifetime" | |
| if "1 year" in r or "one year" in r or "1-year" in r: | |
| return "PASS", "Correctly states 1-year electronics warranty" | |
| if "cloth" in r or "apparel" in r or "fabric" in r: | |
| if "90" in r and "day" in r: | |
| return "PASS", "Correctly states 90-day clothing warranty" | |
| if "1 year" in r and "cloth" in r: | |
| return "FAIL", "FAIL: Clothing warranty is 90 days, not 1 year" | |
| if "water damage" in r or "accidental" in r: | |
| if any(sig in r for sig in NEGATIVE_SIGNALS): | |
| return "PASS", "Correctly states water/accidental damage is not covered" | |
| else: | |
| return "FAIL", "FAIL: Water/accidental damage is NOT covered by warranty" | |
| return "PASS", "Warranty policy check passed" | |
| def check_damaged_product_policy(response: str) -> Tuple[str, str]: | |
| r = response.lower() | |
| if "damaged" in r or "broken" in r: | |
| numbers = extract_numbers(r) | |
| for num in numbers: | |
| n = float(num) | |
| if "hour" in r and n != 48 and n in [24, 72, 96]: | |
| return "FAIL", f"FAIL: Damaged products must be reported within 48 hours, not {num}" | |
| if "48" in numbers: | |
| return "PASS", "Correctly states 48-hour damage reporting window" | |
| return "PASS", "Damaged product policy check passed" | |
| def check_hallucination_signals(response: str) -> Tuple[bool, List[str]]: | |
| r = response.lower() | |
| hallucinations = [] | |
| # Check for invented policies | |
| if "free return" in r and "change of mind" not in r and "defective" not in r: | |
| if "always" in r or "all returns" in r: | |
| hallucinations.append("Claimed all returns have free shipping (incorrect)") | |
| if "lifetime warranty" in r or "lifetime guarantee" in r: | |
| hallucinations.append("Claimed lifetime warranty (policy says 1-year max)") | |
| if "24/7" in r and "support" in r: | |
| if "we offer" in r or "available" in r: | |
| pass # Plausible but not in policy - flag as unverified | |
| if "30%" in r or "50%" in r: | |
| if "discount" in r or "off" in r: | |
| hallucinations.append("Mentioned discount percentages not in policy") | |
| if "7 days" in r and "refund" in r and "business" not in r: | |
| hallucinations.append("Mentioned 7-day refund without specifying 'business days'") | |
| if "price match" in r: | |
| if "14 day" in r or "30 day" in r: | |
| hallucinations.append("Incorrect price match window - should be 7 days") | |
| return len(hallucinations) > 0, hallucinations | |
| def code_eval(query: str, response: str, category: str = None) -> Tuple[str, str, List[str]]: | |
| checks_run = [] | |
| all_details = [] | |
| failures = [] | |
| q_lower = query.lower() | |
| r_lower = response.lower() | |
| # Determine which checks to run based on query + response content | |
| run_return = any(k in q_lower for k in ["return", "send back", "returns"]) | |
| run_refund = any(k in q_lower for k in ["refund", "money back", "refunded"]) | |
| run_shipping = any(k in q_lower for k in ["shipping", "delivery", "ship", "deliver"]) | |
| run_cancel = any(k in q_lower for k in ["cancel", "cancellation"]) | |
| run_warranty = any(k in q_lower for k in ["warranty", "guarantee", "defective"]) | |
| run_damaged = any(k in q_lower for k in ["damaged", "broken", "arrived broken"]) | |
| if run_return: | |
| result, detail = check_return_policy(response) | |
| checks_run.append("return_policy") | |
| all_details.append(f"Return check: {detail}") | |
| if result == "FAIL": | |
| failures.append(detail) | |
| if run_refund: | |
| result, detail = check_refund_policy(response) | |
| checks_run.append("refund_policy") | |
| all_details.append(f"Refund check: {detail}") | |
| if result == "FAIL": | |
| failures.append(detail) | |
| if run_shipping: | |
| result, detail = check_shipping_policy(response) | |
| checks_run.append("shipping_policy") | |
| all_details.append(f"Shipping check: {detail}") | |
| if result == "FAIL": | |
| failures.append(detail) | |
| if run_cancel: | |
| result, detail = check_cancellation_policy(response) | |
| checks_run.append("cancellation_policy") | |
| all_details.append(f"Cancellation check: {detail}") | |
| if result == "FAIL": | |
| failures.append(detail) | |
| if run_warranty: | |
| result, detail = check_warranty_policy(response) | |
| checks_run.append("warranty_policy") | |
| all_details.append(f"Warranty check: {detail}") | |
| if result == "FAIL": | |
| failures.append(detail) | |
| if run_damaged: | |
| result, detail = check_damaged_product_policy(response) | |
| checks_run.append("damaged_policy") | |
| all_details.append(f"Damaged product check: {detail}") | |
| if result == "FAIL": | |
| failures.append(detail) | |
| # Always run hallucination signal check | |
| hallucinated, hall_list = check_hallucination_signals(response) | |
| if hallucinated: | |
| for h in hall_list: | |
| failures.append(f"Hallucination signal: {h}") | |
| all_details.extend([f"Hallucination: {h}" for h in hall_list]) | |
| if not checks_run: | |
| return "PASS", "No specific policy checks applicable", [] | |
| final = "FAIL" if failures else "PASS" | |
| return final, " | ".join(all_details), failures | |