diff --git a/ai_service/app/config.py b/ai_service/app/config.py index 4af3ad2128..76be4151be 100644 --- a/ai_service/app/config.py +++ b/ai_service/app/config.py @@ -204,6 +204,24 @@ class Settings(BaseSettings): jwt_algorithm: str = "HS256" jwt_token_expiry_minutes: int = 43200 # 30 days in minutes (matching Java 2592000000ms) + # ---- Copy-check VISION grading ---------------------------------------- + # When true (product default), the copy-check pipeline sends the actual + # answer-sheet page image(s) to a vision LLM and grades the handwriting from + # the image, using PaddleOCR only as an assistive text hint + annotation + # anchor. When false, it falls back to the legacy text-only OCR grader + # unchanged. Toggle with COPY_CHECK_VISION_GRADING=false. + copy_check_vision_grading: bool = True + # DPI to rasterize pages at for the LLM (150–200 is plenty for handwriting; + # higher just inflates the payload). Longest side is then downscaled to + # copy_check_vision_max_image_px before JPEG encoding. + copy_check_vision_dpi: int = 150 + copy_check_vision_max_image_px: int = 2000 + # Cost guards: pages sent per question, and total image sends per copy. + copy_check_vision_max_pages_per_question: int = 3 + copy_check_vision_max_images_per_copy: int = 40 + # Hard ceiling on pages rasterized from one PDF (pathological-document guard). + copy_check_vision_max_pages_per_copy: int = 50 + # Internal service-to-service auth. # Used by admin_core_service when calling /credits/v1/internal/* endpoints # (credit-pack purchase fulfillment from the Razorpay webhook handler). diff --git a/ai_service/app/services/copy_check/grader.py b/ai_service/app/services/copy_check/grader.py index d207d03045..6699632a46 100644 --- a/ai_service/app/services/copy_check/grader.py +++ b/ai_service/app/services/copy_check/grader.py @@ -14,16 +14,26 @@ import json import logging +import os import re from typing import Any, Optional +from ...config import get_settings from ..chat_llm_client import ChatLLMClient -from .prompt_builder import GRADING_SYSTEM, build_grading_prompt +from .prompt_builder import GRADING_SYSTEM, GRADING_SYSTEM_VISION, build_grading_prompt logger = logging.getLogger(__name__) +# Text-only path (legacy: grades the printed-text OCR of the handwriting). DEFAULT_MODEL = "google/gemini-2.5-flash-lite" ESCALATION_MODEL = "google/gemini-2.5-flash" +# Vision path: the LLM reads the actual page image. Defaults follow the repo's +# "google/..." OpenRouter model-id convention (see config.py's validated list); +# override via env without a code change. +VISION_MODEL = os.getenv("COPY_CHECK_VISION_MODEL", "google/gemini-2.5-flash") +VISION_ESCALATION_MODEL = os.getenv( + "COPY_CHECK_VISION_ESCALATION_MODEL", "google/gemini-2.5-pro" +) ESCALATION_CONF_THRESHOLD = 0.60 MAX_ESCALATIONS_PER_COPY = 2 # Budget tuned for typical 8-question copies. Each grading call re-sends the @@ -73,6 +83,10 @@ def __init__( self._prompt_tokens = 0 self._completion_tokens = 0 self._escalations_used = 0 + # Vision-image accounting: how many page images we've attached across all + # grading calls for this copy, and the per-copy cap that bounds cost. + self._images_used = 0 + self._max_images_per_copy = get_settings().copy_check_vision_max_images_per_copy def add_tokens(self, n: int) -> None: """External counter for non-grading calls (e.g. criteria generation @@ -119,6 +133,10 @@ def add_usage(self, usage: dict[str, Any]) -> None: def tokens_used(self) -> int: return self._tokens_used + @property + def images_used(self) -> int: + return self._images_used + @property def prompt_tokens(self) -> int: return self._prompt_tokens @@ -133,9 +151,21 @@ async def grade_question( rubric: dict[str, Any], layout_map: dict[str, Any], preferred_model: Optional[str] = None, + page_images: Optional[list[str]] = None, ) -> dict[str, Any]: - model = preferred_model or DEFAULT_MODEL - verdict = await self._call(question, rubric, layout_map, model) + """Grade one question. + + `page_images`: base64 image data URLs of the page(s) this question's + answer occupies. When present, the LLM reads the handwriting from the + image and the OCR transcript is demoted to an assistive hint (vision + path). When None/empty, the legacy text-only path runs unchanged. + """ + has_images = bool(page_images) + # Teacher's explicit model pick always wins; otherwise choose the vision + # default when images are attached, else the cheap text default. + model = preferred_model or (VISION_MODEL if has_images else DEFAULT_MODEL) + escalation_model = VISION_ESCALATION_MODEL if has_images else ESCALATION_MODEL + verdict = await self._call(question, rubric, layout_map, model, page_images) if ( float(verdict.get("confidence", 0)) < ESCALATION_CONF_THRESHOLD and self._escalations_used < MAX_ESCALATIONS_PER_COPY @@ -143,10 +173,12 @@ async def grade_question( self._escalations_used += 1 logger.info( "Escalating Q%s to %s (conf=%.2f)", - question["question_id"], ESCALATION_MODEL, verdict.get("confidence", 0), + question["question_id"], escalation_model, verdict.get("confidence", 0), ) try: - verdict = await self._call(question, rubric, layout_map, ESCALATION_MODEL) + verdict = await self._call( + question, rubric, layout_map, escalation_model, page_images, + ) except Exception as e: logger.warning(f"Escalation failed, keeping initial verdict: {e}") return verdict @@ -157,15 +189,42 @@ async def _call( rubric: dict[str, Any], layout_map: dict[str, Any], model: str, + page_images: Optional[list[str]] = None, ) -> dict[str, Any]: if self._tokens_used >= FAIL_TOKENS_PER_COPY: raise RuntimeError( f"copy-check token budget exhausted: {self._tokens_used} >= {FAIL_TOKENS_PER_COPY}" ) - prompt = build_grading_prompt(question, rubric, layout_map) + # Enforce the per-copy image budget: trim (or drop) this call's images if + # we're at/over the cap, so a big copy can't run up unbounded vision cost. + imgs = list(page_images or []) + if imgs: + remaining = self._max_images_per_copy - self._images_used + if remaining <= 0: + logger.warning( + "copy-check image budget exhausted (%d/%d sent); grading Q%s text-only", + self._images_used, self._max_images_per_copy, question.get("question_id"), + ) + imgs = [] + elif len(imgs) > remaining: + logger.info( + "copy-check image budget: trimming Q%s from %d to %d page image(s)", + question.get("question_id"), len(imgs), remaining, + ) + imgs = imgs[:remaining] + vision = bool(imgs) + system = GRADING_SYSTEM_VISION if vision else GRADING_SYSTEM + prompt = build_grading_prompt(question, rubric, layout_map, vision=vision) + user_msg: dict[str, Any] = {"role": "user", "content": prompt} + if vision: + # Shape expected by ChatLLMClient._convert_to_multimodal_messages: + # a user message with an `attachments` list of {"type","url"} dicts; + # the client turns each into an OpenAI "image_url" content part. + user_msg["attachments"] = [{"type": "image", "url": u} for u in imgs] + self._images_used += len(imgs) messages = [ - {"role": "system", "content": GRADING_SYSTEM}, - {"role": "user", "content": prompt}, + {"role": "system", "content": system}, + user_msg, ] try: response = await self.llm.chat_completion( diff --git a/ai_service/app/services/copy_check/mathpix_fallback.py b/ai_service/app/services/copy_check/mathpix_fallback.py index b5125bc108..5c6ff5450c 100644 --- a/ai_service/app/services/copy_check/mathpix_fallback.py +++ b/ai_service/app/services/copy_check/mathpix_fallback.py @@ -14,9 +14,8 @@ from pathlib import Path from typing import Any -import httpx - from ..mathpix_service import MathpixService +from .page_images import download_pdf, rasterize_pages logger = logging.getLogger(__name__) @@ -51,10 +50,13 @@ async def enrich_layout_for_math(self, pdf_url: str, layout_map: dict[str, Any]) with tempfile.TemporaryDirectory(prefix="mathpix-crops-") as tmp: pdf_path = Path(tmp) / "input.pdf" - await _download(pdf_url, pdf_path) - page_imgs = await asyncio.get_event_loop().run_in_executor( - None, _rasterize_pages, pdf_path, - ) + await download_pdf(pdf_url, pdf_path) + # 200 DPI matches the layout_map's box coordinates (full_res px), so + # crops line up. rasterize_pages returns [(page_id, img)]; keyed by + # page_id here for the crop lookup below. + page_imgs = dict(await asyncio.get_event_loop().run_in_executor( + None, rasterize_pages, pdf_path, + )) for page, line in flagged: if not self.can_run: logger.info("Mathpix budget exhausted (%d crops), skipping rest", self._used) @@ -76,47 +78,6 @@ async def enrich_layout_for_math(self, pdf_url: str, layout_map: dict[str, Any]) return layout_map -async def _download(url: str, dest: Path) -> None: - async with httpx.AsyncClient(timeout=60, follow_redirects=True) as client: - async with client.stream("GET", url) as resp: - resp.raise_for_status() - with dest.open("wb") as f: - async for chunk in resp.aiter_bytes(chunk_size=1 << 16): - f.write(chunk) - - -def _rasterize_pages(pdf_path: Path) -> dict[str, Any]: - """Return {page_id: PIL.Image} for all pages of the PDF at 200 DPI. - - Handles all PyMuPDF colorspaces (gray/RGB/CMYK) since PDFs in the wild - aren't always sRGB. A naive Image.frombytes("RGB", …) corrupts the buffer - for pix.n != 3. - """ - import fitz # PyMuPDF - from PIL import Image - - out: dict[str, Any] = {} - doc = fitz.open(pdf_path) - try: - matrix = fitz.Matrix(200 / 72.0, 200 / 72.0) - for i, page in enumerate(doc): - pix = page.get_pixmap(matrix=matrix, alpha=False) - mode_map = {1: "L", 3: "RGB", 4: "CMYK"} - mode = mode_map.get(pix.n) - if mode is None: - # Drop alpha or unknown channels via an intermediate RGB pixmap. - pix_rgb = fitz.Pixmap(fitz.csRGB, pix) - img = Image.frombytes("RGB", (pix_rgb.width, pix_rgb.height), pix_rgb.samples) - else: - img = Image.frombytes(mode, (pix.width, pix.height), pix.samples) - if mode != "RGB": - img = img.convert("RGB") - out[f"p{i + 1}"] = img - finally: - doc.close() - return out - - def _crop_to_base64(img, box: list[int]) -> str: from PIL import Image # noqa: F401 — typed via duck typing diff --git a/ai_service/app/services/copy_check/orchestrator.py b/ai_service/app/services/copy_check/orchestrator.py index 9ada052f3d..6232d780b5 100644 --- a/ai_service/app/services/copy_check/orchestrator.py +++ b/ai_service/app/services/copy_check/orchestrator.py @@ -15,14 +15,21 @@ from sqlalchemy.orm import Session +from ...config import get_settings from ...models.ai_token_usage import RequestType from ..ai_billing import record_tool_billing from ..api_key_resolver import ApiKeyResolver from ..chat_llm_client import ChatLLMClient from ...repositories.copy_check_rubric_repository import CopyCheckRubricRepository from . import annotator, callbacks, cancellation -from .grader import DEFAULT_MODEL, CopyCheckGrader, call_llm_for_criteria +from .grader import DEFAULT_MODEL, VISION_MODEL, CopyCheckGrader, call_llm_for_criteria from .mathpix_fallback import MathpixFallback +from .page_images import ( + PageImage, + ordered_page_ids, + render_page_images, + select_pages_for_question, +) from .render_client import CopyCheckRenderClient, OcrCancelled from .rubric import RubricResolver, load_snapshot from .validator import validate_and_cap @@ -34,6 +41,30 @@ def _new_job_id() -> str: return str(uuid.uuid4()) +def _manual_review_verdict( + question: dict[str, Any], question_number: int, reason: str, +) -> dict[str, Any]: + """Verdict that routes a question to manual review instead of releasing a + (likely wrong) zero. Used when vision grading is required but the page + image is unavailable — grading handwriting from OCR alone is the exact + accuracy bug this pipeline change removes, so we never silently fall back + to it. Same shape as the retry-failure verdict the orchestrator already + emits, so the callback contract to Java is unchanged.""" + return { + "question_id": question["question_id"], + "question_number": question_number, + "marks_awarded": 0.0, + "max_marks": float(question.get("max_marks") or 0), + "extracted_answer": "", + "feedback": "This answer could not be evaluated automatically and needs manual review.", + "confidence": 0.0, + "criteria_breakdown": [], + "annotations": [], + "status": "FAILED", + "error_detail": f"vision grading unavailable: {reason}"[:500], + } + + def _render_client() -> CopyCheckRenderClient: # ai-service uses RENDER_SERVER_URL/RENDER_SERVER_KEY as the cluster-wide # convention (already set on the deployment). Fall through to the names @@ -145,18 +176,93 @@ async def _llm_for_criteria(system: str, user: str, model: str | None) -> dict[s cancellation.check(job_id, process_id) layout_map = await mathpix.enrich_layout_for_math(pdf_url, layout_map) + # 2b. VISION: render each PDF page to a base64 image ONCE for the whole + # copy so the grader can send the actual handwriting (not just its OCR). + # Guarded by COPY_CHECK_VISION_GRADING; when off, page_images stays empty + # and grading runs the legacy text-only path unchanged. + settings = get_settings() + vision_enabled = bool(settings.copy_check_vision_grading) + page_url_by_id: dict[str, str] = {} + rendered_page_ids: list[str] = [] + vision_render_error: Optional[str] = None + if vision_enabled: + cancellation.check(job_id, process_id) + try: + page_imgs: list[PageImage] = await render_page_images( + pdf_url, + dpi=settings.copy_check_vision_dpi, + max_px=settings.copy_check_vision_max_image_px, + max_pages=settings.copy_check_vision_max_pages_per_copy, + ) + if not page_imgs: + vision_render_error = "no pages rendered from the answer sheet" + else: + page_url_by_id = {p.page_id: p.data_url for p in page_imgs} + # Prefer the layout_map's page ordering (page_index) so image + # selection and annotation agree; fall back to render order. + rendered_page_ids = [ + pid for pid in ordered_page_ids(layout_map) if pid in page_url_by_id + ] or [p.page_id for p in page_imgs] + except cancellation.Cancelled: + raise + except Exception as e: + logger.exception("copy-check vision page rendering failed") + vision_render_error = str(e) or type(e).__name__ + # 3. Per-question grading. total_awarded = 0.0 total_max = 0.0 evaluated = 0 + max_pages_per_q = settings.copy_check_vision_max_pages_per_question # Kept so the annotator can draw every verdict in one pass at the end; # the per-question callback already fired for each of these. verdicts: list[dict[str, Any]] = [] - for q in questions: + for ordinal, q in enumerate(questions, 1): cancellation.check(job_id, process_id) + + # Resolve this question's page image(s) for the vision path. + q_page_images: Optional[list[str]] = None + if vision_enabled: + if vision_render_error is not None: + # Required image is missing — route to manual review rather + # than grading handwriting from OCR and releasing a wrong zero. + verdict = _manual_review_verdict(q, ordinal, vision_render_error) + total_awarded += verdict["marks_awarded"] + total_max += verdict["max_marks"] + evaluated += 1 + verdicts.append(verdict) + await callbacks.question_done( + callback_base, process_id, job_id, verdict, rubric_version=rubric_version, + ) + continue + selected_ids, reason = select_pages_for_question( + ordinal, layout_map, rendered_page_ids, max_pages_per_q, + ) + q_page_images = [ + page_url_by_id[pid] for pid in selected_ids if pid in page_url_by_id + ] + if not q_page_images: + verdict = _manual_review_verdict( + q, ordinal, "no page image available for this question", + ) + total_awarded += verdict["marks_awarded"] + total_max += verdict["max_marks"] + evaluated += 1 + verdicts.append(verdict) + await callbacks.question_done( + callback_base, process_id, job_id, verdict, rubric_version=rubric_version, + ) + continue + logger.info( + "Q%s vision: %d page image(s) [%s] via %s", + q.get("question_id"), len(q_page_images), ",".join(selected_ids), reason, + ) + try: rubric = await rubric_resolver.resolve(q, preferred_model) - raw = await grader.grade_question(q, rubric, layout_map, preferred_model) + raw = await grader.grade_question( + q, rubric, layout_map, preferred_model, page_images=q_page_images, + ) verdict = validate_and_cap(raw, q, layout_map) except cancellation.Cancelled: raise @@ -167,12 +273,15 @@ async def _llm_for_criteria(system: str, user: str, model: str | None) -> dict[s # cap being hit mid-batch — both of which a single retry # with a clean state often clears. Without this, a single # failure mid-batch silently steals marks from the student. + retry_model = VISION_MODEL if q_page_images else DEFAULT_MODEL logger.warning( - f"Grading failed for question {q.get('question_id')}: {e}; retrying once with {DEFAULT_MODEL}", + f"Grading failed for question {q.get('question_id')}: {e}; retrying once with {retry_model}", ) try: - rubric = await rubric_resolver.resolve(q, DEFAULT_MODEL) - raw = await grader.grade_question(q, rubric, layout_map, DEFAULT_MODEL) + rubric = await rubric_resolver.resolve(q, retry_model) + raw = await grader.grade_question( + q, rubric, layout_map, retry_model, page_images=q_page_images, + ) verdict = validate_and_cap(raw, q, layout_map) except cancellation.Cancelled: raise @@ -203,7 +312,7 @@ async def _llm_for_criteria(system: str, user: str, model: str | None) -> dict[s total_awarded += verdict["marks_awarded"] total_max += verdict["max_marks"] evaluated += 1 - verdict.setdefault("question_number", q.get("question_number") or evaluated) + verdict.setdefault("question_number", q.get("question_number") or ordinal) verdicts.append(verdict) await callbacks.question_done( callback_base, process_id, job_id, verdict, rubric_version=rubric_version, @@ -229,8 +338,10 @@ async def _llm_for_criteria(system: str, user: str, model: str | None) -> dict[s evaluated_file_id=evaluated_file_id, ) logger.info( - "copy-check job %s complete: %s/%s, %d Mathpix crops used, %d tokens", - job_id, total_awarded, total_max, mathpix.used, grader.tokens_used, + "copy-check job %s complete: %s/%s, %d Mathpix crops used, " + "%d page images sent (vision=%s), %d tokens", + job_id, total_awarded, total_max, mathpix.used, + grader.images_used, vision_enabled, grader.tokens_used, ) # Meter the copy: charge the institute's credits once per completed diff --git a/ai_service/app/services/copy_check/page_images.py b/ai_service/app/services/copy_check/page_images.py new file mode 100644 index 0000000000..9c0739129b --- /dev/null +++ b/ai_service/app/services/copy_check/page_images.py @@ -0,0 +1,228 @@ +"""Rasterize a copy's PDF pages to base64 image data URLs for VISION grading, +and pick which page(s) to send per question. + +Shares the PyMuPDF rasterization core with mathpix_fallback (both turn the +student's PDF into per-page raster images); factored here so the vision grader +and the selective math fallback don't each reimplement fitz page rendering. + +fitz/PIL are imported lazily inside the worker functions so this module — and +anything importing it — stays loadable on hosts where PyMuPDF isn't installed, +matching mathpix_fallback's pattern. +""" +from __future__ import annotations + +import asyncio +import base64 +import io +import logging +import re +import tempfile +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Optional + +import httpx + +logger = logging.getLogger(__name__) + + +@dataclass +class PageImage: + """One rendered PDF page, ready to attach to a multimodal LLM message.""" + + page_id: str # render_worker's 1-based id, e.g. "p1" + page_index: int # 0-based order in the PDF + data_url: str # "data:image/jpeg;base64,..." + + +# --------------------------- Rendering (shared) ------------------------------ + +async def download_pdf(url: str, dest: Path) -> None: + """Stream a PDF to disk. Shared by mathpix_fallback and vision rendering.""" + async with httpx.AsyncClient(timeout=60, follow_redirects=True) as client: + async with client.stream("GET", url) as resp: + resp.raise_for_status() + with dest.open("wb") as f: + async for chunk in resp.aiter_bytes(chunk_size=1 << 16): + f.write(chunk) + + +def rasterize_pages(pdf_path: Path, dpi: int = 200) -> list[tuple[str, Any]]: + """Return ordered [(page_id, PIL.Image)] for every page at the given DPI. + + page_id is render_worker's 1-based "p{i+1}" so callers can key by it. Handles + all PyMuPDF colorspaces (gray/RGB/CMYK) since PDFs in the wild aren't always + sRGB — a naive Image.frombytes("RGB", ...) corrupts the buffer for pix.n != 3. + """ + import fitz # PyMuPDF (lazy: keeps this module importable without it) + from PIL import Image + + out: list[tuple[str, Any]] = [] + doc = fitz.open(pdf_path) + try: + matrix = fitz.Matrix(dpi / 72.0, dpi / 72.0) + for i, page in enumerate(doc): + pix = page.get_pixmap(matrix=matrix, alpha=False) + mode_map = {1: "L", 3: "RGB", 4: "CMYK"} + mode = mode_map.get(pix.n) + if mode is None: + # Drop alpha or unknown channels via an intermediate RGB pixmap. + pix_rgb = fitz.Pixmap(fitz.csRGB, pix) + img = Image.frombytes("RGB", (pix_rgb.width, pix_rgb.height), pix_rgb.samples) + else: + img = Image.frombytes(mode, (pix.width, pix.height), pix.samples) + if mode != "RGB": + img = img.convert("RGB") + out.append((f"p{i + 1}", img)) + finally: + doc.close() + return out + + +def image_to_data_url(img: Any, max_px: int = 2000, jpeg_quality: int = 80) -> str: + """Downscale so the longest side <= max_px, then JPEG-encode as a data URL. + + JPEG (not PNG) keeps a multi-page handwritten payload small enough to send + several pages in one request without blowing the token/size budget. The + "data:image/jpeg;base64,..." shape is what OpenRouter / the multimodal + client expects for an inline image. + """ + from PIL import Image + + w, h = img.width, img.height + longest = max(w, h) + if longest > max_px: + scale = max_px / float(longest) + img = img.resize((max(1, round(w * scale)), max(1, round(h * scale))), Image.LANCZOS) + if img.mode != "RGB": + img = img.convert("RGB") + buf = io.BytesIO() + img.save(buf, format="JPEG", quality=jpeg_quality, optimize=True) + return "data:image/jpeg;base64," + base64.b64encode(buf.getvalue()).decode("ascii") + + +def _render_sync( + pdf_path: Path, dpi: int, max_px: int, jpeg_quality: int, max_pages: int, +) -> list[PageImage]: + pages = rasterize_pages(pdf_path, dpi=dpi) + if len(pages) > max_pages: + logger.warning( + "copy-check vision: PDF has %d pages > cap %d; encoding only the first %d", + len(pages), max_pages, max_pages, + ) + pages = pages[:max_pages] + return [ + PageImage( + page_id=pid, + page_index=i, + data_url=image_to_data_url(img, max_px=max_px, jpeg_quality=jpeg_quality), + ) + for i, (pid, img) in enumerate(pages) + ] + + +async def render_page_images( + pdf_url: str, + dpi: int = 150, + max_px: int = 2000, + jpeg_quality: int = 80, + max_pages: int = 50, +) -> list[PageImage]: + """Download `pdf_url` and rasterize every page to a JPEG data URL, once. + + Runs the CPU-bound rasterize+encode in a thread (like mathpix_fallback) so + the event loop isn't blocked. `max_pages` caps pathological PDFs so a runaway + document can't exhaust memory or the image budget. + """ + with tempfile.TemporaryDirectory(prefix="vision-pages-") as tmp: + pdf_path = Path(tmp) / "input.pdf" + await download_pdf(pdf_url, pdf_path) + loop = asyncio.get_event_loop() + return await loop.run_in_executor( + None, _render_sync, pdf_path, dpi, max_px, jpeg_quality, max_pages, + ) + + +# --------------------------- Per-question page mapping ----------------------- + +# Reading-order tokens that commonly precede an answer's question number on a +# handwritten copy: "Q1", "Ans 1.", "Answer 1)", "(1)", "1.", "1-". +_ANSWER_MARKER_PREFIX = r"(?:q(?:ues(?:tion)?)?\.?\s*|ans(?:wer)?\.?\s*)?" + + +def _answer_marker_regex(number: int) -> "re.Pattern[str]": + n = re.escape(str(int(number))) + # Number (optionally zero-padded / bracketed) followed by a delimiter, + # anchored to the start of a line so a stray "1" mid-sentence never matches. + return re.compile( + rf"^\s*{_ANSWER_MARKER_PREFIX}[\(\[]?0*{n}[\)\].:\-]", + re.IGNORECASE, + ) + + +def ordered_page_ids(layout_map: dict[str, Any]) -> list[str]: + """page_ids in PDF order (page_index when present, else declaration order), + mirroring annotator._page_order so image selection and annotation agree.""" + pages = list(enumerate(layout_map.get("pages") or [])) + + def sort_key(item: tuple[int, dict[str, Any]]) -> int: + i, page = item + idx = page.get("page_index") + return int(idx) if isinstance(idx, int) and idx >= 0 else i + + return [p.get("page_id") for _, p in sorted(pages, key=sort_key) if p.get("page_id")] + + +def _anchor_page_pos( + layout_map: dict[str, Any], page_ids: list[str], number: int, +) -> Optional[int]: + """Position (index into `page_ids`) of the first line that reads like the + answer marker for `number`, else None.""" + rx = _answer_marker_regex(number) + pos_by_id = {pid: i for i, pid in enumerate(page_ids)} + best: Optional[int] = None + for page in layout_map.get("pages") or []: + pos = pos_by_id.get(page.get("page_id")) + if pos is None: + continue + for line in page.get("lines") or []: + if rx.match((line.get("text") or "").strip()): + if best is None or pos < best: + best = pos + break + return best + + +def select_pages_for_question( + question_number: Optional[int], + layout_map: dict[str, Any], + page_ids: list[str], + max_pages: int, +) -> tuple[list[str], str]: + """Pick the page_id(s) whose images to send for this question. + + Returns (selected_page_ids, reason). Strategy: + 1. If the whole copy fits the per-question budget, send all pages — no + segmentation risk and the model sees the complete answer. + 2. Otherwise localize the answer via its question-number marker in the OCR + and send that page through the page before the next question's marker, + capped to `max_pages`. + 3. If it can't be localized on a large copy, send the first `max_pages` + pages as a bounded fallback (uncertain — the low-confidence verdict + then escalates, and the OCR transcript still covers the rest). + """ + n = len(page_ids) + if n == 0: + return [], "no_pages" + if n <= max_pages: + return list(page_ids), "all_pages_within_budget" + + if question_number is not None: + start = _anchor_page_pos(layout_map, page_ids, question_number) + if start is not None: + end = _anchor_page_pos(layout_map, page_ids, question_number + 1) + stop = end if (end is not None and end > start) else n + window = page_ids[start:stop] or page_ids[start:start + 1] + return window[:max_pages], "anchored" + + return list(page_ids[:max_pages]), "uncertain_bounded" diff --git a/ai_service/app/services/copy_check/prompt_builder.py b/ai_service/app/services/copy_check/prompt_builder.py index e2297565dd..cf7a32826d 100644 --- a/ai_service/app/services/copy_check/prompt_builder.py +++ b/ai_service/app/services/copy_check/prompt_builder.py @@ -50,12 +50,31 @@ def build_criteria_prompt( # ---------------------------- Grading prompt --------------------------------- -GRADING_SYSTEM = ( +_GRADING_INTRO_TEXT = ( "You are an expert evaluator. Grade the student's handwritten answer based " "strictly on the provided rubric. The student's pages have been OCR'd into " "a numbered transcript of line_ids — when you flag an error or correctness, " "you MUST reference the line_id (e.g. \"L1_32\"), never pixel coordinates. " - "Ignore OCR/spelling errors; focus on intent and meaning.\n\n" + "Ignore OCR/spelling errors; focus on intent and meaning." +) + +# Vision path: the real page images are attached. The OCR is demoted to an +# assistive hint so the model reads the handwriting itself instead of grading a +# printed-text OCR of handwriting (the accuracy bug this pipeline change fixes). +_GRADING_INTRO_VISION = ( + "You are an expert evaluator. The student's ACTUAL handwritten answer pages " + "are ATTACHED AS IMAGES — read the handwriting directly from the image(s); " + "the image is the source of truth. A best-effort OCR transcript is also " + "provided as a numbered list of line_ids, but it is ASSISTIVE ONLY and may " + "contain recognition errors: use it only (a) as a hint when the handwriting " + "is hard to read and (b) to choose the line_id to anchor each annotation to " + "— never grade from the OCR alone. When you flag an error or correctness, " + "you MUST reference the line_id (e.g. \"L1_32\"), never pixel coordinates. " + "Grade against the rubric and award partial credit. Ignore spelling/OCR " + "errors; focus on intent and meaning." +) + +_ANNOTATION_DISCIPLINE = ( "ANNOTATION DISCIPLINE (these rules are non-negotiable — teachers rely on " "them to audit your grading):\n" "1. WRITE LIKE A TEACHER'S PEN. Annotation `text` is written ON the copy, " @@ -99,6 +118,11 @@ def build_criteria_prompt( "conclusion line." ) +# Text-only path (legacy) and vision path share the same annotation discipline; +# only the intro differs in where the model is told to read the answer from. +GRADING_SYSTEM = _GRADING_INTRO_TEXT + "\n\n" + _ANNOTATION_DISCIPLINE +GRADING_SYSTEM_VISION = _GRADING_INTRO_VISION + "\n\n" + _ANNOTATION_DISCIPLINE + def _transcript_for_prompt(layout_map: dict[str, Any]) -> str: parts: list[str] = [] @@ -185,12 +209,28 @@ def build_grading_prompt( question: dict[str, Any], rubric: dict[str, Any], layout_map: dict[str, Any], + vision: bool = False, ) -> str: max_marks = float(rubric.get("max_marks") or question.get("max_marks") or 10) rubric_json = json.dumps(rubric, indent=2) + if vision: + source_instruction = ( + "The student's actual answer page(s) are ATTACHED AS IMAGE(S). READ " + "THE HANDWRITING FROM THE IMAGE(S) — that is the source of truth. The " + "OCR transcript below is ASSISTIVE ONLY (it may misread handwriting); " + "use it only as a hint and to pick the line_ids for your annotations, " + "never as the sole basis for the grade.\n\n" + ) + transcript_header = ( + "**Assistive OCR transcript (line_id + text per page — MAY CONTAIN " + "ERRORS; rely on the attached image, not this text):**" + ) + else: + source_instruction = "" + transcript_header = "**Student's OCR'd transcript (line_id + text per page):**" return f"""Grade the student's handwritten answer. -**Question ID:** {question['question_id']} +{source_instruction}**Question ID:** {question['question_id']} **Question type:** {question.get('question_type')} **Question:** {question['question_text']} @@ -201,7 +241,7 @@ def build_grading_prompt( **Evaluation rubric (JSON):** {rubric_json} -**Student's OCR'd transcript (line_id + text per page):** +{transcript_header} {_transcript_for_prompt(layout_map)} **Type-specific instructions:**