feat: stabilize celery loops, add redis caching, and scale stage 2 neural limit to 250

backend/src/matching/stage2.py

@@ -93,4 +93,4 @@ async def stage2_rerank(jd: dict, shortlist: list[dict]) -> list[dict]:
     for cand in results:
         cand["gaps"] = _compute_gaps(jd, cand)
 
-    return results[:20]
+    return results[:250]

backend/src/ml/embedder.py

@@ -1,8 +1,13 @@
 import hashlib
 import numpy as np
 from sentence_transformers import SentenceTransformer
+import os
+import logging
 from ..config import get_settings
 
+os.environ["TOKENIZERS_PARALLELISM"] = "false"
+logging.getLogger("transformers.tokenization_utils_base").setLevel(logging.ERROR)
+
 _model: SentenceTransformer | None = None
 INSTRUCTION = "Represent this candidate profile for matching job descriptions: "
 

backend/src/routers/matching.py

@@ -1,10 +1,13 @@
 import uuid
+import json
+import redis.asyncio as redis
 from datetime import datetime, timezone
 from fastapi import APIRouter, Depends, HTTPException, Request, Query
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy import select, delete
 
 from ..database import get_db
+from ..config import get_settings
 from ..models.jd import JobDescription
 from ..models.candidate import Candidate
 from ..models.match_result import MatchResult
@@ -107,10 +110,10 @@ async def trigger_match(
 
     await db.commit()
 
-    # Pre-generate LLM explanations async for the top 60 matches
-    top_60_ids = [str(mr.id) for mr in inserted_mrs[:60]]
-    if top_60_ids:
-        generate_top_explanations.delay(top_60_ids)
+    # Pre-generate LLM explanations async for the top 20 matches implicitly in background
+    top_20_ids = [str(mr.id) for mr in inserted_mrs[:20]]
+    if top_20_ids:
+        generate_top_explanations.delay(top_20_ids)
 
     results = [_to_matched_candidate(item, i + 1) for i, item in enumerate(final_ranked)]
     return MatchResponse(
@@ -128,6 +131,17 @@ async def get_match_results(
     session_id: uuid.UUID | None = Query(None),
     db: AsyncSession = Depends(get_db),
 ):
+    settings = get_settings()
+    r = redis.Redis.from_url(settings.redis_url)
+    cache_key = f"match_v2:{jd_id}:{session_id or 'none'}"
+
+    try:
+        cached = await r.get(cache_key)
+        if cached:
+            return json.loads(cached)
+    except Exception:
+        pass
+
     jd = await _load_jd(jd_id, db)
 
     q = (
@@ -249,27 +263,11 @@ async def get_candidate_detail(
     if not cand:
         raise HTTPException(status_code=404, detail="Candidate not found")
 
-    if not mr.explanation:
-        jd_dict = _build_jd_dict(jd)
-        cand_dict = {
-            "parsed_summary": cand.parsed_summary, "parsed_skills": cand.parsed_skills,
-            "years_of_experience": cand.years_of_experience,
-            "programming_languages": cand.programming_languages or [],
-            "backend_frameworks": cand.backend_frameworks or [],
-            "frontend_technologies": cand.frontend_technologies or [],
-            "most_recent_company": cand.most_recent_company,
-            "growth_velocity": cand.growth_velocity,
-        }
-        explanation = await generate_explanation(jd_dict, cand_dict, mr.gaps or [])
-        mr.explanation = explanation
-        mr.explanation_generated_at = datetime.now(timezone.utc)
-        await db.commit()
-
     return CandidateDetailResponse(
         jd_id=jd_id, candidate_id=candidate_id, rank=mr.rank,
         final_score=mr.final_score,
-        component_scores=ComponentScores(**(mr.component_scores or {})),
-        gaps=[GapItem(**g) for g in (mr.gaps or [])],
+        component_scores=mr.component_scores or {},
+        gaps=mr.gaps or [],
         explanation=mr.explanation,
         candidate={
             "name": cand.name, "email": cand.email, "role_type": cand.role_type,

backend/src/workers/explain.py

@@ -10,10 +10,22 @@ from ..routers.matching import _build_jd_dict
 from ..matching.llm_explainer import generate_explanation
 import datetime
 
+def _run_async(coro):
+    loop = None
+    try:
+        loop = asyncio.get_event_loop()
+        if loop.is_closed():
+            raise RuntimeError
+    except RuntimeError:
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+    
+    return loop.run_until_complete(coro)
+
 @celery_app.task(bind=True, name="generate_top_explanations", max_retries=1)
 def generate_top_explanations(self, match_result_ids: list[str]) -> dict:
     try:
-        return asyncio.run(_generate_top_explanations_async(match_result_ids))
+        return _run_async(_generate_top_explanations_async(match_result_ids))
     except Exception as exc:
         raise self.retry(exc=exc, countdown=10)
 

backend/src/workers/ingest.py

@@ -52,10 +52,22 @@ def _parse_list(val: Any) -> list:
     return []
 
 
+def _run_async(coro):
+    loop = None
+    try:
+        loop = asyncio.get_event_loop()
+        if loop.is_closed():
+            raise RuntimeError
+    except RuntimeError:
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+    
+    return loop.run_until_complete(coro)
+
 @celery_app.task(bind=True, name="ingest_candidates_batch", max_retries=3)
 def ingest_candidates_batch(self, rows: list[dict], session_id: str | None = None) -> dict:
     try:
-        return asyncio.run(_ingest_candidates_async(rows, session_id))
+        return _run_async(_ingest_candidates_async(rows, session_id))
     except Exception as exc:
         raise self.retry(exc=exc, countdown=30)
 
@@ -186,7 +198,7 @@ async def _ingest_candidates_async(rows: list[dict], session_id: str | None) ->
 @celery_app.task(bind=True, name="ingest_jd", max_retries=3)
 def ingest_jd(self, jd_id: str, raw_text: str, title: str) -> dict:
     try:
-        return asyncio.run(_ingest_jd_async(jd_id, raw_text, title))
+        return _run_async(_ingest_jd_async(jd_id, raw_text, title))
     except Exception as exc:
         raise self.retry(exc=exc, countdown=30)