Spaces:

hembad
/

reranker

Sleeping

reranker / src /rank /explain /compose.py

Hemprasad Badgujar

Add observability logging for LLM and embedder

fbbd1c5 4 months ago

17.3 kB

"""Layer 2 — advanced rule-based NLG: aggregate grounded facts into 3–4 fluent sentences (NO model).
Builds ordered positive + negative (gap) clause pools from the banks, aggregates same-type facts under
connectives, and assembles tier-shaped sentences. Phrasing variety comes from a ``VariantPicker`` — either blind
``random`` or grounded ``potion`` top-k (cosine of each variant vs the candidate's own evidence, then random
among the top-k → grounded *and* varied). Negatives reach the CSV only where the tier allows (≤1 soft gap, T3/T4);
the ``explanation`` text carries them all. Pure Python, microseconds per candidate.
"""

from __future__ import annotations

import random
import re
from typing import Any, Callable

_LEAD_CONNECTIVE = re.compile(r"^(though|however,?|but|with|while)\s+", re.IGNORECASE)
# Seniority words that may already sit in a title — if any is present, don't prefix the band label again.
_TITLE_SENIORITY = ("senior", "staff", "lead", "principal", "junior", "head", "chief", "vp", "director", "intern")

from jd import JDSpec

from . import banks
from .banks import BANKS, NEG_BANKS
from .signals import BEHAVIOR_LEAD, BEHAVIOR_TAGS, PRIORITY, SIG_NEG_BANKS

# A picker: (variant tuple, per-candidate evidence vector) → chosen variant string.
Picker = Callable[[tuple[str, ...], Any], str]


def random_picker(variants: tuple[str, ...], _evidence: Any = None) -> str:
    return random.choice(variants)


class PotionPicker:
    """Grounded variant selection — cosine of each variant vs the candidate evidence, random among the top-k.
    Reuses the already-loaded potion-base-32M (Model2Vec). The fixed variant vocabulary is embedded once. Any
    failure (model missing, encode error) raises at construction so the caller falls back to ``random_picker``.
    """

    def __init__(self, repo: str, topk: int) -> None:
        import numpy as np
        from retrieve.potion import _embed_unit, _load_potion

        self._np = np
        self._embed = _embed_unit
        self._model = _load_potion(repo)
        self._topk = max(1, topk)
        vecs = _embed_unit(self._model, list(banks.ALL_VARIANTS))
        self._vec = {v: vecs[i] for i, v in enumerate(banks.ALL_VARIANTS)}

    def evidence_vector(self, text: str) -> Any:
        return self._embed(self._model, [text or "candidate"])[0]

    def _vec_of(self, variant: str) -> Any:
        """Variant embedding, lazily computed + cached for any slot not in the precomputed vocabulary."""
        vec = self._vec.get(variant)
        if vec is None:
            vec = self._embed(self._model, [variant])[0]
            self._vec[variant] = vec
        return vec

    def __call__(self, variants: tuple[str, ...], evidence: Any) -> str:
        if evidence is None:
            return random.choice(variants)
        scored = sorted(variants, key=lambda v: float(evidence @ self._vec_of(v)), reverse=True)
        return random.choice(scored[: self._topk])


def make_picker(settings: Any) -> tuple[Picker, Any]:
    """Build the configured picker; return (picker, potion_or_None). Falls back to random on any failure."""
    reasoning = settings.reasoning
    if reasoning.variant_select == "potion_topk":
        try:
            potion = PotionPicker(settings.models.embedder_broad, reasoning.variant_topk)
            return potion, potion
        except Exception as exc:  # noqa: BLE001 — potion unavailable → blind random variety (still grounded by facts)
            from common.logging import get_logger

            get_logger("rank.explain").warning(
                "explain.potion_picker_unavailable", model=settings.models.embedder_broad, error=str(exc)
            )
            return random_picker, None
    return random_picker, None


def _fill(slot: str, picker: Picker, evidence: Any, neg: bool = False, **kw: Any) -> str:
    bank = (NEG_BANKS if neg else BANKS)[slot]
    return picker(bank, evidence).format(**kw)


_FINAL_CONNECTIVE = ("and", "and", "and", "as well as")  # mostly "and", occasional "as well as"; no casual "plus"
_LEAD_JOINER = re.compile(r"^(and|plus|also|as well as)\s+", re.IGNORECASE)  # defensive: no "and also brings…"


