# domain/math_validator.py — V1.1 (STABLE REGEX POLYGRAPH) import re import logging import re import logging import time import asyncio from typing import Tuple, List, Optional, Any import sympy from sympy.parsing.sympy_parser import parse_expr from utils.math_utils import aggressive_sympy_sanitizer logger = logging.getLogger(__name__) # Expressions that are structurally unparseable but pedagogically harmless _PIPE_SEPARATED_RESULT = re.compile(r'\|') _HEBREW_ONLY = re.compile(r'^[\u0590-\u05FF\s]+$') # LaTeX commands that SymPy cannot parse — strip layout but KEEP math functions _LATEX_STRIP = re.compile( r'\\(?:left|right|div|pm|mp|leq|geq|neq' r'|approx|infty|text|mathrm|mathbf|boxed|underbrace|overbrace|hat|bar|vec|dot|overline|underline)\b' ) # V307.0: Functions that often lack parentheses in LLM output (e.g. lnx) _MATH_FUNC_PARENS = ['ln', 'sin', 'cos', 'tan', 'sqrt', 'log', 'exp', 'Abs'] def _latex_to_sympy_str(latex_str: str) -> str: """ Best-effort LaTeX → SymPy-parseable string. V310.0: Aggressive Hebrew stripping and malformed notation cleanup. """ if latex_str is None: return "" s = str(latex_str).strip() # 0. V310.0: Strip Hebrew characters and BOM/Zero-width chars immediately s = re.sub(r'[\u0590-\u05FF\u200B-\u200D\uFEFF]', ' ', s) # 1. Handle \frac{a}{b} → (a)/(b) loop_counter = 0 max_loops = 15 while r'\frac' in s and loop_counter < max_loops: old_s = s s = re.sub(r'\\frac\s*\{([^{}]*)\}\s*\{([^{}]*)\}', r'(\1)/(\2)', s) if old_s == s: s = s.replace(r'\frac', '(frac_err)') break loop_counter += 1 # 2. Convert LaTeX functions to plain words (e.g. \ln -> ln) s = s.replace(r'\ln', ' ln ').replace(r'\sin', ' sin ').replace(r'\cos', ' cos ') s = s.replace(r'\tan', ' tan ').replace(r'\sqrt', ' sqrt ').replace(r'\log', ' log ') s = s.replace(r'\exp', ' exp ').replace(r'\pi', ' pi ').replace(r'\theta', ' theta ') # --- 🚀 BEGIN HOTFIX V9.0.3: LATEX MULTIPLICATION --- s = s.replace(r'\cdot', '*').replace(r'\times', '*') # --- END HOTFIX --- # 3. Remove remaining purely structural LaTeX commands s = _LATEX_STRIP.sub(' ', s) # 4. Remove LaTeX delimiters/wrappers s = s.replace('{', '(').replace('}', ')').replace('$', '') s = s.replace(r'\left', '').replace(r'\right', '') # 5. V307.0: Fix implicit function arguments (e.g. lnx -> ln(x)) for func in _MATH_FUNC_PARENS: pattern = r'\b' + func + r'\b\s*([^( \t\n\r\f\v,]+)' s = re.sub(pattern, func + r'(\1)', s) # 6. Handle absolute value pipes |x| -> Abs(x) loop_counter = 0 while '|' in s and s.count('|') >= 2 and loop_counter < 10: s = re.sub(r'\|([^|]+)\|', r'Abs(\1)', s) loop_counter += 1 # 7. Implicit multiplication: 2x → 2*x (only if not inside a word) s = re.sub(r'(\d)([a-zA-Z(])', r'\1*\2', s) # 8. V280.0: Equals sign handling is now moved to _check_segment # for more robust parsing of equations. # 9. Final cleanup: Remove illegal SymPy chars like ', ", ?, ! s = re.sub(r'[?!\'"]', '', s) s = re.sub(r'\s+', ' ', s) return s.strip() def _is_plaintext(expr_str: str) -> bool: if _HEBREW_ONLY.match(expr_str): return True if _PIPE_SEPARATED_RESULT.search(expr_str) and not any(c in expr_str for c in ['+', '-', '*', '/', '^', '=']): return True return False class MathPolygraph: TIMEOUT_SECONDS = 3 @staticmethod async def _validate_single(text: str, step_id) -> Tuple[bool, str]: """ V280.0 REDESIGN: 1. No Blind Stripping: Extracts $...$ or $$...$$ using re.finditer with DOTALL. 2. Security: Uses parse_expr(evaluate=False). 3. Equations: Splits by '=' and validates parts to bypass SymPy's '=' limitation. 4. Multi-Equal: Handles x=y=5 without crashing. 5. Empty Guard: Skips $$$$. """ if not text or not text.strip(): return True, "" # regex: find both $$display$$ and $inline$ blocks. DOTALL allows multi-line display math. # Group 1 = display math, Group 2 = inline math math_pattern = re.compile(r'\$\$(.*?)\$\$|\$(.*?)\$', re.DOTALL) matches = list(re.finditer(math_pattern, text)) if not matches: # V280.0 Rule: If no delimiters are found, treat the whole string as plain text # or try to parse if it looks like math (existing behavior for backward compatibility) if _is_plaintext(text): return True, "" return await MathPolygraph._check_segment(text, step_id) for match in matches: # Group 1 (Display) or Group 2 (Inline) content = (match.group(1) or match.group(2) or "").strip() # 5. Empty String Guard if not content: continue # V280.0 Fix: Multi-line display math might contain multiple equations. # Split by newline before validating segments. sub_segments = [s.strip() for s in content.split('\n') if s.strip()] for sub in sub_segments: ok, reason = await MathPolygraph._check_segment(sub, step_id) if not ok: return False, reason return True, "" @staticmethod async def _run_safe_math_op(func, *args, timeout_sec: float = 0.3) -> Any: """ V9.0.5: Executes a CPU-bound SymPy operation in a separate thread with a strict timeout. Returns the result of the function, or raises TimeoutError/Exception. """ try: return await asyncio.wait_for( asyncio.to_thread(func, *args), timeout=timeout_sec ) except asyncio.TimeoutError: raise asyncio.TimeoutError("SymPy operation timed out") @staticmethod async def _check_segment(raw_segment: str, step_id) -> Tuple[bool, str]: """Internal helper to validate a single extracted math segment.""" # V9.0.1: Use Aggressive Sanitizer to handle Hebrew, Arrows, and Comma-splitting sanitized_parts = aggressive_sympy_sanitizer(raw_segment) if not sanitized_parts: return True, "" for part in sanitized_parts: sympy_str = _latex_to_sympy_str(part) if not sympy_str or sympy_str in ('', '-', '()', '( )'): continue try: # V9.0.5: Wrap CPU-bound sympify/parsing in a non-blocking thread with 300ms timeout # We bypass the slow multiprocessing approach for SSE safety. def parse_and_eval(s): # Character Whitelist (from legacy _sympify_worker) safe_pattern = r'^[a-zA-Z0-9\s\+\-\*\/\^\(\)\.\,\!\%\=]+$' if not re.match(safe_pattern, s): return False res = parse_expr(s, evaluate=False) if res is not None: evaluated = res.doit() if hasattr(evaluated, 'is_finite') and evaluated.is_finite is False: return False if hasattr(evaluated, 'is_nan') and evaluated.is_nan: return False return True status = await MathPolygraph._run_safe_math_op(parse_and_eval, sympy_str) if status is False: logger.warning(f"🛡️ [SOFT FAIL] SymPy Parse Error on part '{part}'. Bypassing validator.") return True, "" # Soft Fail: allow stream to continue except asyncio.TimeoutError: logger.warning(f"🛡️ [SOFT FAIL] SymPy TIMEOUT (300ms) on part '{part}'. Bypassing validator.") return True, "" # Soft Fail: allow stream to continue except Exception as e: logger.warning(f"🛡️ [SOFT FAIL] Unexpected validation crash: {e} for part '{part}'. Bypassing.") return True, "" # Soft Fail: allow stream to continue return True, "" @staticmethod async def validate_step_sequence(steps: List[dict], topic: str = "GENERAL") -> Tuple[bool, str]: if not steps: return True, "" # V8.9.4: Skip deep SymPy parsing for discrete sequence steps to avoid false-positive SyntaxErrors is_sequence = topic and "SEQUENCE" in topic.upper() for step in steps: step_id = step.get('step_id', step.get('step_number', '?')) math_fields = [] for field in ('math_latex', 'block_math', 'math'): val = step.get(field) if val and isinstance(val, str) and val.strip(): math_fields.append(val.strip()) if not math_fields: continue # If sequence, we only check if it's "valid-ish" LaTeX vs deep SymPy check if is_sequence: # Basic sanity check for LaTeX balance if math_fields[0].count('{') != math_fields[0].count('}'): return False, f"LATEX_BRACKET_MISMATCH:step_{step_id}" continue ok, reason = await MathPolygraph._validate_single(math_fields[0], step_id) if not ok: return False, reason return True, "" @staticmethod async def are_equivalent(latex1: str, latex2: str) -> bool: """ V9.0.5: Checks if two LaTeX expressions are mathematically equivalent (Non-Blocking). Supports expressions and equations (by converting to 'expr = 0'). """ try: # 1. Handle Equations in Equivalence Check if '=' in latex1 and '=' in latex2 and latex1.count('=') == 1 and latex2.count('=') == 1: parts1 = [p.strip() for p in latex1.split('=') if p.strip()] parts2 = [p.strip() for p in latex2.split('=') if p.strip()] if len(parts1) == 2 and len(parts2) == 2: res1 = await MathPolygraph.are_equivalent(parts1[0], parts2[0]) res2 = await MathPolygraph.are_equivalent(parts1[1], parts2[1]) return res1 and res2 s1_raw = _latex_to_sympy_str(latex1) s2_raw = _latex_to_sympy_str(latex2) # Check for inequalities in raw LaTeX to be safe inequalities = ['<', '>', r'\leq', r'\geq', r'\neq', r'\leq', r'\geq'] if any(iq in latex1 for iq in inequalities) or any(iq in latex2 for iq in inequalities): return latex1.strip() == latex2.strip() # Security: Strict Whitelist for Equivalence Check safe_pattern = r'^[a-zA-Z0-9\s\+\-\*\/\^\(\)\.\,\!\=]+$' def is_safe(s): clean = s.replace('\\', '').replace('_', '').replace('{', '(').replace('}', ')') return bool(re.match(safe_pattern, clean)) if not (is_safe(s1_raw) and is_safe(s2_raw)): return latex1.strip() == latex2.strip() # V9.0.5: Wrap CPU-bound SymPy simplify in a thread with timeout def calc_equivalence(str1, str2): expr1 = parse_expr(str1, evaluate=False) expr2 = parse_expr(str2, evaluate=False) # "Variable Trap": Basic structural equivalence if variables are involved if len(expr1.free_symbols) > 0 or len(expr2.free_symbols) > 0: return sympy.simplify(expr1 - expr2) == 0 # Numerical Identity check diff = sympy.simplify(expr1 - expr2) return diff == 0 return await MathPolygraph._run_safe_math_op(calc_equivalence, s1_raw, s2_raw) except asyncio.TimeoutError: logger.warning(f"🛡️ [SOFT FAIL] Equivalence check TIMEOUT (300ms) for {latex1} vs {latex2}") return True # Soft Fail: Assume equivalent if we can't prove otherwise in time except Exception as e: logger.warning(f"🛡️ [SOFT FAIL] Equivalence check failed: {e}") return True # Soft Fail: Assume equivalent on error @staticmethod async def verify_algebraic_consistency(steps: List[dict], topic: str = "GENERAL") -> Tuple[bool, str]: """ V1.3: Checks if a sequence of steps is algebraically consistent. Currently checks if subsequent steps are equivalent (for simplifications). """ # V8.9.4: Skip deep SymPy parsing for discrete sequence steps if topic and "SEQUENCE" in topic.upper(): return True, "" math_steps = [] for step in steps: math = step.get('math_latex') or step.get('block_math') or step.get('math') if math and isinstance(math, str) and math.strip(): # Avoid validating plaintext logic blocks if not _is_plaintext(math): math_steps.append({'id': step.get('step_id', '?'), 'math': math}) if len(math_steps) < 2: return True, "" for i in range(len(math_steps) - 1): s1 = math_steps[i]['math'] s2 = math_steps[i+1]['math'] # Simple heuristic: Only verify if they look like comparable equations/expressions if not await MathPolygraph.are_equivalent(s1, s2): logger.info(f"[POLYGRAPH] Consistency warning between {s1} and {s2}") # We return False only if we are VERY sure. # For now, we'll return False to trigger self-correction as requested. return False, f"ALGEBRAIC_INCONSISTENCY:step_{math_steps[i+1]['id']}" return True, ""