Hemprasad Badgujar
Add bm25 prefilter stage; unify candidates cache; validate reasoning quality
cd5f9d4 | """Preprocess phase orchestrator — the single source of truth for sub-stage order. | |
| The phase is modeled as: load the pool, then fold a list of small frame→frame ``Stage``s over it. Exposing | |
| the stages as data (``preprocess_stages``) means the orchestrator, the CLI, and the Streamlit explorer all | |
| drive the SAME sequence — so each sub-stage can be run, previewed, and debugged independently. New stages | |
| (enrich → materialize → expand) land by appending one line below. | |
| The expensive part (parsing the 465MB JSONL) is cached inside ``load_candidates`` itself, not here — see its | |
| docstring. validate/materialize/quality always run fresh on every call: they're cheap (~2-3s combined) and | |
| ``quality`` APPENDS to the ``text`` column, so caching+replaying them would double-append and corrupt it. | |
| """ | |
| from __future__ import annotations | |
| from collections.abc import Callable | |
| from dataclasses import dataclass | |
| from datetime import date | |
| from pathlib import Path | |
| import polars as pl | |
| from common.checkpoint import checkpoint | |
| from common.logging import get_logger | |
| from config import Settings, load_settings | |
| from .load import load_candidates | |
| from .materialize import materialize | |
| from .quality import quality | |
| from .validate import validate | |
| # Fixed reference "now" all recency/age features measure against. Recency features are relative deltas | |
| # (``as_of - date``), so a fixed anchor shifts every candidate uniformly — it keeps the pass deterministic | |
| # and avoids an O(n) scan for the data's max date. It sits at/after the dataset horizon so deltas stay ≥ 0. | |
| DATASET_AS_OF: date = date(2026, 6, 6) | |
| class Stage: | |
| """One preprocess sub-stage: a label + a frame→frame transform (each logs its own ``step()``).""" | |
| name: str | |
| run: Callable[[pl.DataFrame], pl.DataFrame] | |
| def preprocess_stages(as_of: date) -> list[Stage]: | |
| """The ordered frame→frame sub-stages applied AFTER the initial load. Order is the contract.""" | |
| return [ | |
| Stage("validate", lambda frame: validate(frame, as_of)), | |
| Stage("materialize", lambda frame: materialize(frame, as_of)), | |
| Stage("quality", quality), | |
| # enrich + expand_skills intentionally omitted (raw flat_skills; experience_band computed inline). | |
| ] | |
| def run_preprocess( | |
| candidates: str | Path, limit: int | None = None, settings: Settings | None = None | |
| ) -> tuple[pl.DataFrame, date]: | |
| """Run the full preprocess phase: load → validate → … → returns (frame, reference date). | |
| Validation runs on raw input (before enrichment) so flags reflect what the candidate submitted. | |
| Each phase's output is written to a Parquet checkpoint when ``settings.debug.checkpoints`` is on. | |
| ``load_candidates`` transparently serves from its own Parquet cache for the real, unlimited run — see its | |
| docstring; this orchestrator doesn't need to know that. | |
| """ | |
| active = settings or load_settings() | |
| as_of = DATASET_AS_OF | |
| frame = load_candidates(candidates, limit=limit) | |
| checkpoint(0, "load", frame, active) | |
| for index, stage in enumerate(preprocess_stages(as_of), start=1): | |
| frame = stage.run(frame) | |
| checkpoint(index, stage.name, frame, active) | |
| get_logger("preprocess").info( | |
| "preprocess.done", rows=frame.height, columns=frame.width, as_of=str(as_of) | |
| ) | |
| return frame, as_of | |