97 lines
3.4 KiB
Python
97 lines
3.4 KiB
Python
"""Bounded batch orchestration.
|
|
|
|
Two properties this module exists to guarantee:
|
|
|
|
* Concurrency is bounded by a semaphore. There is no unbounded ``asyncio.gather``.
|
|
* The shared job-description prefix is cached before the batch fans out. A cache entry
|
|
only becomes readable once the first response has begun, so launching all candidates
|
|
at once means every one of them pays full input price and none reads the cache.
|
|
The first candidate is therefore awaited alone, priming the prefix for the rest.
|
|
|
|
``score_batch`` returns results in **input order**. Sorting is :func:`sort_results`,
|
|
applied by the caller once extraction failures have been merged back in.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import time
|
|
|
|
from app.core.errors import classify_error
|
|
from app.models.scoring import CandidateResult, CompletedCandidate, FailedCandidate
|
|
from app.services.llm import Scorer
|
|
from app.services.pdf import ExtractedResume
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _sort_key(result: CandidateResult) -> tuple[int, int]:
|
|
"""Completed first by descending score; failures last.
|
|
|
|
``sorted`` is stable and ``score_batch`` preserves input order, so ties and
|
|
failures both retain their original upload order.
|
|
"""
|
|
if isinstance(result, CompletedCandidate):
|
|
return (0, -result.match_score)
|
|
return (1, 0)
|
|
|
|
|
|
def sort_results(results: list[CandidateResult]) -> list[CandidateResult]:
|
|
return sorted(results, key=_sort_key)
|
|
|
|
|
|
async def score_batch(
|
|
items: list[ExtractedResume],
|
|
*,
|
|
job_description: str,
|
|
scorer: Scorer,
|
|
concurrency: int,
|
|
) -> list[CandidateResult]:
|
|
"""Score every extracted resume, isolating per-candidate failures."""
|
|
if not items:
|
|
return []
|
|
|
|
semaphore = asyncio.Semaphore(concurrency)
|
|
|
|
async def score_one(item: ExtractedResume) -> CandidateResult:
|
|
async with semaphore:
|
|
started = time.perf_counter()
|
|
try:
|
|
score = await scorer.score(job_description, item.text)
|
|
except asyncio.CancelledError:
|
|
# Never swallow cancellation.
|
|
raise
|
|
except Exception as exc:
|
|
error_code, error_message = classify_error(exc)
|
|
logger.exception(
|
|
"candidate_scoring_failed",
|
|
extra={
|
|
"file_name": item.filename,
|
|
"candidate_id": item.candidate_id,
|
|
"error_code": error_code,
|
|
"duration_ms": round((time.perf_counter() - started) * 1000),
|
|
},
|
|
)
|
|
return FailedCandidate(
|
|
filename=item.filename,
|
|
error_code=error_code,
|
|
error_message=error_message,
|
|
)
|
|
|
|
logger.info(
|
|
"candidate_scored",
|
|
extra={
|
|
"file_name": item.filename,
|
|
"candidate_id": item.candidate_id,
|
|
"status": "completed",
|
|
"duration_ms": round((time.perf_counter() - started) * 1000),
|
|
},
|
|
)
|
|
return CompletedCandidate(filename=item.filename, **score.model_dump())
|
|
|
|
# Prime the shared prefix cache on the first candidate, then fan out.
|
|
first = await score_one(items[0])
|
|
rest = await asyncio.gather(*(score_one(item) for item in items[1:]))
|
|
return [first, *rest]
|