def _oxford(items: list[str]) -> str:
    """Join clause fragments naturally with a varied final connective: 'a', 'a and b', 'a, b and c'."""
    items = [_LEAD_JOINER.sub("", i.strip()) for i in items if i and i.strip()]
    if len(items) <= 1:
        return items[0] if items else ""
    conn = random.choice(_FINAL_CONNECTIVE)
    if len(items) == 2:
        return f"{items[0]} {conn} {items[1]}"
    return ", ".join(items[:-1]) + f" {conn} {items[-1]}"


_A_BEFORE_VOWEL = re.compile(r"\b([Aa])\s+(?=[aeioAEIO])")  # "a AI" → "an AI"; skip 'u' (keeps "a UI"/"a URL")


def _fix_articles(text: str) -> str:
    """Correct the indefinite article before a vowel-letter word ('a AI Engineer' → 'an AI Engineer')."""
    return _A_BEFORE_VOWEL.sub(lambda m: ("An" if m.group(1) == "A" else "an") + " ", text)


def _sentence(text: str) -> str:
    text = text.strip().rstrip(" ,;")
    if not text:
        return ""
    return _fix_articles(text[0].upper() + text[1:]) + "."


def _evidence_text(facts: dict[str, Any]) -> str:
    """The candidate's own grounded evidence — what potion compares phrasing variants against."""
    parts = [facts["title"], facts["seniority"], *facts["matched_skills"][:4],
             *facts["matched_industries"][:1], facts["city"]]
    return " ".join(p for p in parts if p)


def _verdict_clauses(facts: dict[str, Any], picker: Picker, ev: Any) -> list[str]:
    """Sentence-1 VERDICT: how the candidate fits the role (experience band + title fit)."""
    out: list[str] = []
    out.append(_fill(f"experience.{facts['exp_fit']}", picker, ev))
    if facts["title_band"] == "align":
        out.append(_fill("title.align", picker, ev, title_jd=facts["title_match"]))
    elif facts["title_band"] == "adjacent":
        out.append(_fill("title.adjacent", picker, ev, title_jd=facts["title_match"]))
    return out


def _strength_clauses(facts: dict[str, Any], picker: Picker, ev: Any, skills_cap: int) -> list[str]:
    """Sentence-2 STRENGTHS: concrete evidence (skills, industry, education, certs, location, adjacent depth)."""
    out: list[str] = []
    if facts["matched_skills"]:
        slot = "skills.strong" if facts["skill_band"] == "strong" else "skills.partial"
        out.append(_fill(slot, picker, ev, skills=_oxford(facts["matched_skills"][:skills_cap])))
    # Location high (recruiters care) so it survives the top-3 cap in S2.
    if facts["location_tier"] == "preferred":
        out.append(_fill("location.preferred", picker, ev, city=facts["city"]))
    elif facts["location_tier"] == "acceptable":
        out.append(_fill("location.acceptable", picker, ev, city=facts["city"]))
    if facts["matched_nice"] and not facts["matched_skills"]:  # adjacent depth matters most when core is thin
        out.append(_fill("skills.nice", picker, ev, skills=_oxford(facts["matched_nice"][:skills_cap])))
    if facts["matched_industries"]:
        out.append(_fill("industry.match", picker, ev, ind=facts["matched_industries"][0]))
    if facts["matched_education"]:
        out.append(_fill("education.match", picker, ev, field=facts["matched_education"][0]))
    if facts["matched_certs"]:
        out.append(_fill("cert.match", picker, ev, cert=facts["matched_certs"][0]))
    if facts["matched_nice"] and facts["matched_skills"]:  # core present → nice-to-have is a bonus, lower priority
        out.append(_fill("skills.nice", picker, ev, skills=_oxford(facts["matched_nice"][:skills_cap])))
    return out


def _behavior_tags(facts: dict[str, Any], picker: Picker, ev: Any) -> list[str]:
    """Best-first, impactful behavior tags (deduped); recency leads, then the PRIORITY-ordered signals."""
    sig = facts["signals"]
    tags: list[str] = []
    if facts["recency_band"] in ("month", "quarter"):
        tags.append(picker(BEHAVIOR_TAGS["recency"], ev))
    for key, want, slot, polarity in PRIORITY:
        if polarity != "pos" or slot not in BEHAVIOR_TAGS:
            continue
        value = sig.get(key)
        if value if want is None else value == want:
            tags.append(picker(BEHAVIOR_TAGS[slot], ev))
    if facts["notice_band"] == "short":
        tags.append(picker(BEHAVIOR_TAGS["notice.short"], ev))
    seen: set[str] = set()
    return [t for t in tags if not (t in seen or seen.add(t))]


