eb6ded5b3d
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.
69 lines
2.6 KiB
Python
69 lines
2.6 KiB
Python
"""The vibe-engineer — embedding similarity (ENGINE_DESIGN §3 embedding ranker).
|
|
|
|
Cosine(profile-text, job-text) via `text-embedding-3-small`: one batched call (profile + all jobs),
|
|
the profile embedded alongside. Returns id → cosine in [0,1]; `None` when embeddings are unavailable
|
|
(no key / API error / empty profile) so the sift degrades to white-box only.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from app.config import get_settings
|
|
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", ""),
|
|
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 []),
|
|
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)
|
|
|
|
|
|
def _cos(a, b) -> float:
|
|
import numpy as np
|
|
|
|
a, b = np.asarray(a, dtype=float), np.asarray(b, dtype=float)
|
|
n = float(np.linalg.norm(a) * np.linalg.norm(b))
|
|
return float(a @ b / n) if n else 0.0
|
|
|
|
|
|
def vibe_scores(prefs: dict, ctx: dict | None, jobs: list[dict]) -> dict[str, float] | None:
|
|
"""id → cosine sim (0..1). None if embeddings can't run (caller falls back to white-box only)."""
|
|
client = embed_client()
|
|
if not client or not jobs:
|
|
return None
|
|
ptext = _profile_text(prefs, ctx)
|
|
if not ptext.strip():
|
|
return None
|
|
s = get_settings()
|
|
texts = [ptext] + [_job_text(j) for j in jobs]
|
|
try:
|
|
resp = client.embeddings.create(model=s.EMBED_MODEL, input=texts)
|
|
except Exception:
|
|
return None
|
|
vecs = [d.embedding for d in resp.data]
|
|
pvec, jvecs = vecs[0], vecs[1:]
|
|
return {job["id"]: max(0.0, min(1.0, _cos(pvec, jv))) for job, jv in zip(jobs, jvecs)}
|