Deploy SHL assessment recommender

.gitattributes

@@ -0,0 +1 @@
+data/catalog.index filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -0,0 +1,19 @@
+# Python-generated files
+__pycache__/
+*.py[oc]
+build/
+dist/
+wheels/
+*.egg-info
+
+# Virtual environments
+.venv
+
+# Environment variables & secrets
+.env
+.env.*
+
+# Editor folders
+.vscode/
+.idea/
+

Dockerfile

@@ -0,0 +1,13 @@
+FROM python:3.11-slim
+
+WORKDIR /code
+
+# Copy requirements and install
+COPY requirements.txt /code/requirements.txt
+RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt
+
+# Copy application files and prebuilt index
+COPY . /code
+
+# HF Spaces runs on port 7860 by default
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -0,0 +1,13 @@
+---
+title: SHL Recommender
+emoji: 🤖
+colorFrom: blue
+colorTo: indigo
+sdk: docker
+app_port: 7860
+pinned: false
+---
+
+# SHL Assessment Recommender
+
+This is a conversational FastAPI agent for recommending SHL assessments based on user queries, built using FastAPI, FAISS, and Gemini embeddings.

data/catalog.index

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:4c72e695173f0c19ecc39376b2b9afe50b17837243b6a72fecf5dae5be75236f
+size 1158189

data/catalog_meta.json

@@ -0,0 +1 @@
+["4302", "3827", "4094", "4099", "4018", "4016", "4165", "4178", "4187", "4179", "4188", "4168", "4019", "3778", "4069", "4004", "4223", "4159", "4311", "4028", "4160", "4153", "4021", "4030", "4011", "4073", "4075", "4022", "4115", "4023", "4017", "3786", "4219", "3980", "3981", "3984", "3982", "3983", "4222", "3985", "4229", "4026", "4025", "3988", "4206", "4149", "4161", "4036", "4037", "244", "536", "4038", "4039", "4122", "4077", "4044", "4012", "4024", "4150", "4045", "4046", "4104", "4189", "4296", "4034", "4032", "3458", "4052", "4050", "3933", "3931", "4053", "4186", "778", "779", "785", "4013", "56", "731", "4162", "4101", "4003", "4221", "4239", "4059", "4060", "4062", "1102", "4148", "4009", "4001", "4031", "4002", "4000", "4139", "4180", "4063", "3856", "4287", "3934", "3939", "3935", "3936", "3937", "3938", "4065", "743", "3849", "3899", "4127", "380", "381", "3992", "4083", "4067", "382", "383", "4090", "4071", "4072", "4113", "4043", "4048", "4125", "4066", "4078", "4301", "741", "3900", "3901", "4074", "331", "3845", "4284", "4285", "4076", "4080", "4081", "3999", "4169", "4155", "4005", "4054", "4055", "4007", "249", "333", "4102", "4040", "88", "4084", "4158", "4056", "3809", "4033", "3989", "4086", "4085", "4087", "4152", "3990", "4299", "205", "4170", "742", "3903", "3902", "3904", "4157", "4291", "4293", "4295", "4292", "4294", "4008", "4089", "4006", "4176", "4058", "4119", "3951", "3950", "3949", "3948", "4027", "4091", "4092", "4208", "4207", "3785", "4212", "3789", "3807", "4211", "4210", "4047", "4093", "4174", "4096", "4010", "724", "1308", "1306", "1309", "1048", "4015", "3993", "3995", "4193", "4097", "3994", "4098", "615", "4100", "3997", "4041", "720", "4116", "727", "4298", "748", "749", "750", "4300", "752", "754", "753", "4307", "1050", "1060", "1059", "1058", "1061", "756", "757", "4286", "758", "4289", "1067", "759", "4106", "4105", "4145", "4156", "4154", "4107", "3998", "4108", "4109", "4070", "4110", "4088", "4114", "4061", "4111", "4112", "4103", "3746", "3484", "4042", "4118", "4117", "4095", "4121", "4120", "219", "3769", "393", "4123", "4124", "4177", "3472", "543", "3942", "4204", "4202", "4203", "4126", "697", "395", "4129", "4128", "3932", "3930", "726", "725", "4230", "4283", "4233", "4288", "4130", "4131", "4132", "4133", "4136", "4134", "4135", "4082", "4137", "4064", "4171", "4138", "4140", "3968", "3947", "3946", "3971", "3972", "4141", "4209", "4218", "4205", "4014", "116", "4142", "4051", "40", "4143", "4144", "4035", "4282", "4146", "4147", "3996", "4151", "4216", "3986", "4217", "3987", "4197", "4198", "4199", "4200", "4049", "4079", "4020", "4167", "251", "336", "4201", "4175", "735", "734", "733", "4172", "4173", "3733", "3941", "3908", "3734", "3836", "3745", "3768", "3906", "3940", "3945", "3970", "3969", "3976", "3974", "4290", "4215", "4163", "399", "400", "4164", "3459", "3991", "4068", "4183", "4184", "4185", "17", "18", "4166", "4297", "744"]

