b1e9cdd182
- A1: fire the hard filter (Apify is precise); embedding vibe-ranker (embed.py) blends with the floor-free white-box (rank.py) to sift ~90 -> top 18 - A2: Opus curator (curate.py) reads the 18 -> honest <=count + Gemini-voiced report cards; graceful fallback to the white-box + templated cards - LLM via opencode.ai/zen gateway (llm.py): Opus chat + direct-OpenAI embeddings - Run LLM work off the event loop (asyncio.to_thread) so the Redis response publishes - C1: dedicated Postgres (app/db/) persists the per-user feed; get_scout_feed replays it so matches survive navigation/refresh - match contract + report-card fields (schema.py); skills.py; tests/
61 lines
2.0 KiB
Python
61 lines
2.0 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:
|
|
d = job.get("details") or {}
|
|
parts = [
|
|
job.get("title", ""),
|
|
job.get("organization", ""),
|
|
" ".join(d.get("skills") or []),
|
|
(d.get("description") or "")[:600],
|
|
]
|
|
return " · ".join(p for p in parts if p)
|
|
|
|
|
|
def _profile_text(prefs: dict, ctx: dict | None) -> str:
|
|
ctx = ctx or {}
|
|
parts = [
|
|
prefs.get("title", ""),
|
|
" ".join(prefs.get("role") or []),
|
|
" ".join(ctx.get("skills") or []),
|
|
ctx.get("current_role", "") or "",
|
|
" ".join(prefs.get("industry") or []),
|
|
]
|
|
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)}
|