| """Step 11 multi-band explainability — bands, FactEngine matchers, NLG composition, tier gap-gating, artifact.""" |
|
|
| from __future__ import annotations |
|
|
| from pathlib import Path |
|
|
| import orjson |
| import polars as pl |
|
|
| from config import ReasoningSettings, Settings |
| from jd import JDSpec |
| from rank import add_explanation |
| from rank.explain import _clean_note, _is_grounded, _matched_skills, _title_match |
| from rank.explain.bands import experience_fit, location_tier, notice_band, recency_band, skill_coverage |
| from rank.explain.engine import matched_education, matched_industries, title_fit |
| from rank.explain.signals import signal_bands |
|
|
| _SPEC = JDSpec( |
| must_have_skills=["Python", "embeddings", "ranking"], nice_to_have=["RAG", "HNSW"], |
| job_titles=["AI Engineer", "Search Engineer"], locations_preferred=["Pune", "Noida"], |
| locations_acceptable=["Bangalore"], years_min=5, years_max=9, |
| target_industries=["Fintech", "Healthcare"], education_fields=["Computer Science"], |
| ) |
|
|
|
|
| def _row(**over: object) -> dict[str, object]: |
| base = { |
| "candidate_id": "C1", "current_title": "Senior AI Engineer", "years_of_experience": 7.4, |
| "experience_band": "senior", "location": "Pune, MH", "country": "India", |
| "current_industry": "Fintech", "current_company": "Acme", |
| "flat_skills": ["Python", "embeddings", "ranking"], "flat_career_industries": ["Fintech"], |
| "flat_certifications": [], "education": [{"field_of_study": "Computer Science"}], |
| "days_since_active": 10, "notice_period_days": 15, "is_available_soon": True, |
| "open_to_work_flag": True, "willing_to_relocate": False, "preferred_work_mode": "hybrid", |
| "verification_count": 3, "recruiter_response_rate": 0.8, "saved_by_recruiters_30d": 6, |
| "profile_views_received_30d": 25, "interview_completion_rate": 0.9, "profile_completeness_score": 85, |
| "has_github": True, "github_score_clean": 70.0, "avg_skill_assessment": 75.0, "verified_skills": ["Python"], |
| "bm25_hit_skills": True, "bm25_hit_titles": True, "bm25_hit_education": True, |
| "bm25_hit_locations": True, "bm25_hit_industries": True, |
| "potion_score": 0.9, "soft_score": 0.8, "bm25_score": 0.7, "composite_score": 0.85, |
| "granite_score": 0.9, "reranker_score": 1.9, "candidate_quality": 0.7, "final_score": 1.0, |
| } |
| base.update(over) |
| return base |
|
|
|
|
| def _frame(rows: list[dict[str, object]]) -> pl.DataFrame: |
| return pl.DataFrame(rows) |
|
|
|
|
| def _settings(tmp_path: Path, **reasoning: object) -> Settings: |
| opts = {"mode": "template", "variant_select": "random", |
| "explanations_path": str(tmp_path / "explanations.json")} |
| opts.update(reasoning) |
| return Settings(reasoning=ReasoningSettings(**opts)) |
|
|
|
|
| |
| def test_band_mappers() -> None: |
| assert experience_fit(7.0, 5, 9) == "in" and experience_fit(3.0, 5, 9) == "below" |
| assert experience_fit(11.0, 5, 9) == "above" |
| assert skill_coverage(3, 3) == "strong" and skill_coverage(1, 3) == "partial" and skill_coverage(0, 3) == "thin" |
| assert location_tier("Pune", ["Pune"], ["Bangalore"]) == "preferred" |
| assert location_tier("Bangalore", ["Pune"], ["Bangalore"]) == "acceptable" |
| assert location_tier("Berlin", ["Pune"], ["Bangalore"]) == "outside" |
| assert recency_band(10, 90) == "month" and recency_band(60, 90) == "quarter" and recency_band(400, 90) == "stale" |
| assert notice_band(15, False) == "short" and notice_band(90, False) == "long" |
|
|
|
|
| def test_fact_matchers() -> None: |
| assert _matched_skills(["Python", "Go"], ["python", "sql"]) == ["Python"] |
| assert _title_match(["AI Engineer"], "Senior AI Engineer") == "AI Engineer" |
| assert title_fit(["AI Engineer"], "Senior AI Engineer", True) == ("AI Engineer", "align") |
| |
| assert title_fit(["AI Engineer"], "Accountant", True) == ("AI Engineer", "mismatch") |
| assert matched_industries(["Fintech"], "FinTech Solutions", ["SaaS"]) == ["Fintech"] |
| assert matched_education(["Computer Science"], [{"field_of_study": "Computer Science"}]) == ["Computer Science"] |
|
|
|
|
| def test_signal_bands_sentinel_and_gate() -> None: |
| |
| sig = signal_bands(_row(has_github=False), recency="month", location_outside=False) |
| assert sig["github"] is None |
| assert sig["active_market"] is True and sig["reachability"] == "full" |
| |
| passive = signal_bands(_row(open_to_work_flag=False, recruiter_response_rate=0.2, saved_by_recruiters_30d=0, |
| profile_views_received_30d=0), recency="stale", location_outside=False) |
| assert passive["active_market"] is False |
|
|
|
|
| |
| def test_grounded_multisentence_reasoning(tmp_path: Path) -> None: |
| out = add_explanation(_frame([_row()]), _SPEC, _settings(tmp_path)) |
| top = out.row(0, named=True) |
| assert "Python" in top["reasoning"] and "Pune" in top["reasoning"] |
| assert top["reasoning"].count(".") >= 3 |
| assert "AI Engineer" in top["explanation"] |
|
|
|
|
| def test_t1_has_no_gap_but_explanation_does(tmp_path: Path) -> None: |
| |
| row = _row(flat_skills=["Python"], location="Berlin, DE", country="Germany") |
| out = add_explanation(_frame([row]), _SPEC, _settings(tmp_path, tier1_top=1)) |
| top = out.row(0, named=True) |
| assert "though" not in top["reasoning"].lower() and "outside" not in top["reasoning"].lower() |
| assert "Watch-outs:" in top["explanation"] |
|
|
|
|
| def test_t3_surfaces_one_soft_gap(tmp_path: Path) -> None: |
| |
| rows = [_row(candidate_id="A"), _row(candidate_id="B", flat_skills=["Python"], location="Berlin, DE", |
| days_since_active=400, reranker_score=0.1, final_score=0.1)] |
| out = add_explanation(_frame(rows), _SPEC, _settings(tmp_path, tier1_top=1, tier2_top=1, tier3_top=2)) |
| b = out.filter(pl.col("candidate_id") == "B").row(0, named=True) |
| pivots = ("though", "albeit", "caveat", "watch-out", "however", "main gap") |
| hits = sum(b["reasoning"].lower().count(p) for p in pivots) |
| assert hits == 1 |
|
|
|
|
| |
| def test_clean_note_salvages_and_rejects() -> None: |
| assert _clean_note("Sure! Here it is: A senior engineer with deep retrieval experience.", "fb") \ |
| == "A senior engineer with deep retrieval experience." |
| assert _clean_note("Recruiter Note: /no_think", "fallback line here") == "fallback line here" |
|
|
|
|
| def test_is_grounded() -> None: |
| source = "Senior AI Engineer; strong on embeddings, Pinecone, Weaviate; based in Pune." |
| assert _is_grounded("Senior AI Engineer strong on embeddings and Pinecone, based in Pune.", source) |
| assert not _is_grounded("Senior AI Engineer strong on Kubernetes.", source) |
|
|
|
|
| |
| def test_artifact_schema(tmp_path: Path) -> None: |
| settings = _settings(tmp_path) |
| add_explanation(_frame([_row()]), _SPEC, settings) |
| payload = orjson.loads((tmp_path / "explanations.json").read_bytes()) |
| entry = payload[0] |
| assert entry["rank"] == 1 and entry["tier"] == "T1" |
| assert entry["matched_skills"] == ["Python", "embeddings", "ranking"] |
| assert "bands" in entry and "signals" in entry and "positives" in entry and "sentences" in entry |
| assert entry["bm25_hits"]["skills"] is True |
| assert entry["llm_polished"] is False |
|
|
|
|
| def test_empty_passthrough(tmp_path: Path) -> None: |
| out = add_explanation(_frame([_row()]).head(0), _SPEC, _settings(tmp_path)) |
| assert out.height == 0 and {"reasoning", "explanation"} <= set(out.columns) |
|
|
