growqr-app/matchmaking-v2
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Engine Phase 2: fix the pre-Opus funnel (lifts ALL matches)

Browse Source 
The sift was feeding Opus a magnitude-blind, under-trusted, truncated shortlist — only 68% of the
genuinely-best jobs reached the curator. Fixes:
- embed.py: embed the FULL description (was desc[:600]) + role_category + industry; richer profile
  text (current_role + experience summary + seniority). The best semantic signal, fed real content.
- sift.py: magnitude-preserving min-max fusion (was rank-position, which flattened cosine 0.95 vs
  0.72) and embedding-LED weights (W_VIBE 0.6 / W_WHITEBOX 0.4). Embed runs FIRST so the white-box
  f_semantic reuses the real cosine (threaded as job["_vibe_cosine"]) instead of token overlap.
- config.py: SIFT_TOP_K 18 → 28 (the gate was cutting good jobs before Opus).
- curate.py: richer Opus briefs — full responsibilities (was desc[:500]) + role_category/industry/
  seniority/pay, so even the shortlist is fully described.

Regression (120 pairs): sift membership recall 0.68 → 0.92, score MAE 24.7 → 18.4. 31 tests pass.
This commit is contained in:
raulgupta
2026-06-25 15:50:14 +05:30
parent 04e6fb64bb
commit eb6ded5b3d
7 changed files with 72 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
app/config.py
View File

@@ -40,7 +40,8 @@ class Settings(BaseSettings):
SUGGEST_MODEL: str = "claude-haiku-4-5" # the bubbling engine — fast + cheap (latency-sensitive UI)
EMBED_MODEL: str = "text-embedding-3-small" # the vibe-engineer (direct OpenAI — gateway has no embeddings route)
ENGINE_LLM_ENABLED: bool = True # off / key absent → white-box sift + templated cards (safety net)
SIFT_TOP_K: int = 18 # candidate pool the curator reads
SIFT_TOP_K: int = 28 # candidate pool the curator reads (widened from 18 —
# the funnel was cutting ~32% of the best jobs before Opus)
# Result cache (dev cost-saver). Keyed by (actor, exact input) → saved job set on disk.
# off = always live, never save (PRODUCTION default)

8
app/engine/curate.py
View File

@@ -64,8 +64,12 @@ def _job_brief(j: dict) -> dict:
"title": j.get("title"),
"company": j.get("organization"),
"location": f"{j.get('location_city') or ''} {j.get('location_country') or ''}".strip(),
"skills": (d.get("skills") or [])[:12],
"snippet": (d.get("description") or "")[:500],
"role_category": d.get("role_category"),
"industry": d.get("industry"),
"seniority": j.get("seniority_level"),
"pay": j.get("payLabel"),
"skills": (d.get("skills") or j.get("required_skills") or [])[:12],
"responsibilities": (d.get("description") or "")[:1200], # was 500 — Opus scored from a stub
"prelim_score": j.get("matchScore"),
}

14
app/engine/embed.py
View File

@@ -11,24 +11,32 @@ from app.engine.llm import embed_client
def _job_text(job: dict) -> str:
# The embedding is the best semantic matcher — feed it the FULL signal, not a 600-char stub.
# role_category + industry anchor non-tech JDs (narrative prose, few repeated title tokens).
d = job.get("details") or {}
parts = [
job.get("title", ""),
job.get("organization", ""),
" ".join(d.get("skills") or []),
(d.get("description") or "")[:600],
d.get("role_category") or "",
d.get("industry") or "",
" ".join(d.get("skills") or job.get("required_skills") or []),
(d.get("description") or "")[:2000],
]
return " · ".join(p for p in parts if p)
def _profile_text(prefs: dict, ctx: dict | None) -> str:
ctx = ctx or {}
# Embed the candidate richly too — current_role + experience summary + seniority, not just title+skills.
exp = ctx.get("experience_summary") or ctx.get("summary") or ""
parts = [
prefs.get("title", ""),
" ".join(prefs.get("role") or []),
" ".join(ctx.get("skills") or []),
ctx.get("current_role", "") or "",
" ".join(ctx.get("skills") or []),
" ".join(prefs.get("industry") or []),
" ".join(prefs.get("experience") or []),
str(exp)[:1200],
]
return " · ".join(p for p in parts if p)

4
app/engine/rank.py
View File

@@ -130,6 +130,10 @@ def f_role(p, job): return _cat_match(p.roles, _jrole(job))
def f_semantic(p, job):
# Real embedding cosine when the sift threaded it on (the true semantic signal); else token overlap.
c = job.get("_vibe_cosine")
if c is not None:
return float(c)
if not p.profile_tokens:
return None
blob = f"{job.get('title') or ''} {(job.get('details') or {}).get('description') or ''}".lower()

47
app/engine/sift.py
View File

