| |
| import re |
| import logging |
| from typing import List, Optional |
| import domain.telemetry as telemetry |
| from domain.semantic_bank import get_diversity_engine |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
|
|
| |
|
|
| def validate_placeholders(renderer_text: str, provided_ids: List[str]) -> bool: |
| """ |
| V7.2: Bi-directional placeholder validation. |
| |
| Check 1: Every ID the server provided MUST appear in the text as {{id}}. |
| Check 2: No placeholder the LLM invented (not in provided_ids) may appear. |
| |
| Either failure โ Fail Closed (Hint Mode). |
| """ |
| |
| for step_id in provided_ids: |
| expected = "{{" + step_id + "}}" |
| if expected not in renderer_text: |
| logger.warning( |
| f"[RENDERER GUARD] Missing placeholder for server-provided ID '{step_id}'. " |
| f"Expected '{expected}' in renderer output." |
| ) |
| telemetry.emit_renderer_placeholder_violation("missing", step_id) |
| return False |
|
|
| |
| found_placeholders = re.findall(r'\{\{(\w+)\}\}', renderer_text) |
| for ph in found_placeholders: |
| if ph not in provided_ids: |
| logger.warning( |
| f"[RENDERER GUARD] Invented placeholder '{{{{{{ph}}}}}}' detected! " |
| f"Not in server-provided IDs: {provided_ids}" |
| ) |
| telemetry.emit_renderer_placeholder_violation("invented", ph) |
| return False |
|
|
| return True |
|
|
|
|
| def inject_signed_results(renderer_text: str, signed_steps: List[dict]) -> str: |
| """ |
| Replaces {{step_id}} placeholders in the renderer's text with the actual |
| computed expressions from the server's signed step list. |
| Only called AFTER validate_placeholders() passes. |
| """ |
| result = renderer_text |
| for step in signed_steps: |
| placeholder = "{{" + step["id"] + "}}" |
| result = result.replace(placeholder, f"`{step['expression']}`") |
| return result |
|
|
|
|
| |
|
|
| class PedagogicalRenderer: |
| """ |
| V7.2: LLM #2 โ Pedagogical Renderer. |
| |
| Receives the server's signed step IDs (NOT expressions) and the Planner's rationale. |
| Produces a pedagogical explanation in Hebrew using only {{step_id}} placeholders. |
| Never computes or writes math itself. |
| """ |
|
|
| def __init__(self, llm_gateway): |
| self.llm_gateway = llm_gateway |
|
|
| |
| |
| def _build_renderer_prompt(self, pedagogical_rationale: str, step_ids: List[str]) -> str: |
| ids_list = ", ".join(f"{{{{{sid}}}}}" for sid in step_ids) |
| return f"""You are a Pedagogical Narrator for a Hebrew math tutoring system. |
| Your role is to explain the INTUITION behind each solution step, not to compute anything. |
| The server has already computed all mathematics with certainty โ your job is to tell the story. |
| |
| PEDAGOGICAL RATIONALE: |
| {pedagogical_rationale} |
| |
| SERVER PLACEHOLDERS (use EXACTLY as written): |
| {ids_list} |
| |
| RULES: |
| 1. Hebrew only. |
| 2. Explain the intuition and logic behind each step in human, encouraging language. |
| Use analogies, real-world stories, or visual metaphors. |
| 3. Embed ALL server placeholders exactly: {ids_list} |
| 4. FORBIDDEN โ do NOT write: any digit, number, variable, operator, fraction, or math symbol. |
| Rule: if you feel the urge to write a number or expression, describe its meaning in words instead. |
| Use {{{{step_id}}}} to present the server's computed result. No '=', '/', '^', 'sin(x)', 'pi', etc. |
| 5. No invented placeholders. Only server-provided IDs above. |
| |
| EXAMPLE (single approved format โ "math textbook narrator style"): |
| โ
CORRECT: "ืืื ืืืฆืื ืืช ื ืงืืืืช ืืืืชืื ืขื ืฆืืจ ื-y, ื ืฆืื ืืคืก ืืขืจื ื-x ืื ืงืื: {{{{step_1}}}}." |
| โ FORBIDDEN: "ืืื ืืืฆืื ืืช ืืืืชืื ื ืฆืื x=0 ืื ืงืื y=4." (raw numbers/symbols โ blocked by Guardian) |
| """ |
|
|
| async def render( |
| self, |
| pedagogical_rationale: str, |
| signed_steps: List[dict], |
| action: Optional[str] = None, |
| ) -> dict: |
| """ |
| V7.2.5 Semantic Composer: |
| 1. Try DiversityEngine.compose(action) โ deterministic narrative from SemanticBank. |
| 2. If no bank entry exists โ falls back to LLM #2 (Graceful Degradation). |
| Both paths run renderer_guard() + scan_for_math_leakage() before returning. |
| """ |
| step_ids = [s["id"] for s in signed_steps] |
|
|
| |
| if action is not None: |
| engine = get_diversity_engine() |
| composed = engine.compose(action, signed_steps) |
| if composed is not None: |
| logger.info(f"[RENDERER] ๐ SemanticBank hit for '{action}'. Using deterministic narrative.") |
| |
| final_text = inject_signed_results(composed, signed_steps) |
| |
| if not renderer_guard(composed): |
| logger.error("[RENDERER] renderer_guard failed on SemanticBank output!") |
| return {"success": False, "reason": "SEMANTIC_BANK_GUARD_VIOLATION"} |
| if not scan_for_math_leakage(final_text): |
| logger.error("[RENDERER] Leakage scan failed on SemanticBank output!") |
| return {"success": False, "reason": "SEMANTIC_BANK_LEAKAGE"} |
| logger.info("โ
[RENDERER] Semantic Composer output approved by all guards.") |
| return {"success": True, "rendered_text": final_text, "source": "semantic_bank"} |
|
|
| |
| logger.info(f"[RENDERER] No SemanticBank entry for action='{action}'. Using LLM #2 fallback.") |
| step_ids = [s["id"] for s in signed_steps] |
|
|
| system_prompt = self._build_renderer_prompt(pedagogical_rationale, step_ids) |
| user_prompt = ( |
| f"Write the pedagogical explanation for this solution. " |
| f"Use the placeholders: {', '.join(step_ids)}" |
| ) |
|
|
| logger.info(f"[RENDERER] Requesting pedagogical explanation for steps: {step_ids}") |
|
|
| try: |
| raw_text = await self.llm_gateway.generate_raw(system_prompt, user_prompt) |
|
|
| |
| if not validate_placeholders(raw_text, step_ids): |
| logger.error("[RENDERER] Placeholder guard FAILED. Failing closed.") |
| return {"success": False, "reason": "PLACEHOLDER_GUARD_VIOLATION"} |
|
|
| |
| |
| |
| if not renderer_guard(raw_text): |
| logger.error("[RENDERER] renderer_guard FAILED on raw LLM output. Failing closed.") |
| telemetry.emit_renderer_leakage("renderer_guard_raw") |
| return {"success": False, "reason": "RENDERER_GUARD_VIOLATION"} |
|
|
| |
| final_text = inject_signed_results(raw_text, signed_steps) |
|
|
| |
| if not scan_for_math_leakage(final_text): |
| logger.error( |
| "[RENDERER GUARDIAN] Math leakage in final injected text! Failing closed." |
| ) |
| telemetry.emit_renderer_leakage("post-injection-leak") |
| return {"success": False, "reason": "RENDERER_GUARDIAN_LEAKAGE"} |
|
|
| logger.info("[RENDERER] โ
Both guards passed. Pedagogical explanation approved.") |
| return {"success": True, "rendered_text": final_text} |
|
|
| except Exception as e: |
| logger.error(f"[RENDERER] LLM failure: {e}") |
| return {"success": False, "reason": f"LLM_FAILURE: {e}"} |
|
|
|
|
| |
|
|
| def renderer_guard(text: str) -> bool: |
| """ |
| V7.2.4 Structural Output Lock โ strict blacklist scan on RAW LLM output. |
| Runs BEFORE placeholder injection to catch violations at the source. |
| |
| Blocks: individual digits, operator chars, named math functions (sin/cos/tan/sqrt/pi). |
| This is stricter than scan_for_math_leakage() which uses a whitelist on the FINAL text. |
| |
| Returns True (safe) / False (fail closed). |
| """ |
| |
| clean = re.sub(r'\{\{.*?\}\}', '', text) |
|
|
| |
| FORBIDDEN_PATTERN = r'[0-9]|[+*/^=]|(?<![a-zA-Z\u0590-\u05FF])(sin|cos|tan|sqrt|pi)(?![a-zA-Z\u0590-\u05FF])' |
|
|
| match = re.search(FORBIDDEN_PATTERN, clean) |
| if match: |
| logger.warning( |
| f"[RENDERER_GUARD] Forbidden content in raw LLM output: '{match.group(0)}'. " |
| f"Failing closed." |
| ) |
| return False |
| return True |
|
|
|
|
| |
|
|
| def scan_for_math_leakage(rendered_text: str) -> bool: |
| """ |
| V7.2.3 Hardened Whitelist Scan (NOT a blacklist). |
| After removing {{...}} placeholders, the remaining text may ONLY contain: |
| - Hebrew letters (Unicode block U+0590โU+05FF) |
| - English letters (for natural language words) |
| - Spaces and basic punctuation (. , ; : ! ? ' " -) |
| |
| Two-pass enforcement: |
| Pass 1 โ Full whitelist: any forbidden char โ REJECT. |
| Pass 2 โ Consecutive math chars: >2 consecutive math symbols outside {{}} |
| catches patterns like '3((9)*(+1))', '^2+4', '/6)' even if |
| pass 1 might miss edge-case single chars. |
| """ |
| |
| clean_text = re.sub(r'\{\{\w+\}\}', '', rendered_text).strip() |
|
|
| |
| |
| |
| |
| clean_text = re.sub(r'`[^`]+`', '', clean_text).strip() |
|
|
| if not clean_text: |
| return True |
|
|
| |
| ALLOWED_PATTERN = r'^[\u0590-\u05FFa-zA-Z\s.,;:!?\-\"\'\u2014\u2013\u2012\u200f\u200e`]+$' |
| whitelist_ok = bool(re.match(ALLOWED_PATTERN, clean_text)) |
|
|
| if not whitelist_ok: |
| |
| violations = re.sub(r'[\u0590-\u05FFa-zA-Z\s.,;:!?\-\"\'\u2014\u2013\u2012\u200f\u200e`]+', '', clean_text) |
| logger.warning(f"[UI_GATE] Math leakage detected! Offending chars: '{violations[:50]}'") |
| telemetry.emit_renderer_leakage(violations[:50]) |
| return False |
|
|
| |
| |
| |
| MATH_CHAR_PATTERN = r'[0-9=+\-*/^<>|\\(){}\[\]%@#&~]{3,}' |
| consecutive_match = re.search(MATH_CHAR_PATTERN, clean_text) |
| if consecutive_match: |
| chain = consecutive_match.group(0) |
| logger.warning( |
| f"[UI_GATE] Consecutive math char sequence detected: '{chain[:50]}'. " |
| f"Failing closed (Pass 2 โ V7.2.3 hardening)." |
| ) |
| telemetry.emit_renderer_leakage(f"consecutive_chain:{chain[:30]}") |
| return False |
|
|
| return True |
|
|