main.py

@@ -0,0 +1,45 @@
+import traceback
+from fastapi import FastAPI, HTTPException
+from fastapi.responses import JSONResponse
+from models import ChatRequest, ChatResponse, Recommendation
+import temp  # Imports and loads catalog, index, and advisor configuration once
+
+app = FastAPI(title="SHL Assessment Recommender")
+
+
+@app.get("/health")
+def health():
+    return {"status": "ok"}
+
+
+@app.post("/chat", response_model=ChatResponse)
+def chat(req: ChatRequest):
+    messages = [{"role": m.role, "content": m.content} for m in req.messages]
+
+    if not messages or messages[-1]["role"] != "user":
+        raise HTTPException(
+            status_code=400,
+            detail="messages must be non-empty and end with a user turn",
+        )
+
+    result = temp.agent_turn(messages)
+
+    return ChatResponse(
+        reply=result["reply"],
+        recommendations=[Recommendation(**r) for r in result["recommendations"]],
+        end_of_conversation=result["end_of_conversation"],
+    )
+
+
+@app.exception_handler(Exception)
+async def unhandled_exception_handler(request, exc):
+    print("Unhandled exception occurred:")
+    traceback.print_exc()
+    return JSONResponse(
+        status_code=200,
+        content={
+            "reply": "Something went wrong on our side — could you repeat that?",
+            "recommendations": [],
+            "end_of_conversation": False,
+        },
+    )

models.py

@@ -0,0 +1,23 @@
+from typing import List
+from pydantic import BaseModel
+
+
+class Message(BaseModel):
+    role: str
+    content: str
+
+
+class ChatRequest(BaseModel):
+    messages: List[Message]
+
+
+class Recommendation(BaseModel):
+    name: str
+    url: str
+    test_type: str
+
+
+class ChatResponse(BaseModel):
+    reply: str
+    recommendations: List[Recommendation]
+    end_of_conversation: bool

requirements.txt

@@ -0,0 +1,25 @@
+# ── Web Framework ────────────────────────────────────────────────────────────
+fastapi==0.115.5
+uvicorn[standard]==0.32.1
+python-dotenv==1.0.1
+
+# ── Data validation ───────────────────────────────────────────────────────────
+pydantic==2.10.3
+
+# ── LLM (Groq) ───────────────────────────────────────────────────────────────
+groq==0.13.0
+
+# ── Embeddings (Gemini API — no local model download) ────────────────────────
+google-genai==1.16.0
+
+# ── Vector store ─────────────────────────────────────────────────────────────
+faiss-cpu==1.9.0
+
+# ── Numerical / ML utilities ─────────────────────────────────────────────────
+numpy==1.26.4
+
+# ── HTTP client (optional, for health-check tests) ───────────────────────────
+httpx==0.28.1
+
+# ── Dev / testing ─────────────────────────────────────────────────────────────
+pytest==8.3.4

temp.py