@@ -11,32 +11,49 @@ from app.config import get_settings
from app.engine import embed as _embed
from app.engine import rank as _rank
W_WHITEBOX, W_VIBE = 0.55, 0.45 # white-box (checklist) leads slightly; vibe (semantics) sharpens
W_WHITEBOX, W_VIBE = 0.4, 0.6 # embedding (the truer semantic signal) now LEADS the noisy checklist
def _ranks(score_by_id: dict[str, float]) -> dict[str, float]:
"""id → normalized rank in [0,1] (1 = best, 0 = worst). Position-based, so scale-free."""
order = sorted(score_by_id, key=lambda i: -score_by_id[i])
n = len(order)
return {i: (1.0 - pos / (n - 1)) if n > 1 else 1.0 for pos, i in enumerate(order)}
def _minmax(score_by_id: dict[str, float]) -> dict[str, float]:
"""id → score min-max-normalized into [0,1]. PRESERVES MAGNITUDE (a much-better job stays much
better) — unlike rank-position fusion, which flattened cosine 0.95 and 0.72 to the same gap."""
vals = score_by_id.values()
lo, hi = min(vals), max(vals)
rng = hi - lo
return {i: ((s - lo) / rng if rng else 1.0) for i, s in score_by_id.items()}
def sift(prefs: dict, ctx: dict | None, jobs: list[dict], k: int | None = None):
"""Returns (top_k_jobs, debug). top_k_jobs are ranked best-first; each keeps its white-box match."""
"""Returns (top_k_jobs, debug). top_k_jobs are ranked best-first; each keeps its white-box match.
Order matters: embed FIRST so the white-box `f_semantic` can reuse the real cosine (via the
threaded `_vibe_cosine`) instead of crude token overlap. Then fuse by magnitude, embedding-led."""
s = get_settings()
k = k or s.SIFT_TOP_K
ranked = _rank.rank(prefs, ctx, jobs)["opportunities"] # white-box scores ALL (no early cut)
if not ranked:
if not jobs:
return [], {"mode": "empty", "scored": 0, "k": 0}
wb = {j["id"]: j["matchScore"] / 100.0 for j in ranked}
vibe = _embed.vibe_scores(prefs, ctx, ranked) # id → cosine, or None
# 1. embedding cosine FIRST — thread onto each job so rank.f_semantic reads the real signal.
vibe = _embed.vibe_scores(prefs, ctx, jobs) # id → cosine, or None
if vibe:
wr, vr = _ranks(wb), _ranks(vibe)
fused = {i: W_WHITEBOX * wr[i] + W_VIBE * vr.get(i, 0.0) for i in wb}
mode = "whitebox+vibe"
for j in jobs:
if j["id"] in vibe:
j["_vibe_cosine"] = vibe[j["id"]]
# 2. white-box scores ALL (now semantically-aware where the cosine was threaded).
ranked = _rank.rank(prefs, ctx, jobs)["opportunities"]
if not ranked:
return [], {"mode": "empty", "scored": 0, "k": 0}
wb = {j["id"]: j["matchScore"] / 100.0 for j in ranked}
# 3. magnitude-preserving, embedding-led fusion.
if vibe:
wbn = _minmax(wb)
vbn = _minmax({i: vibe[i] for i in wb if i in vibe})
fused = {i: W_WHITEBOX * wbn[i] + W_VIBE * vbn.get(i, 0.0) for i in wb}
mode = "vibe-led(magnitude)"
else:
fused = _ranks(wb)
fused = wb
mode = "whitebox-only"
top = sorted(ranked, key=lambda j: -fused[j["id"]])[:k]

13
tests/run_regression.py
View File

@@ -20,6 +20,7 @@ from pathlib import Path
from app.engine import rank as _rank
from app.engine import sift as _sift
from app.engine import embed as _embed
from app.config import get_settings
FIXTURE = Path(__file__).parent / "fixtures" / "regression_set.json"
@@ -39,9 +40,17 @@ def run(emit_json: bool = False):
for c in cases:
prefs, ctx, jobs = c["prefs"], c.get("user_context"), c["jobs"]
targets = {str(j): int(s) for j, s in c["targets"].items()}
# engine: white-box scores + the sift top-K the curator would see
ranked = _rank.rank(prefs, ctx, [dict(j) for j in jobs])["opportunities"]
# engine score = the funnel's white-box AFTER the embedding cosine is threaded in (what the
# fallback shows + feeds the sift) — replicate embed→thread→rank so MAE reflects the real funnel.
jc = [dict(j) for j in jobs]
vibe = _embed.vibe_scores(prefs, ctx, jc)
if vibe:
for j in jc:
if j["id"] in vibe:
j["_vibe_cosine"] = vibe[j["id"]]
ranked = _rank.rank(prefs, ctx, jc)["opportunities"]
eng = {j["id"]: j.get("matchScore", 0) for j in ranked}
# the sift top-K the curator actually sees
top, _ = _sift.sift(prefs, ctx, [dict(j) for j in jobs], k=k)
top_ids = {j["id"] for j in top}

11
tests/test_sift.py
View File

@@ -1,12 +1,13 @@
"""A1 — sift: rank-fusion math + graceful degradation when the vibe-engineer is unavailable.
"""A1 — sift: magnitude-fusion math + graceful degradation when the vibe-engineer is unavailable.
(Offline — no network; the embedding path is exercised in the cached e2e verification.)"""
from app.engine import sift as S
def test_ranks_normalization():
r = S._ranks({"a": 0.9, "b": 0.5, "c": 0.1})
assert r["a"] == 1.0 and r["c"] == 0.0 and 0.0 < r["b"] < 1.0
assert S._ranks({"x": 0.5}) == {"x": 1.0} # single item → top
def test_minmax_normalization():
# magnitude-preserving (min-max), NOT rank-position: a much-better score stays much better.
r = S._minmax({"a": 0.9, "b": 0.5, "c": 0.1})
assert r["a"] == 1.0 and r["c"] == 0.0 and abs(r["b"] - 0.5) < 1e-9
assert S._minmax({"x": 0.5}) == {"x": 1.0} # single item / no range → top
def test_sift_degrades_to_whitebox_when_no_vibe(monkeypatch):