Download app/services/answer_correction.py from asnannp/DocDoeAI: direct link, hf CLI and curl.
- Browser
- Download file 27.2 kB
-
https://huggingface.co/spaces/asnannp/DocDoeAI/resolve/main/app/services/answer_correction.py
- Command line
-
hf download hf://spaces/asnannp/DocDoeAI/app/services/answer_correction.py
-
curl -L -o answer_correction.py https://huggingface.co/spaces/asnannp/DocDoeAI/resolve/main/app/services/answer_correction.py
27.2 kB
| from __future__ import annotations | |
| import re | |
| from typing import Any | |
| from pydantic import BaseModel, Field | |
| from app.services.syllabus_teacher import ( | |
| SyllabusTeachingContext, | |
| infer_syllabus_context, | |
| ) | |
| SOURCE_ANSWER_NOT_FOUND = "This was not found in your selected source/syllabus." | |
| UNKNOWN_MARKS_PROMPT = "Is this a 2-mark, 3-mark, or 5-mark answer?" | |
| class AnswerCorrectionOutput(BaseModel): | |
| syllabus_position: str | |
| mark_scheme_assumption: str | |
| score: str | |
| what_correct: list[str] = Field(default_factory=list) | |
| marks_lost: list[str] = Field(default_factory=list) | |
| missing_keywords: list[str] = Field(default_factory=list) | |
| corrected_board_answer: str | |
| how_to_improve: list[str] = Field(default_factory=list) | |
| quick_retry_task: str | |
| source_truth: str | |
| source_evidence: list[str] = Field(default_factory=list) | |
| assumed_marks: int | |
| official_scheme_available: bool = False | |
| _STOPWORDS = { | |
| "answer", | |
| "board", | |
| "derive", | |
| "explain", | |
| "find", | |
| "given", | |
| "mark", | |
| "marks", | |
| "question", | |
| "show", | |
| "student", | |
| "write", | |
| } | |
| def build_answer_correction_output( | |
| *, | |
| question: str, | |
| student_answer: str, | |
| subject: str | None = None, | |
| board: str | None = None, | |
| class_level: str | None = None, | |
| marks: Any = None, | |
| source_context: str = "", | |
| source_title: str | None = None, | |
| metadata: dict[str, Any] | None = None, | |
| ) -> dict[str, Any]: | |
| metadata = dict(metadata or {}) | |
| if subject: | |
| metadata["subject"] = subject | |
| if board: | |
| metadata["board"] = board | |
| if class_level: | |
| metadata["class_level"] = class_level | |
| metadata.setdefault("question", question) | |
| clean_source_context = _strip_evidence_policy(source_context) | |
| ctx = infer_syllabus_context( | |
| question, | |
| metadata, | |
| source_context=clean_source_context, | |
| has_source=bool(clean_source_context.strip()), | |
| ) | |
| assumed_marks, marks_unknown = _parse_marks(marks) | |
| mark_scheme_assumption = _mark_scheme_assumption(assumed_marks, marks_unknown) | |
| syllabus_position = _syllabus_position(ctx, source_title) | |
| source_supported, evidence = _source_supports_answer( | |
| question=question, | |
| student_answer=student_answer, | |
| ctx=ctx, | |
| source_context=clean_source_context, | |
| ) | |
| if clean_source_context.strip() and not source_supported: | |
| return AnswerCorrectionOutput( | |
| syllabus_position=syllabus_position, | |
| mark_scheme_assumption=mark_scheme_assumption, | |
| score="Estimated score: not scored against selected source", | |
| what_correct=[ | |
| "Your answer may belong to another chapter, but the selected source does not support this correction.", | |
| ], | |
| marks_lost=[ | |
| f"Source mismatch: {SOURCE_ANSWER_NOT_FOUND}", | |
| "Marks are not estimated from this source because the source evidence is for a different topic.", | |
| ], | |
| missing_keywords=[], | |
| corrected_board_answer=SOURCE_ANSWER_NOT_FOUND, | |
| how_to_improve=[ | |
| "Select the source or syllabus that contains this question.", | |
| "Run correction again after choosing the matching material.", | |
| ], | |
| quick_retry_task="Choose the matching source, then paste the same answer again.", | |
| source_truth=SOURCE_ANSWER_NOT_FOUND, | |
| source_evidence=[], | |
| assumed_marks=assumed_marks, | |
| official_scheme_available=False, | |
| ).model_dump() | |
| subject_key = (subject or ctx.subject or "").lower() | |
| lower_question = _normalise_text(question) | |
| if subject_key == "physics" or "v u at" in lower_question or "equations of motion" in lower_question: | |
| output = _physics_correction( | |
| question=question, | |
| student_answer=student_answer, | |
| ctx=ctx, | |
| assumed_marks=assumed_marks, | |
| marks_unknown=marks_unknown, | |
| source_title=source_title, | |
| source_evidence=evidence, | |
| ) | |
| elif subject_key == "chemistry" or _looks_like_chemistry_numerical(question): | |
| output = _chemistry_correction( | |
| question=question, | |
| student_answer=student_answer, | |
| ctx=ctx, | |
| assumed_marks=assumed_marks, | |
| marks_unknown=marks_unknown, | |
| source_title=source_title, | |
| source_evidence=evidence, | |
| ) | |
| elif subject_key in {"math", "maths", "mathematics"} or _looks_like_math_proof(question): | |
| output = _maths_correction( | |
| question=question, | |
| student_answer=student_answer, | |
| ctx=ctx, | |
| assumed_marks=assumed_marks, | |
| marks_unknown=marks_unknown, | |
| source_title=source_title, | |
| source_evidence=evidence, | |
| ) | |
| else: | |
| output = _general_correction( | |
| question=question, | |
| student_answer=student_answer, | |
| ctx=ctx, | |
| assumed_marks=assumed_marks, | |
| marks_unknown=marks_unknown, | |
| source_title=source_title, | |
| source_evidence=evidence, | |
| ) | |
| if marks_unknown and UNKNOWN_MARKS_PROMPT not in output["mark_scheme_assumption"]: | |
| output["mark_scheme_assumption"] = f"{output['mark_scheme_assumption']} {UNKNOWN_MARKS_PROMPT}" | |
| return output | |
| def merge_ai_answer_correction( | |
| deterministic_output: dict[str, Any], | |
| ai_output: dict[str, Any] | None, | |
| ) -> dict[str, Any]: | |
| """Use live AI wording where safe, while preserving source/marks guardrails.""" | |
| if not ai_output or deterministic_output.get("source_truth") == SOURCE_ANSWER_NOT_FOUND: | |
| return deterministic_output | |
| merged = dict(deterministic_output) | |
| # Only let AI enhance the corrected answer and improvement tips. | |
| # Keep deterministic marks_lost, what_correct, missing_keywords for consistency. | |
| ai_enhanceable = { | |
| "corrected_board_answer", | |
| "how_to_improve", | |
| "quick_retry_task", | |
| } | |
| for key in ai_enhanceable: | |
| value = ai_output.get(key) | |
| if isinstance(value, list) and value: | |
| merged[key] = [str(item) for item in value if str(item).strip()] | |
| elif isinstance(value, str) and value.strip(): | |
| merged[key] = value.strip() | |
| score = str(ai_output.get("score") or "").strip() | |
| if score: | |
| merged["score"] = score if "estimated" in score.lower() else f"Estimated {score[0].lower()}{score[1:]}" | |
| # These are deterministic contract fields; do not let the model loosen them. | |
| merged["syllabus_position"] = deterministic_output.get("syllabus_position", "") | |
| merged["mark_scheme_assumption"] = deterministic_output.get("mark_scheme_assumption", "") | |
| merged["source_truth"] = deterministic_output.get("source_truth", "") | |
| merged["source_evidence"] = deterministic_output.get("source_evidence", []) | |
| merged["assumed_marks"] = deterministic_output.get("assumed_marks", 5) | |
| merged["official_scheme_available"] = False | |
| return AnswerCorrectionOutput(**merged).model_dump() | |
| def answer_correction_task_prompt(*, question: str, student_answer: str, marks: int) -> str: | |
| return ( | |
| "Correct this student answer like a strict but helpful tuition teacher. " | |
| "Return marks lost, missing exact keywords, and a corrected board-exam answer. " | |
| f"Question: {question}\n" | |
| f"Student answer: {student_answer}\n" | |
| f"Marks: {marks}\n" | |
| "Label the score as estimated unless an official mark scheme is present." | |
| ) | |
| def _physics_correction( | |
| *, | |
| question: str, | |
| student_answer: str, | |
| ctx: SyllabusTeachingContext, | |
| assumed_marks: int, | |
| marks_unknown: bool, | |
| source_title: str | None, | |
| source_evidence: list[str], | |
| ) -> dict[str, Any]: | |
| answer = _normalise_text(student_answer) | |
| lost: list[tuple[float, str]] = [] | |
| correct: list[str] = [] | |
| if "acceleration" in answer and ("velocity" in answer or "v" in answer) and ("time" in answer or "t" in answer): | |
| correct.append("You connected acceleration with velocity and time.") | |
| if _has_formula_vuat(answer): | |
| correct.append("You wrote the final formula v = u + at.") | |
| if "change" not in answer and "v-u" not in answer and "v - u" not in answer: | |
| lost.append((1.0, "acceleration definition is incomplete; it must be change in velocity per unit time.")) | |
| if "uniform" not in answer and "constant" not in answer: | |
| lost.append((0.5, "did not mention uniform acceleration, the key assumption for this derivation.")) | |
| if not _mentions_symbols(answer, ("u", "v", "a", "t")): | |
| lost.append((1.0, "did not define symbols u, v, a, and t.")) | |
| if "(v-u)/t" not in answer and "v-u" not in answer and "v - u" not in answer: | |
| lost.append((1.0, "skipped the derivation step a = (v - u) / t.")) | |
| if "therefore" not in answer and "hence" not in answer: | |
| lost.append((0.5, "final formula was not presented as a derived result.")) | |
| if not correct: | |
| correct.append("You attempted the correct topic and tried to connect acceleration with the final formula.") | |
| score_value = _score_value(assumed_marks, lost) | |
| mark_scheme = _mark_scheme_assumption(assumed_marks, marks_unknown) | |
| output = AnswerCorrectionOutput( | |
| syllabus_position=_syllabus_position(ctx, source_title), | |
| mark_scheme_assumption=mark_scheme, | |
| score=_score_text(score_value, assumed_marks), | |
| what_correct=correct, | |
| marks_lost=_format_lost(lost), | |
| missing_keywords=[ | |
| "change in velocity", | |
| "uniform acceleration", | |
| "initial velocity u", | |
| "final velocity v", | |
| "time t", | |
| "a = (v - u) / t", | |
| "v = u + at", | |
| ], | |
| corrected_board_answer=( | |
| "For uniformly accelerated motion, let u be the initial velocity, v the final velocity, " | |
| "a the acceleration, and t the time. Acceleration is the change in velocity per unit time, " | |
| "so a = (v - u) / t. Therefore at = v - u, and v = u + at. " | |
| "This equation is valid only when acceleration is uniform. SI unit of velocity is m s^-1." | |
| ), | |
| how_to_improve=[ | |
| "Start derivations by defining every symbol.", | |
| "Write the assumption before the equation.", | |
| "Show the algebra step before the final formula.", | |
| ], | |
| quick_retry_task="Rewrite only the derivation from a = (v - u) / t to v = u + at in three lines.", | |
| source_truth=_source_truth(source_title), | |
| source_evidence=source_evidence, | |
| assumed_marks=assumed_marks, | |
| official_scheme_available=False, | |
| ) | |
| return output.model_dump() | |
| _KNOWN_MOLAR_MASSES: dict[str, str] = { | |
| "water": "18", | |
| "h2o": "18", | |
| "co2": "44", | |
| "carbon dioxide": "44", | |
| "naoh": "40", | |
| "hcl": "36.5", | |
| "h2so4": "98", | |
| "nacl": "58.5", | |
| "caco3": "100", | |
| "o2": "32", | |
| "n2": "28", | |
| "h2": "2", | |
| "ch4": "16", | |
| "c6h12o6": "180", | |
| "glucose": "180", | |
| "ethanol": "46", | |
| "c2h5oh": "46", | |
| } | |
| def _parse_question_values(question: str) -> tuple[str, str, str]: | |
| """Extract mass value, substance name, and molar mass from a chemistry numerical question. | |
| Returns (mass_str, substance, molar_mass_str). Any field may be empty if not detected. | |
| """ | |
| q_lower = question.lower() | |
| mass = "" | |
| substance = "" | |
| molar_mass = "" | |
| mass_match = re.search(r"(\d+(?:\.\d+)?)\s*g\s*(?:of|in)\s+([A-Za-z0-9]+)", q_lower) | |
| if mass_match: | |
| mass = mass_match.group(1) | |
| substance = mass_match.group(2) | |
| if not mass: | |
| mass_match2 = re.search(r"mass\s+(?:of\s+)?([A-Za-z0-9]+)\s*=\s*(\d+(?:\.\d+)?)\s*g", q_lower) | |
| if mass_match2: | |
| substance = mass_match2.group(1) | |
| mass = mass_match2.group(2) | |
| mm_match = re.search(r"molar\s+mass\s+(?:of\s+)?[A-Za-z0-9]*\s*=\s*(\d+(?:\.\d+)?)\s*g", q_lower) | |
| if mm_match: | |
| molar_mass = mm_match.group(1) | |
| if not substance: | |
| for token in ("h2o", "water", "co2", "naoh", "hcl", "h2so4", "nacl", "caco3"): | |
| if token in q_lower: | |
| substance = token | |
| break | |
| if not molar_mass and substance: | |
| molar_mass = _KNOWN_MOLAR_MASSES.get(substance, "") | |
| return mass, substance, molar_mass | |
| def _compute_correct_answer(mass: str, molar_mass: str) -> str: | |
| """Compute the correct numerical answer as a string, rounded to 2 decimals.""" | |
| try: | |
| m = float(mass) | |
| mm = float(molar_mass) | |
| if mm == 0: | |
| return "" | |
| result = m / mm | |
| if result == int(result): | |
| return str(int(result)) | |
| return f"{result:.2f}".rstrip("0").rstrip(".") | |
| except (ValueError, ZeroDivisionError): | |
| return "" | |
| def _chemistry_correction( | |
| *, | |
| question: str, | |
| student_answer: str, | |
| ctx: SyllabusTeachingContext, | |
| assumed_marks: int, | |
| marks_unknown: bool, | |
| source_title: str | None, | |
| source_evidence: list[str], | |
| ) -> dict[str, Any]: | |
| answer = _normalise_text(student_answer) | |
| lost: list[tuple[float, str]] = [] | |
| correct: list[str] = [] | |
| q_mass, q_substance, q_molar_mass = _parse_question_values(question) | |
| correct_answer = _compute_correct_answer(q_mass, q_molar_mass) if q_mass and q_molar_mass else "" | |
| substance_label = q_substance or "substance" | |
| if any(token in answer for token in ("mole", "mol", "molar", "equation", "formula")): | |
| correct.append("You identified that this needs a formula/equation method.") | |
| if "given" not in answer and q_mass and q_mass not in answer: | |
| lost.append((0.5, "did not write the given value clearly.")) | |
| uses_division = "/" in answer or "divided" in answer or "over" in answer | |
| uses_multiplication = ("x" in answer and "/" not in answer) or "times" in answer or "*" in answer | |
| has_formula = "formula" in answer or "n=" in answer or "mass/molar" in answer or uses_division | |
| if not has_formula: | |
| if uses_multiplication: | |
| lost.append((1.0, "wrong formula: used multiplication instead of division. Correct formula: n = given mass / molar mass.")) | |
| else: | |
| lost.append((1.0, "missing formula n = given mass / molar mass.")) | |
| elif q_mass and q_molar_mass: | |
| expected_sub = f"{q_mass}/{q_molar_mass}" | |
| expected_sub_spaces = f"{q_mass} / {q_molar_mass}" | |
| if expected_sub not in answer and expected_sub_spaces not in answer: | |
| if correct_answer and correct_answer not in answer: | |
| lost.append((1.0, f"missing or wrong substitution step. Expected: n = {q_mass} / {q_molar_mass}")) | |
| if correct_answer: | |
| student_has_correct_answer = ( | |
| correct_answer in answer | |
| or (correct_answer.rstrip("0").rstrip(".") in answer and len(correct_answer.rstrip("0").rstrip(".")) > 0) | |
| ) | |
| if not student_has_correct_answer and "mol" in answer: | |
| lost.append((1.0, f"wrong final answer. Correct answer: {correct_answer} mol.")) | |
| if "mol" not in answer and "mole" not in answer: | |
| lost.append((0.5, "final answer has no unit.")) | |
| if not correct: | |
| correct.append("You attempted the chemistry calculation, but the board-answer steps are not complete.") | |
| if q_mass and q_molar_mass and correct_answer: | |
| corrected_board = ( | |
| f"Given: mass of {substance_label} = {q_mass} g. Molar mass of {substance_label} = {q_molar_mass} g mol^-1. " | |
| f"Formula: number of moles, n = given mass / molar mass. " | |
| f"Substitution: n = {q_mass} / {q_molar_mass} = {correct_answer}. Answer: {correct_answer} mol." | |
| ) | |
| else: | |
| corrected_board = ( | |
| "Given: write the given mass and molar mass clearly. " | |
| "Formula: number of moles, n = given mass / molar mass. " | |
| "Substitution: n = given mass / molar mass. Answer: calculate and add unit mol." | |
| ) | |
| output = AnswerCorrectionOutput( | |
| syllabus_position=_syllabus_position(ctx, source_title), | |
| mark_scheme_assumption=_mark_scheme_assumption(assumed_marks, marks_unknown), | |
| score=_score_text(_score_value(assumed_marks, lost), assumed_marks), | |
| what_correct=correct, | |
| marks_lost=_format_lost(lost), | |
| missing_keywords=[ | |
| "given", | |
| "molar mass", | |
| "formula", | |
| "substitution", | |
| "mol", | |
| ], | |
| corrected_board_answer=corrected_board, | |
| how_to_improve=[ | |
| "Use the order: given, formula, substitution, answer, unit.", | |
| "Always divide mass by molar mass, never multiply.", | |
| "Do not skip units in the final line.", | |
| ], | |
| quick_retry_task="Write the same numerical again in four lines: Given, Formula, Substitution, Answer.", | |
| source_truth=_source_truth(source_title), | |
| source_evidence=source_evidence, | |
| assumed_marks=assumed_marks, | |
| official_scheme_available=False, | |
| ) | |
| return output.model_dump() | |
| def _maths_correction( | |
| *, | |
| question: str, | |
| student_answer: str, | |
| ctx: SyllabusTeachingContext, | |
| assumed_marks: int, | |
| marks_unknown: bool, | |
| source_title: str | None, | |
| source_evidence: list[str], | |
| ) -> dict[str, Any]: | |
| answer = _normalise_text(student_answer) | |
| lost: list[tuple[float, str]] = [] | |
| correct: list[str] = [] | |
| has_proof_steps = ( | |
| "pythagoras" in answer | |
| or "a^2" in answer | |
| or "hypotenuse" in answer | |
| or "opposite" in answer | |
| or "adjacent" in answer | |
| or "right triangle" in answer | |
| or "=" in answer and ("sin" in answer or "cos" in answer) | |
| ) | |
| has_reasoning = ( | |
| "therefore" in answer | |
| or "hence" in answer | |
| or "thus" in answer | |
| or "by pythagoras" in answer | |
| or "by definition" in answer | |
| or "by identity" in answer | |
| or "definition of" in answer | |
| or "opposite" in answer and "hypotenuse" in answer | |
| ) | |
| if "sin" in answer and "cos" in answer: | |
| correct.append("You used the correct identity area: sine and cosine.") | |
| if "1" in answer and ("=" in answer or "result" in answer or "prove" in question.lower()): | |
| correct.append("You reached the expected final result.") | |
| if not has_proof_steps: | |
| lost.append((1.5, "answer is too short for a proof. No theorem, identity, or reasoning steps shown.")) | |
| elif not has_reasoning: | |
| lost.append((1.0, "missing reasoning or justification for each step.")) | |
| if "therefore" not in answer and "hence" not in answer and "thus" not in answer: | |
| lost.append((0.5, "final result is not concluded properly with a concluding statement.")) | |
| if not correct: | |
| correct.append("You attempted the proof, but the reasoning chain is missing.") | |
| output = AnswerCorrectionOutput( | |
| syllabus_position=_syllabus_position(ctx, source_title), | |
| mark_scheme_assumption=_mark_scheme_assumption(assumed_marks, marks_unknown), | |
| score=_score_text(_score_value(assumed_marks, lost), assumed_marks), | |
| what_correct=correct, | |
| marks_lost=_format_lost(lost), | |
| missing_keywords=[ | |
| "Given", | |
| "To prove", | |
| "right triangle", | |
| "sin x = opposite / hypotenuse", | |
| "cos x = adjacent / hypotenuse", | |
| "Pythagoras theorem", | |
| "therefore", | |
| ], | |
| corrected_board_answer=( | |
| "Given: a right triangle with angle x. To prove: sin^2 x + cos^2 x = 1. " | |
| "Let opposite side = a, adjacent side = b, and hypotenuse = c. " | |
| "sin x = a/c and cos x = b/c. Therefore sin^2 x + cos^2 x = a^2/c^2 + b^2/c^2 " | |
| "= (a^2 + b^2)/c^2. By Pythagoras theorem, a^2 + b^2 = c^2. " | |
| "Hence sin^2 x + cos^2 x = c^2/c^2 = 1." | |
| ), | |
| how_to_improve=[ | |
| "Start proof answers with Given and To prove.", | |
| "Write one reason beside each major step.", | |
| "Do not quote the identity as proof; derive it from definitions or theorem.", | |
| ], | |
| quick_retry_task="Rewrite the proof with one reason after each equality.", | |
| source_truth=_source_truth(source_title), | |
| source_evidence=source_evidence, | |
| assumed_marks=assumed_marks, | |
| official_scheme_available=False, | |
| ) | |
| return output.model_dump() | |
| def _general_correction( | |
| *, | |
| question: str, | |
| student_answer: str, | |
| ctx: SyllabusTeachingContext, | |
| assumed_marks: int, | |
| marks_unknown: bool, | |
| source_title: str | None, | |
| source_evidence: list[str], | |
| ) -> dict[str, Any]: | |
| answer = student_answer.strip() | |
| lost = [] | |
| if len(answer.split()) < 20 and assumed_marks >= 3: | |
| lost.append((1.0, "answer is too short for the assumed marks.")) | |
| if not any(char in answer for char in (".", ";", ":")): | |
| lost.append((0.5, "answer needs clearer point-wise presentation.")) | |
| if not _shared_keywords(question, answer): | |
| lost.append((1.0, "missing exact keywords from the question.")) | |
| topic = ctx.topic or question[:80] | |
| output = AnswerCorrectionOutput( | |
| syllabus_position=_syllabus_position(ctx, source_title), | |
| mark_scheme_assumption=_mark_scheme_assumption(assumed_marks, marks_unknown), | |
| score=_score_text(_score_value(assumed_marks, lost), assumed_marks), | |
| what_correct=["You attempted the question and gave a relevant start."], | |
| marks_lost=_format_lost(lost), | |
| missing_keywords=_keyword_terms(question)[:8], | |
| corrected_board_answer=( | |
| f"Board-answer version for {topic}: start with a direct definition, add exact keywords from the question, " | |
| "write points according to marks, and finish with one example/formula/condition if the subject needs it." | |
| ), | |
| how_to_improve=[ | |
| "Match answer length to marks.", | |
| "Use exact subject keywords.", | |
| "Write point-wise instead of one loose sentence.", | |
| ], | |
| quick_retry_task="Rewrite the answer in three bullet points with the exact keywords underlined.", | |
| source_truth=_source_truth(source_title), | |
| source_evidence=source_evidence, | |
| assumed_marks=assumed_marks, | |
| official_scheme_available=False, | |
| ) | |
| return output.model_dump() | |
| def _source_supports_answer( | |
| *, | |
| question: str, | |
| student_answer: str, | |
| ctx: SyllabusTeachingContext, | |
| source_context: str, | |
| ) -> tuple[bool, list[str]]: | |
| if not source_context.strip(): | |
| return True, [] | |
| source_norm = _normalise_text(source_context) | |
| terms = _keyword_terms(" ".join([question, student_answer, ctx.topic, ctx.chapter, ctx.syllabus_point])) | |
| if not terms: | |
| return True, [] | |
| hits = [term for term in terms if term in source_norm] | |
| threshold = 2 if len(terms) >= 4 else 1 | |
| evidence = _evidence_snippets(source_context, hits or terms) | |
| return len(hits) >= threshold, evidence | |
| def _evidence_snippets(source_context: str, terms: list[str]) -> list[str]: | |
| snippets: list[str] = [] | |
| sentences = re.split(r"(?<=[.!?])\s+|\n+", source_context) | |
| for sentence in sentences: | |
| clean = sentence.strip() | |
| if len(clean) < 12: | |
| continue | |
| normal = _normalise_text(clean) | |
| if any(term in normal for term in terms[:8]): | |
| snippets.append(clean[:240]) | |
| if len(snippets) >= 2: | |
| break | |
| return snippets | |
| def _parse_marks(marks: Any) -> tuple[int, bool]: | |
| if isinstance(marks, int) and 1 <= marks <= 6: | |
| return marks, False | |
| if isinstance(marks, str): | |
| raw = marks.strip().lower() | |
| if raw and raw not in {"not_sure", "not sure", "unknown", "unsure"}: | |
| match = re.search(r"\d+", raw) | |
| if match: | |
| value = int(match.group(0)) | |
| if 1 <= value <= 6: | |
| return value, False | |
| return 5, True | |
| def _mark_scheme_assumption(marks: int, unknown: bool) -> str: | |
| if unknown: | |
| return f"Marks not provided. Assuming this is a {marks}-mark answer. {UNKNOWN_MARKS_PROMPT}" | |
| return f"Assuming this is a {marks}-mark answer." | |
| def _score_value(total: int, lost: list[tuple[float, str]]) -> float: | |
| lost_total = sum(item[0] for item in lost) | |
| return max(0.0, min(float(total), float(total) - lost_total)) | |
| def _score_text(score: float, total: int) -> str: | |
| score_value = int(score) if score.is_integer() else score | |
| return f"Estimated score: {score_value}/{total}" | |
| def _format_lost(lost: list[tuple[float, str]]) -> list[str]: | |
| if not lost: | |
| return ["Lost 0 marks: answer covers the expected board points for the assumed marks."] | |
| return [f"Lost {_format_mark(mark)} mark{'s' if mark != 1 else ''}: {reason}" for mark, reason in lost] | |
| def _format_mark(mark: float) -> str: | |
| return str(int(mark)) if mark.is_integer() else str(mark) | |
| def _source_truth(source_title: str | None) -> str: | |
| if source_title: | |
| return f"Corrected using selected source first: {source_title}." | |
| return "No selected source was attached. This correction uses standard board-answer rules." | |
| def _syllabus_position(ctx: SyllabusTeachingContext, source_title: str | None) -> str: | |
| parts = [ | |
| " ".join(part for part in (ctx.board, ctx.class_level) if part).strip(), | |
| ctx.subject, | |
| ctx.chapter, | |
| ctx.topic, | |
| ] | |
| position = " -> ".join(part for part in parts if part) | |
| if not position: | |
| position = ctx.label | |
| if source_title: | |
| return f"{position}\nUsing: {source_title}" | |
| return position | |
| def _strip_evidence_policy(context: str) -> str: | |
| return context.split("# Evidence-bound answer policy", 1)[0].strip() | |
| def _normalise_text(value: str) -> str: | |
| text = value.lower() | |
| text = text.replace("²", "^2").replace("−", "-") | |
| text = re.sub(r"[^a-z0-9^=+\-/.\s]", " ", text) | |
| return re.sub(r"\s+", " ", text).strip() | |
| def _keyword_terms(value: str) -> list[str]: | |
| terms = [] | |
| for term in re.findall(r"[a-z0-9]+", _normalise_text(value)): | |
| if len(term) < 4 or term in _STOPWORDS: | |
| continue | |
| if term not in terms: | |
| terms.append(term) | |
| return terms | |
| def _shared_keywords(question: str, answer: str) -> list[str]: | |
| answer_norm = _normalise_text(answer) | |
| return [term for term in _keyword_terms(question) if term in answer_norm] | |
| def _has_formula_vuat(answer: str) -> bool: | |
| compact = re.sub(r"\s+", "", answer.lower()) | |
| return "v=u+at" in compact or "v=u+a*t" in compact or "v=u+at" in compact or "v=u+ a t" in answer.lower() | |
| def _mentions_symbols(answer: str, symbols: tuple[str, ...]) -> bool: | |
| return all(re.search(rf"\b{re.escape(symbol)}\b", answer) for symbol in symbols) | |
| def _looks_like_chemistry_numerical(question: str) -> bool: | |
| lower = _normalise_text(question) | |
| return any(token in lower for token in ("mole", "molar", "mass", "solution", "calculate", "chemistry")) | |
| def _looks_like_math_proof(question: str) -> bool: | |
| lower = _normalise_text(question) | |
| return any(token in lower for token in ("prove", "proof", "sin", "cos", "theorem", "identity")) | |