def _behavior_sentence(facts: dict[str, Any], picker: Picker, ev: Any) -> str:
    """Sentence-3 BEHAVIOR as a narrative — one of several structural forms (variety, never a fixed template)."""
    tags = _behavior_tags(facts, picker, ev)[:3]
    if not tags:
        return ""
    body = _oxford(tags)
    if not facts["signals"].get("active_market"):
        return body  # lighter: just the behaviors, capitalized by _sentence
    forms = (
        f"{picker(BEHAVIOR_LEAD, ev)}: {body}",
        f"behaviourally, this candidate is {body}",
        f"on engagement, they are {body}",
        f"the behavioural read is positive: {body}",
        body,
    )
    return random.choice(forms)


def _behavioral_negs(facts: dict[str, Any], picker: Picker, ev: Any) -> list[str]:
    """Behavioral GAP clauses (low responsiveness, thin profile, …) for the audit + the one allowed CSV gap."""
    sig = facts["signals"]
    neg: list[str] = []
    for key, want, slot, polarity in PRIORITY:
        if polarity != "neg":
            continue
        value = sig.get(key)
        if value if want is None else value == want:
            neg.append(picker(SIG_NEG_BANKS[slot], ev))
    return neg


def _gap_clauses(facts: dict[str, Any], picker: Picker, ev: Any) -> list[str]:
    """Ordered JD-fit gaps, most material first (missing must-have > title > location > company > notice/recency)."""
    gaps: list[str] = []
    if facts["skill_band"] == "thin" and not facts["matched_skills"]:
        gaps.append(_fill("skills.thin", picker, ev, neg=True))
    elif facts["missing_must"] and facts["skill_band"] != "strong":
        gaps.append(_fill("skills.missing", picker, ev, neg=True,
                          missing=_oxford(facts["missing_must"][:2])))
    if facts["title_band"] == "mismatch":
        gaps.append(_fill("title.mismatch", picker, ev, neg=True, title_jd=facts["title_match"]))
    if facts["location_tier"] == "outside":
        gaps.append(_fill("location.outside", picker, ev, neg=True))
    if facts["country_fit"] == "outside":
        gaps.append(_fill("country.outside", picker, ev, neg=True))
    if facts["exclude_hit"]:
        gaps.append(_fill("company.exclude", picker, ev, neg=True))
    if facts["recency_band"] == "stale":
        gaps.append(_fill("recency.stale", picker, ev, neg=True))
    if facts["notice_band"] == "long":
        gaps.append(_fill("notice.long", picker, ev, neg=True))
    return gaps


# Varied "but" pivots for the single soft gap (no dash). All must read with a NOUN-PHRASE gap (e.g. "thin overlap
# on the core stack") — so subordinators / participials only, never "…is that <noun phrase>".
_GAP_PIVOT = ("though", "albeit", "though notably", "the main gap being", "with the caveat of")


def _gap_join(core: str, gap: str) -> str:
    """Fold ONE soft gap into a strengths clause with a varied pivot (composer owns the pivot)."""
    gap = _LEAD_CONNECTIVE.sub("", gap.strip())
    pivot = random.choice(_GAP_PIVOT)
    if not core:
        return f"{pivot[0].upper() + pivot[1:]} {gap}"
    return f"{core}, {pivot} {gap}"


def _assemble(lead: str, verdict: list[str], strengths: list[str], behavior: str, gap: str | None,
              signal: str, target: int) -> tuple[str, list[str]]:
    """A+B structure: S1 identity+verdict, S2 strengths(+gap), S3 behavior narrative, S4 why-ranked."""
    s1 = _sentence(f"{lead}, {_oxford(verdict[:2])}" if verdict else lead)
    s2_core = _oxford(strengths[:3])
    if gap:
        s2_core = _gap_join(s2_core, gap)
    s2 = _sentence(s2_core)
    s3 = _sentence(behavior) if behavior else ""
    s4 = _sentence(signal)
    sentences = [s for s in (s1, s2, s3, s4) if s]
    # Pad toward the target by promoting any leftover strength clauses into their own sentence (before the signal).
    leftover = strengths[3:]
    while len(sentences) < target and leftover:
        sentences.insert(-1, _sentence(_oxford(leftover[:2])))
        leftover = leftover[2:]
    sentences = sentences[:target]
    return " ".join(sentences), sentences


# The closing "why ranked" sentence — varied opener + the dominant signal + which JD fields matched lexically.
# We deliberately DON'T restate the rank ("top 10") — that's already the rank column; instead we add the signal
# and concrete BM25 field hits (more information, not redundant).
_SIGNAL_LEAD = ("ranked on", "surfaced on", "stood out on", "selected on", "rose on the strength of",
                "shortlisted on")
