Buckets:
| #!/usr/bin/env python3 | |
| """Shared code detection and prompt intent helpers for Mythos SFT pipeline.""" | |
| from __future__ import annotations | |
| import re | |
| from difflib import SequenceMatcher | |
| CODE_MARKERS = ( | |
| "<!DOCTYPE html>", | |
| "<html", | |
| "```html", | |
| "```javascript", | |
| "```css", | |
| "```js", | |
| "function ", | |
| "const ", | |
| "let ", | |
| "class=", | |
| "import ", | |
| "export ", | |
| "addEventListener", | |
| "requestAnimationFrame", | |
| "getElementById", | |
| "querySelector", | |
| "classList", | |
| ) | |
| WEAK_SOLUTION_PHRASES = ( | |
| "read the files", | |
| "make targeted edits", | |
| "plus additional targeted edits", | |
| "additional targeted edits", | |
| "apply targeted edits", | |
| "inspect the relevant project files", | |
| "outline the smallest safe change", | |
| ) | |
| CODE_REQUEST_PATTERNS = re.compile( | |
| r"\b(" | |
| r"create|build|generate|write|provide|return|output|give me" | |
| r")\b.*\b(" | |
| r"complete|full|entire|whole|single" | |
| r")?\s*(file|index\.html|html|css|js|javascript|component|script|app|code|snippet)" | |
| r"|" | |
| r"\bcomplete\s+(single\s+)?index\.html\b" | |
| r"|" | |
| r"\bfull\s+(file|code|runnable)\b" | |
| r"|" | |
| r"\boutput\s+(the\s+)?full\s+file\b" | |
| r"|" | |
| r"\breturn\s+all\s+file\b" | |
| r"|" | |
| r"\bprovide\s+(the\s+)?code\b" | |
| r"|" | |
| r"\bwrite\s+(complete|full)\b", | |
| re.IGNORECASE, | |
| ) | |
| DIAGNOSIS_TASK_TYPES = frozenset({ | |
| "bug_fix", | |
| "terminal_debug", | |
| "ui_repair", | |
| "migration", | |
| "refactor", | |
| }) | |
| def has_code(text: str) -> bool: | |
| """True when text contains actual code structure, not keyword mentions in prose.""" | |
| t = str(text or "") | |
| if "```" in t: | |
| return True | |
| if "<!DOCTYPE" in t or "<html" in t.lower(): | |
| return True | |
| if re.search(r"\b(function|const|let|class|import|export)\s+[\w{$(\[]", t): | |
| return True | |
| if any(call in t for call in ("addEventListener(", "getElementById(", "querySelector(", "classList.")): | |
| return True | |
| if re.search(r"^\s*[\w.#][\w\-#.[\]=>]*\s*\{", t, re.MULTILINE): | |
| return True | |
| return False | |
| def asks_for_code(prompt: str) -> bool: | |
| p = str(prompt or "").strip() | |
| if not p: | |
| return False | |
| if CODE_REQUEST_PATTERNS.search(p): | |
| return True | |
| lower = p.lower() | |
| quick = ( | |
| "index.html", | |
| "complete file", | |
| "full file", | |
| "full code", | |
| "output the full", | |
| "return all file", | |
| "provide code", | |
| "write code", | |
| "give me code", | |
| "html/css/js", | |
| ) | |
| return any(k in lower for k in quick) | |
| def is_code_generation_intent(row: dict) -> bool: | |
| if asks_for_code(row.get("user_prompt", "")): | |
| return True | |
| task_type = str(row.get("task_type", "")) | |
| if task_type == "feature_build" and asks_for_code(row.get("user_prompt", "")): | |
| return True | |
| solution = str(row.get("solution", "")) | |
| if has_code(solution) and task_type in {"feature_build", "website_vibe"}: | |
| if any(k in str(row.get("user_prompt", "")).lower() for k in ("create", "build", "write", "generate")): | |
| return True | |
| return False | |
| def uses_diagnosis_format(row: dict) -> bool: | |
| task_type = str(row.get("task_type", "")) | |
| if task_type in DIAGNOSIS_TASK_TYPES: | |
| return True | |
| if asks_for_code(row.get("user_prompt", "")): | |
| return False | |
| return task_type not in {"feature_build", "website_vibe", "test_writing"} | |
| def is_weak_solution(solution: str) -> bool: | |
| s = str(solution or "").strip().lower() | |
| if not s: | |
| return True | |
| if has_code(solution): | |
| return False | |
| if any(phrase in s for phrase in WEAK_SOLUTION_PHRASES): | |
| return True | |
| if s.startswith("1)") and not has_code(solution): | |
| # numbered agent_behavior echo without code | |
| return True | |
| return False | |
| def prompt_overlap_ratio(user_prompt: str, assistant: str) -> float: | |
| user = str(user_prompt or "").strip().lower() | |
| assistant = str(assistant or "").strip().lower() | |
| if not user or not assistant: | |
| return 0.0 | |
| return SequenceMatcher(None, user, assistant).ratio() | |
| def assistant_repeats_user_prompt(user_prompt: str, assistant: str, threshold: float = 0.5) -> bool: | |
| user = str(user_prompt or "").strip() | |
| if not user: | |
| return False | |
| assistant_lower = str(assistant or "").lower() | |
| user_lower = user.lower() | |
| if user_lower in assistant_lower: | |
| return True | |
| if prompt_overlap_ratio(user, assistant) >= threshold: | |
| return True | |
| # Diagnosis copying full prompt | |
| if assistant_lower.startswith("diagnosis:"): | |
| diag = assistant_lower.split("plan:", 1)[0] | |
| if user_lower[:80] and user_lower[:80] in diag: | |
| return True | |
| return False | |
| def verification_is_vague(verification: str) -> bool: | |
| v = str(verification or "").strip().lower() | |
| if not v: | |
| return True | |
| vague = ( | |
| "confirm the task requirements are met", | |
| "verify visually in the browser", | |
| "re-test the affected workflow", | |
| "take screenshots", | |
| "works in chrome and firefox", | |
| ) | |
| if any(x in v for x in vague) and not any( | |
| x in v for x in ("click", "open", "run", "npm", "console.assert", "devtools", "assert", "check that") | |
| ): | |
| return True | |
| return False | |
Xet Storage Details
- Size:
- 5.26 kB
- Xet hash:
- b89a2216d3e2c71d2589c2a3a57ffce507eb21e8dd6573349c925525effc8bd3
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.