@@ -0,0 +1,311 @@
+import json
+import os
+import re
+import numpy as np
+import faiss
+from dotenv import load_dotenv
+from google import genai
+from google.genai import types
+from groq import Groq
+
+load_dotenv()
+
+INDEX_PATH = "data/catalog.index"
+META_PATH = "data/catalog_meta.json"
+CATALOG_PATH = "shl_product_catalog.json"
+EMBED_MODEL = "models/gemini-embedding-2"
+EMBED_DIM = 768
+GROQ_MODEL = "llama-3.3-70b-versatile"
+MAX_USER_TURNS = 4  # hard budget: 4 user + 4 agent = 8 total, matches evaluator cap
+
+gemini_client = genai.Client(api_key=os.environ["GOOGLE_API_KEY"])
+groq_client = Groq(api_key=os.environ["GROQ_API_KEY"])
+index = faiss.read_index(INDEX_PATH)
+meta = json.load(open(META_PATH))
+catalog_list = json.load(open(CATALOG_PATH, encoding="utf-8"))
+catalog_map = {item["entity_id"]: item for item in catalog_list}
+# url -> entity_id for fast post-LLM validation
+url_to_id = {item["link"]: item["entity_id"] for item in catalog_list}
+
+# ── Key-to-letter mapping (derived from catalog keys field) ──────────────────
+KEY_LETTER = {
+    "Knowledge & Skills": "K",
+    "Personality & Behavior": "P",
+    "Ability & Aptitude": "A",
+    "Simulations": "S",
+    "Biodata & Situational Judgment": "B",
+    "Competencies": "C",
+    "Development & 360": "D",
+    "Assessment Exercises": "E",
+}
+
+
+def keys_to_type(keys: list[str]) -> str:
+    """Return comma-joined letter codes for a catalog item's keys list."""
+    seen, letters = set(), []
+    for k in keys:
+        letter = KEY_LETTER.get(k)
+        if letter and letter not in seen:
+            seen.add(letter)
+            letters.append(letter)
+    return ",".join(letters) if letters else "K"
+
+
+# ── Retrieval ────────────────────────────────────────────────────────────────
+# All semantic decisions (EOC, intent classification, refusals) are owned by the LLM.
+# Retrieval uses fixed anchor queries unconditionally — no regex routing needed.
+
+_ANCHORS = [
+    # Cognitive / personality — useful for nearly all selection queries
+    "cognitive ability reasoning personality behaviour workplace",
+    # Development / reskilling — surfaces GSA, 360, dev reports
+    "development reskilling skills self-assessment 360 feedback report",
+    # Spoken language / SVAR — surfaces voice assessment variants by language
+    "spoken language voice assessment English Spanish accent SVAR",
+]
+
+
+def build_intent(messages: list[dict]) -> str:
+    """
+    Consolidated intent string from conversation history.
+    The latest user message is repeated to weight recent refinements
+    (e.g. 'add AWS, drop REST') more heavily in the embedding.
+    """
+    user_msgs = [m["content"] for m in messages if m["role"] == "user"]
+    if not user_msgs:
+        return ""
+    earlier = " ".join(user_msgs[:-1])
+    latest = user_msgs[-1]
+    return f"{earlier} {latest} {latest}".strip()
+
+
+def retrieve(messages: list[dict], k_main: int = 15, k_anchor: int = 3) -> list[dict]:
+    """
+    Returns a de-duplicated, rank-ordered list of catalog entries.
+    Rank order matters now: it's used as the fallback shortlist when the
+    LLM fails to commit by the turn cap (see agent_turn).
+    """
+    intent = build_intent(messages)
+
+    # Primary search on the full conversation intent
+    _, idx_main = index.search(embed(intent), k_main)
+
+    # Fixed anchors — always run all three, let the LLM pick what's relevant.
+    # Kept behind the main results so the fallback path prioritises the
+    # query-specific matches over the generic anchors.
+    anchor_ids: list[int] = []
+    for anchor in _ANCHORS:
+        _, idx_anchor = index.search(embed(anchor), k_anchor)
+        anchor_ids += list(idx_anchor[0])
+
+    seen, results = set(), []
+    for i in list(idx_main[0]) + anchor_ids:
+        if i != -1 and meta[i] not in seen:
+            seen.add(meta[i])
+            results.append(catalog_map[meta[i]])
+    return results
+
+
+def embed(text: str) -> np.ndarray:
+    result = gemini_client.models.embed_content(
+        model=EMBED_MODEL,
+        contents=text,
+        config=types.EmbedContentConfig(
+            task_type="RETRIEVAL_QUERY",
+            output_dimensionality=EMBED_DIM,
+        ),
+    )
+    vec = np.array(result.embeddings[0].values, dtype="float32")
+    vec /= np.linalg.norm(vec)
+    return vec.reshape(1, -1)
+
+
+# ── Catalog snippet formatter ────────────────────────────────────────────────
+def format_snippets(items: list[dict]) -> str:
+    parts = []
+    for i, item in enumerate(items, 1):
+        type_letter = keys_to_type(item["keys"])
+        snippet = (
+            f"[{i}] name={item['name']}\n"
+            f"    url={item['link']}\n"
+            f"    test_type={type_letter} | "
+            f"categories={', '.join(item['keys'])} | "
+            f"duration={item['duration'] or 'N/A'} "
+            f"| remote={item['remote']} | adaptive={item['adaptive']}\n"
+            f"    levels={', '.join(item['job_levels'])}\n"
+            f"    {item['description'][:400]}"
+        )
+        parts.append(snippet)
+    return "\n\n".join(parts)
+
+
+def to_recommendation(item: dict) -> dict:
+    return {
+        "name": item["name"],
+        "url": item["link"],
+        "test_type": keys_to_type(item["keys"]),
+    }
+
+
+# ── Post-LLM recommendation validator ───────────────────────────────────────
+def validate_recommendations(recs: list[dict] | None) -> list[dict] | None:
+    """
+    Strip any recommendation whose URL is not in the catalog.
+    Also correct the test_type letter from the catalog (never trust the LLM for this),
+    and de-dupe on entity_id so near-duplicate SKUs (e.g. two 'Verify G+' variants)
+    can't both slip through if the LLM names both.
+    """
+    if not recs:
+        return recs
+    seen_ids, validated = set(), []
+    for r in recs:
+        url = r.get("url", "")
+        entity_id = url_to_id.get(url)
+        if entity_id is None:
+            # Try a forgiving match: strip trailing slash differences
+            url_norm = url.rstrip("/") + "/"
+            entity_id = url_to_id.get(url_norm)
+        if entity_id is None or entity_id in seen_ids:
+            # URL not in catalog, or a duplicate — drop this item
+            continue
+        seen_ids.add(entity_id)
+        validated.append(to_recommendation(catalog_map[entity_id]))
+    return validated if validated else None
+
+
+# ── System prompt ────────────────────────────────────────────────────────────
+# ── System prompt ────────────────────────────────────────────────────────────
+SYSTEM_PROMPT = """You are an SHL advisor. Recommend from CATALOG ENTRIES below only. Output JSON only.
+
+RULES:
+1. Gather only: role, yoe, tech stack, purpose (selection/dev) in turns 1-3. Ask questions; recommend early only if confident and explicitly ask for feedback/extra requirements unless context is fully clear.
+2. Turn {current_turn} of 4. By turn 4, output 7-10 recommendations and set end_of_conversation=true.
+3. Recommend leadership/personality for seniors; cognitive/coding for juniors. No hallucinations.
+4. Refuse legal/general HR advice. Refine shortlist in-place on feedback. Compare using catalog only.
+
+CATALOG ENTRIES:
+{catalog_snippets}
+
+OUTPUT:
+{{"reply": "text reply/question", "recommendations": [], "end_of_conversation": false}}"""
+
+
+# ── LLM call ─────────────────────────────────────────────────────────────────
+def call_llm(messages: list[dict], snippets: str, current_turn: int) -> str:
+    system = (
+        SYSTEM_PROMPT
+        .replace("{catalog_snippets}", snippets)
+        .replace("{current_turn}", str(current_turn))
+        .replace("{max_turns}", str(MAX_USER_TURNS))
+        .replace("{last_clarify_turn}", str(MAX_USER_TURNS - 1))
+    )
+    resp = groq_client.chat.completions.create(
+        model=GROQ_MODEL,
+        messages=[{"role": "system", "content": system}] + messages,
+        temperature=0.2,
+        max_tokens=1500,
+        response_format={"type": "json_object"},
+    )
+    return resp.choices[0].message.content
+
+
+# ── JSON parser ───────────────────────────────────────────────────────────────
+def parse(raw: str) -> dict:
+    # Strip markdown code fences if present (belt-and-braces even with response_format=json_object)
+    cleaned = re.sub(r"```(?:json)?", "", raw).strip()
+    match = re.search(r"\{.*\}", cleaned, re.DOTALL)
+    if not match:
+        raise json.JSONDecodeError("no JSON object found", raw, 0)
+    return json.loads(match.group())
+
+
+# ── Main agent turn ───────────────────────────────────────────────────────────
+def agent_turn(messages: list[dict]) -> dict:
+    """
+    Full agent pipeline for one turn.
+    Returns a dict: {reply, recommendations, end_of_conversation}
+    Guarantees: on the last allowed user turn, recommendations is always
+    non-empty (falls back to top retrieved catalog items) and
+    end_of_conversation is always true.
+    """
+    # Count user turns so far (the last message is the current user turn)
+    current_turn = sum(1 for m in messages if m["role"] == "user")
+    is_last_turn = current_turn >= MAX_USER_TURNS
+
+    # 1. Retrieve catalog items (rank-ordered — first items are the best matches)
+    catalog_items = retrieve(messages)
+    snippets = format_snippets(catalog_items)
+
+    # 2. LLM call — all intent decisions owned by the model
+    try:
+        raw = call_llm(messages, snippets, current_turn)
+        parsed = parse(raw)
+    except (json.JSONDecodeError, Exception):
+        parsed = {}
+
+    reply = parsed.get("reply") or ""
+    raw_recs = parsed.get("recommendations")
+
+    # 3. Validate URLs + test_types; normalise to list
+    if isinstance(raw_recs, list) and raw_recs:
+        recommendations = validate_recommendations(raw_recs) or []
+    else:
+        recommendations = []
+    recommendations = recommendations[:10]
+
+    # 4. Force-commit on the last turn: guarantee a non-empty, catalog-grounded
+    #    shortlist even if the LLM stalled, refused, or returned malformed JSON.
+    if is_last_turn and not recommendations:
+        recommendations = [to_recommendation(item) for item in catalog_items[:7]]
+        if not reply:
+            reply = (
+                "Based on everything discussed, here is a shortlist that fits "
+                "your requirements."
+            )
+
+    # 5. EOC: respect the LLM's signal, but always force true on the last turn.
+    eoc = bool(parsed.get("end_of_conversation", False)) or is_last_turn
+
+    return {
+        "reply": reply,
+        "recommendations": recommendations,
+        "end_of_conversation": eoc,
+    }
+
+
+# ── CLI entrypoint ────────────────────────────────────────────────────────────
+def main():
+    messages = []
+    print("SHL Assessment Advisor")
+    print("-" * 50)
+
+    while True:
+        user_input = input("You: ").strip()
+        if not user_input or user_input.lower() in ("quit", "exit"):
+            break
+
+        messages.append({"role": "user", "content": user_input})
+        result = agent_turn(messages)
+
+        reply = result["reply"]
+        recommendations = result["recommendations"]
+        eoc = result["end_of_conversation"]
+
+        print(f"\nAgent: {reply}")
+
+        if recommendations:
+            print("\nRecommendations:")
+            for i, r in enumerate(recommendations, 1):
+                print(f"  {i}. [{r['test_type']}] {r['name']}")
+                print(f"     {r['url']}")
+
+        print()
+        messages.append({"role": "assistant", "content": reply})
+
+        if eoc:
+            print("Conversation complete.")
+            break
+
+
+if __name__ == "__main__":
+    main()