_BM25_FIELD_LABEL = {"skills": "skills", "titles": "title", "education": "education",
                     "locations": "location", "industries": "industry"}
# Natural ways to say "these fields are RELEVANT to the JD requirement" (not the mechanical "keyword matches on").
# Each entry INCLUDES its own leading punctuation/connector, and the openers AND the ending noun are varied
# (role / brief / spec / posting / listing / needs / ask / requirements) so the closing clause never feels like a
# fixed template. ``{fields}`` is filled with the relevant field list.
_SIGNAL_MATCH = (
    ", with {fields} relevant to the role",
    ", a strong match on {fields}",
    "; {fields} map cleanly to the brief",
    ", relevant across {fields}",
    "; {fields} fit what the role asks",
    "; their {fields} answer the spec",
    ", a close fit on {fields}",
    "; {fields} are on point for the posting",
    ", strong on the {fields} this role calls for",
    ", well-matched on {fields}",
    "; {fields} line up with the listing",
    "; {fields} meet the brief",
    ", relevant on {fields} for what the role needs",
    ", a clean match across {fields}",
    "; {fields} track the role's needs",
    "; {fields} speak to the requirements",
    ", squarely on the {fields} the role wants",
    "; {fields} answer the ask",
)


def _signal_match_tail(hits: list[str], picker: Picker, ev: Any) -> str:
    """A varied, natural 'relevant to the requirement' tail (own punctuation); '' when nothing matched."""
    if not hits:
        return ""
    return picker(_SIGNAL_MATCH, ev).format(fields=_oxford(hits[:3]))


def _signal_sentence(facts: dict[str, Any], picker: Picker, ev: Any, brief: bool = False) -> str:
    """"Why ranked": dominant signal + which of the candidate's fields are RELEVANT to the JD (no rank ordinal)."""
    lead = picker(_SIGNAL_LEAD, ev)
    sentence = f"{lead} {facts['signal_phrase']}"
    hits = [_BM25_FIELD_LABEL[f] for f, hit in facts["bm25_hits"].items() if hit and f in _BM25_FIELD_LABEL]
    if not brief:  # the relevance tail (own leading punctuation) is the first thing dropped when trimming length
        sentence += _signal_match_tail(hits, picker, ev)
    return sentence


