97 lines
3.5 KiB
Python
97 lines
3.5 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 _embedding_vectors(s, texts: list[str]) -> list[list[float]] | None:
|
|
if s.OPENAI_API_BASE:
|
|
import httpx
|
|
|
|
try:
|
|
resp = httpx.post(
|
|
s.OPENAI_API_BASE.rstrip("/") + "/embeddings",
|
|
headers={
|
|
"Authorization": f"Bearer {s.OPENAI_API_KEY}",
|
|
"Content-Type": "application/json",
|
|
"HTTP-Referer": "https://sai-onchain.me",
|
|
"X-Title": "Sai Onchain Matchmaking",
|
|
},
|
|
json={"model": s.EMBED_MODEL, "input": texts},
|
|
timeout=30,
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
return [item["embedding"] for item in data.get("data", [])]
|
|
except Exception:
|
|
return None
|
|
client = embed_client()
|
|
if not client:
|
|
return None
|
|
try:
|
|
resp = client.embeddings.create(model=s.EMBED_MODEL, input=texts)
|
|
return [d.embedding for d in resp.data]
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
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 "")[:1000], # JD signal is front-loaded; 1k keeps embeds lean + cheap
|
|
]
|
|
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)."""
|
|
if 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]
|
|
vecs = _embedding_vectors(s, texts)
|
|
if not vecs or len(vecs) != len(texts):
|
|
return None
|
|
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)}
|