def compose_one(
    facts: dict[str, Any], spec: JDSpec, picker: Picker, ev_for: Callable[[dict], Any],
    *, sentences: int, allow_neg: bool, max_neg: int, skills_cap: int,
) -> dict[str, Any]:
    """Compose the NLG reasoning + explanation for one candidate at its tier settings."""
    ev = ev_for(facts)
    # Avoid doubled seniority ("Senior Senior…", "Lead Staff…") — if the title already carries ANY seniority word
    # (even a different one than the band), lead with the title alone.
    seniority = facts["seniority"]
    title_l = facts["title"].lower()
    if seniority and (seniority.lower() in title_l or any(w in title_l for w in _TITLE_SENIORITY)):
        lead = facts["title"]
    else:
        lead = _fill("identity.lead", picker, ev, title=facts["title"], seniority=seniority)
    verdict = _verdict_clauses(facts, picker, ev)
    strengths = _strength_clauses(facts, picker, ev, skills_cap)
    behavior = _behavior_sentence(facts, picker, ev)
    gaps = _gap_clauses(facts, picker, ev) + _behavioral_negs(facts, picker, ev)
    signal = _signal_sentence(facts, picker, ev)

    csv_gap = gaps[0] if (allow_neg and max_neg > 0 and gaps) else None
    # No char-count trim: the sentence-count target (3-4) and skills_cap already bound the line structurally,
    # so a fixed char ceiling only ever discarded content (the keyword tail, then strengths) for no real reason.
    reasoning, sent_list = _assemble(lead, verdict, strengths, behavior, csv_gap, signal, sentences)

    # explanation = the NLG body + the full audit tail (ALL gaps, title detail) — always LLM-free.
    audit_bits = []
    if gaps:
        audit_bits.append("Watch-outs: " + "; ".join(gaps) + ".")
    if facts["title_match"]:
        audit_bits.append(f"Title fit: {facts['title_match']} ({facts['title_band']}).")
    explanation = (reasoning + " " + " ".join(audit_bits)).strip()
    return {
        "reasoning": reasoning,
        "explanation": explanation,
        "sentences": sent_list,
        "positives": verdict + strengths + ([behavior] if behavior else []),
        "negatives": gaps,
    }

	"""Layer 2 — advanced rule-based NLG: aggregate grounded facts into 3–4 fluent sentences (NO model).

	Builds ordered positive + negative (gap) clause pools from the banks, aggregates same-type facts under
	connectives, and assembles tier-shaped sentences. Phrasing variety comes from a ``VariantPicker`` — either blind
	``random`` or grounded ``potion`` top-k (cosine of each variant vs the candidate's own evidence, then random
	among the top-k → grounded and varied). Negatives reach the CSV only where the tier allows (≤1 soft gap, T3/T4);
	the ``explanation`` text carries them all. Pure Python, microseconds per candidate.
	"""

	from __future__ import annotations

	import random
	import re
	from typing import Any, Callable

	_LEAD_CONNECTIVE = re.compile(r"^(though\|however,?\|but\|with\|while)\s+", re.IGNORECASE)
	# Seniority words that may already sit in a title — if any is present, don't prefix the band label again.
	_TITLE_SENIORITY = ("senior", "staff", "lead", "principal", "junior", "head", "chief", "vp", "director", "intern")

	from jd import JDSpec

	from . import banks
	from .banks import BANKS, NEG_BANKS
	from .signals import BEHAVIOR_LEAD, BEHAVIOR_TAGS, PRIORITY, SIG_NEG_BANKS

	# A picker: (variant tuple, per-candidate evidence vector) → chosen variant string.
	Picker = Callable[[tuple[str, ...], Any], str]


	def random_picker(variants: tuple[str, ...], _evidence: Any = None) -> str:
	return random.choice(variants)


	class PotionPicker:
	"""Grounded variant selection — cosine of each variant vs the candidate evidence, random among the top-k.

	Reuses the already-loaded potion-base-32M (Model2Vec). The fixed variant vocabulary is embedded once. Any
	failure (model missing, encode error) raises at construction so the caller falls back to ``random_picker``.
	"""

	def __init__(self, repo: str, topk: int) -> None:
	import numpy as np
	from retrieve.potion import _embed_unit, _load_potion

	self._np = np
	self._embed = _embed_unit
	self._model = _load_potion(repo)
	self._topk = max(1, topk)
	vecs = _embed_unit(self._model, list(banks.ALL_VARIANTS))
	self._vec = {v: vecs[i] for i, v in enumerate(banks.ALL_VARIANTS)}

	def evidence_vector(self, text: str) -> Any:
	return self._embed(self._model, [text or "candidate"])[0]

	def _vec_of(self, variant: str) -> Any:
	"""Variant embedding, lazily computed + cached for any slot not in the precomputed vocabulary."""
	vec = self._vec.get(variant)
	if vec is None:
	vec = self._embed(self._model, [variant])[0]
	self._vec[variant] = vec
	return vec

	def __call__(self, variants: tuple[str, ...], evidence: Any) -> str:
	if evidence is None:
	return random.choice(variants)
	scored = sorted(variants, key=lambda v: float(evidence @ self._vec_of(v)), reverse=True)
	return random.choice(scored[: self._topk])


	def make_picker(settings: Any) -> tuple[Picker, Any]:
	"""Build the configured picker; return (picker, potion_or_None). Falls back to random on any failure."""
	reasoning = settings.reasoning
	if reasoning.variant_select == "potion_topk":
	try:
	potion = PotionPicker(settings.models.embedder_broad, reasoning.variant_topk)
	return potion, potion
	except Exception as exc: # noqa: BLE001 — potion unavailable → blind random variety (still grounded by facts)
	from common.logging import get_logger

	get_logger("rank.explain").warning(
	"explain.potion_picker_unavailable", model=settings.models.embedder_broad, error=str(exc)
	)
	return random_picker, None
	return random_picker, None


	def _fill(slot: str, picker: Picker, evidence: Any, neg: bool = False, **kw: Any) -> str:
	bank = (NEG_BANKS if neg else BANKS)[slot]
	return picker(bank, evidence).format(**kw)


	_FINAL_CONNECTIVE = ("and", "and", "and", "as well as") # mostly "and", occasional "as well as"; no casual "plus"
	_LEAD_JOINER = re.compile(r"^(and\|plus\|also\|as well as)\s+", re.IGNORECASE) # defensive: no "and also brings…"


	def _oxford(items: list[str]) -> str:
	"""Join clause fragments naturally with a varied final connective: 'a', 'a and b', 'a, b and c'."""
	items = [_LEAD_JOINER.sub("", i.strip()) for i in items if i and i.strip()]
	if len(items) <= 1:
	return items[0] if items else ""
	conn = random.choice(_FINAL_CONNECTIVE)
	if len(items) == 2:
	return f"{items[0]} {conn} {items[1]}"
	return ", ".join(items[:-1]) + f" {conn} {items[-1]}"


	_A_BEFORE_VOWEL = re.compile(r"\b([Aa])\s+(?=[aeioAEIO])") # "a AI" → "an AI"; skip 'u' (keeps "a UI"/"a URL")


	def _fix_articles(text: str) -> str:
	"""Correct the indefinite article before a vowel-letter word ('a AI Engineer' → 'an AI Engineer')."""
	return _A_BEFORE_VOWEL.sub(lambda m: ("An" if m.group(1) == "A" else "an") + " ", text)


	def _sentence(text: str) -> str:
	text = text.strip().rstrip(" ,;")
	if not text:
	return ""
	return _fix_articles(text[0].upper() + text[1:]) + "."


	def _evidence_text(facts: dict[str, Any]) -> str:
	"""The candidate's own grounded evidence — what potion compares phrasing variants against."""
	parts = [facts["title"], facts["seniority"], *facts["matched_skills"][:4],
	*facts["matched_industries"][:1], facts["city"]]
	return " ".join(p for p in parts if p)


	def _verdict_clauses(facts: dict[str, Any], picker: Picker, ev: Any) -> list[str]:
	"""Sentence-1 VERDICT: how the candidate fits the role (experience band + title fit)."""
	out: list[str] = []
	out.append(_fill(f"experience.{facts['exp_fit']}", picker, ev))
	if facts["title_band"] == "align":
	out.append(_fill("title.align", picker, ev, title_jd=facts["title_match"]))
	elif facts["title_band"] == "adjacent":
	out.append(_fill("title.adjacent", picker, ev, title_jd=facts["title_match"]))
	return out


	def _strength_clauses(facts: dict[str, Any], picker: Picker, ev: Any, skills_cap: int) -> list[str]:
	"""Sentence-2 STRENGTHS: concrete evidence (skills, industry, education, certs, location, adjacent depth)."""
	out: list[str] = []
	if facts["matched_skills"]:
	slot = "skills.strong" if facts["skill_band"] == "strong" else "skills.partial"
	out.append(_fill(slot, picker, ev, skills=_oxford(facts["matched_skills"][:skills_cap])))
	# Location high (recruiters care) so it survives the top-3 cap in S2.
	if facts["location_tier"] == "preferred":
	out.append(_fill("location.preferred", picker, ev, city=facts["city"]))
	elif facts["location_tier"] == "acceptable":
	out.append(_fill("location.acceptable", picker, ev, city=facts["city"]))
	if facts["matched_nice"] and not facts["matched_skills"]: # adjacent depth matters most when core is thin
	out.append(_fill("skills.nice", picker, ev, skills=_oxford(facts["matched_nice"][:skills_cap])))
	if facts["matched_industries"]:
	out.append(_fill("industry.match", picker, ev, ind=facts["matched_industries"][0]))
	if facts["matched_education"]:
	out.append(_fill("education.match", picker, ev, field=facts["matched_education"][0]))
	if facts["matched_certs"]:
	out.append(_fill("cert.match", picker, ev, cert=facts["matched_certs"][0]))
	if facts["matched_nice"] and facts["matched_skills"]: # core present → nice-to-have is a bonus, lower priority
	out.append(_fill("skills.nice", picker, ev, skills=_oxford(facts["matched_nice"][:skills_cap])))
	return out


	def _behavior_tags(facts: dict[str, Any], picker: Picker, ev: Any) -> list[str]:
	"""Best-first, impactful behavior tags (deduped); recency leads, then the PRIORITY-ordered signals."""
	sig = facts["signals"]
	tags: list[str] = []
	if facts["recency_band"] in ("month", "quarter"):
	tags.append(picker(BEHAVIOR_TAGS["recency"], ev))
	for key, want, slot, polarity in PRIORITY:
	if polarity != "pos" or slot not in BEHAVIOR_TAGS:
	continue
	value = sig.get(key)
	if value if want is None else value == want:
	tags.append(picker(BEHAVIOR_TAGS[slot], ev))
	if facts["notice_band"] == "short":
	tags.append(picker(BEHAVIOR_TAGS["notice.short"], ev))
	seen: set[str] = set()
	return [t for t in tags if not (t in seen or seen.add(t))]


	def _behavior_sentence(facts: dict[str, Any], picker: Picker, ev: Any) -> str:
	"""Sentence-3 BEHAVIOR as a narrative — one of several structural forms (variety, never a fixed template)."""
	tags = _behavior_tags(facts, picker, ev)[:3]
	if not tags:
	return ""
	body = _oxford(tags)
	if not facts["signals"].get("active_market"):
	return body # lighter: just the behaviors, capitalized by _sentence
	forms = (
	f"{picker(BEHAVIOR_LEAD, ev)}: {body}",
	f"behaviourally, this candidate is {body}",
	f"on engagement, they are {body}",
	f"the behavioural read is positive: {body}",
	body,
	)
	return random.choice(forms)


	def _behavioral_negs(facts: dict[str, Any], picker: Picker, ev: Any) -> list[str]:
	"""Behavioral GAP clauses (low responsiveness, thin profile, …) for the audit + the one allowed CSV gap."""
	sig = facts["signals"]
	neg: list[str] = []
	for key, want, slot, polarity in PRIORITY:
	if polarity != "neg":
	continue
	value = sig.get(key)
	if value if want is None else value == want:
	neg.append(picker(SIG_NEG_BANKS[slot], ev))
	return neg


	def _gap_clauses(facts: dict[str, Any], picker: Picker, ev: Any) -> list[str]:
	"""Ordered JD-fit gaps, most material first (missing must-have > title > location > company > notice/recency)."""
	gaps: list[str] = []
	if facts["skill_band"] == "thin" and not facts["matched_skills"]:
	gaps.append(_fill("skills.thin", picker, ev, neg=True))
	elif facts["missing_must"] and facts["skill_band"] != "strong":
	gaps.append(_fill("skills.missing", picker, ev, neg=True,
	missing=_oxford(facts["missing_must"][:2])))
	if facts["title_band"] == "mismatch":
	gaps.append(_fill("title.mismatch", picker, ev, neg=True, title_jd=facts["title_match"]))
	if facts["location_tier"] == "outside":
	gaps.append(_fill("location.outside", picker, ev, neg=True))
	if facts["country_fit"] == "outside":
	gaps.append(_fill("country.outside", picker, ev, neg=True))
	if facts["exclude_hit"]:
	gaps.append(_fill("company.exclude", picker, ev, neg=True))
	if facts["recency_band"] == "stale":
	gaps.append(_fill("recency.stale", picker, ev, neg=True))
	if facts["notice_band"] == "long":
	gaps.append(_fill("notice.long", picker, ev, neg=True))
	return gaps


	# Varied "but" pivots for the single soft gap (no dash). All must read with a NOUN-PHRASE gap (e.g. "thin overlap
	# on the core stack") — so subordinators / participials only, never "…is that <noun phrase>".
	_GAP_PIVOT = ("though", "albeit", "though notably", "the main gap being", "with the caveat of")


	def _gap_join(core: str, gap: str) -> str:
	"""Fold ONE soft gap into a strengths clause with a varied pivot (composer owns the pivot)."""
	gap = _LEAD_CONNECTIVE.sub("", gap.strip())
	pivot = random.choice(_GAP_PIVOT)
	if not core:
	return f"{pivot[0].upper() + pivot[1:]} {gap}"
	return f"{core}, {pivot} {gap}"


	def _assemble(lead: str, verdict: list[str], strengths: list[str], behavior: str, gap: str \| None,
	signal: str, target: int) -> tuple[str, list[str]]:
	"""A+B structure: S1 identity+verdict, S2 strengths(+gap), S3 behavior narrative, S4 why-ranked."""
	s1 = _sentence(f"{lead}, {_oxford(verdict[:2])}" if verdict else lead)
	s2_core = _oxford(strengths[:3])
	if gap:
	s2_core = _gap_join(s2_core, gap)
	s2 = _sentence(s2_core)
	s3 = _sentence(behavior) if behavior else ""
	s4 = _sentence(signal)
	sentences = [s for s in (s1, s2, s3, s4) if s]
	# Pad toward the target by promoting any leftover strength clauses into their own sentence (before the signal).
	leftover = strengths[3:]
	while len(sentences) < target and leftover:
	sentences.insert(-1, _sentence(_oxford(leftover[:2])))
	leftover = leftover[2:]
	sentences = sentences[:target]
	return " ".join(sentences), sentences


	# The closing "why ranked" sentence — varied opener + the dominant signal + which JD fields matched lexically.
	# We deliberately DON'T restate the rank ("top 10") — that's already the rank column; instead we add the signal
	# and concrete BM25 field hits (more information, not redundant).
	_SIGNAL_LEAD = ("ranked on", "surfaced on", "stood out on", "selected on", "rose on the strength of",
	"shortlisted on")
	_BM25_FIELD_LABEL = {"skills": "skills", "titles": "title", "education": "education",
	"locations": "location", "industries": "industry"}
	# Natural ways to say "these fields are RELEVANT to the JD requirement" (not the mechanical "keyword matches on").
	# Each entry INCLUDES its own leading punctuation/connector, and the openers AND the ending noun are varied
	# (role / brief / spec / posting / listing / needs / ask / requirements) so the closing clause never feels like a
	# fixed template. ``{fields}`` is filled with the relevant field list.
	_SIGNAL_MATCH = (
	", with {fields} relevant to the role",
	", a strong match on {fields}",
	"; {fields} map cleanly to the brief",
	", relevant across {fields}",
	"; {fields} fit what the role asks",
	"; their {fields} answer the spec",
	", a close fit on {fields}",
	"; {fields} are on point for the posting",
	", strong on the {fields} this role calls for",
	", well-matched on {fields}",
	"; {fields} line up with the listing",
	"; {fields} meet the brief",
	", relevant on {fields} for what the role needs",
	", a clean match across {fields}",
	"; {fields} track the role's needs",
	"; {fields} speak to the requirements",
	", squarely on the {fields} the role wants",
	"; {fields} answer the ask",
	)


	def _signal_match_tail(hits: list[str], picker: Picker, ev: Any) -> str:
	"""A varied, natural 'relevant to the requirement' tail (own punctuation); '' when nothing matched."""
	if not hits:
	return ""
	return picker(_SIGNAL_MATCH, ev).format(fields=_oxford(hits[:3]))


	def _signal_sentence(facts: dict[str, Any], picker: Picker, ev: Any, brief: bool = False) -> str:
	""""Why ranked": dominant signal + which of the candidate's fields are RELEVANT to the JD (no rank ordinal)."""
	lead = picker(_SIGNAL_LEAD, ev)
	sentence = f"{lead} {facts['signal_phrase']}"
	hits = [_BM25_FIELD_LABEL[f] for f, hit in facts["bm25_hits"].items() if hit and f in _BM25_FIELD_LABEL]
	if not brief: # the relevance tail (own leading punctuation) is the first thing dropped when trimming length
	sentence += _signal_match_tail(hits, picker, ev)
	return sentence


	def compose_one(
	facts: dict[str, Any], spec: JDSpec, picker: Picker, ev_for: Callable[[dict], Any],
	*, sentences: int, allow_neg: bool, max_neg: int, skills_cap: int,
	) -> dict[str, Any]:
	"""Compose the NLG reasoning + explanation for one candidate at its tier settings."""
	ev = ev_for(facts)
	# Avoid doubled seniority ("Senior Senior…", "Lead Staff…") — if the title already carries ANY seniority word
	# (even a different one than the band), lead with the title alone.
	seniority = facts["seniority"]
	title_l = facts["title"].lower()
	if seniority and (seniority.lower() in title_l or any(w in title_l for w in _TITLE_SENIORITY)):
	lead = facts["title"]
	else:
	lead = _fill("identity.lead", picker, ev, title=facts["title"], seniority=seniority)
	verdict = _verdict_clauses(facts, picker, ev)
	strengths = _strength_clauses(facts, picker, ev, skills_cap)
	behavior = _behavior_sentence(facts, picker, ev)
	gaps = _gap_clauses(facts, picker, ev) + _behavioral_negs(facts, picker, ev)
	signal = _signal_sentence(facts, picker, ev)

	csv_gap = gaps[0] if (allow_neg and max_neg > 0 and gaps) else None
	# No char-count trim: the sentence-count target (3-4) and skills_cap already bound the line structurally,
	# so a fixed char ceiling only ever discarded content (the keyword tail, then strengths) for no real reason.
	reasoning, sent_list = _assemble(lead, verdict, strengths, behavior, csv_gap, signal, sentences)

	# explanation = the NLG body + the full audit tail (ALL gaps, title detail) — always LLM-free.
	audit_bits = []
	if gaps:
	audit_bits.append("Watch-outs: " + "; ".join(gaps) + ".")
	if facts["title_match"]:
	audit_bits.append(f"Title fit: {facts['title_match']} ({facts['title_band']}).")
	explanation = (reasoning + " " + " ".join(audit_bits)).strip()
	return {
	"reasoning": reasoning,
	"explanation": explanation,
	"sentences": sent_list,
	"positives": verdict + strengths + ([behavior] if behavior else []),
	"negatives": gaps,
	}