138 lines
5.2 KiB
Python
138 lines
5.2 KiB
Python
"""OpenAI adapter.
|
|
|
|
One shared ``AsyncOpenAI`` is created at startup and reused for every candidate --
|
|
required both for connection reuse and for prompt caching to behave predictably.
|
|
|
|
``responses.parse`` is used rather than a hand-built JSON schema. Pydantic emits
|
|
keywords the structured-outputs schema dialect rejects; ``parse`` derives and submits
|
|
a conforming schema, then validates the reply back into :class:`ATSScore`, so the
|
|
constraints on that model still gate every result.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import logging
|
|
from typing import Any, Protocol
|
|
|
|
from openai import AsyncOpenAI
|
|
|
|
from app.core.config import supports_reasoning
|
|
from app.core.errors import (
|
|
ModelRefusedError,
|
|
ModelResponseInvalidError,
|
|
ModelUnavailableError,
|
|
)
|
|
from app.models.scoring import ATSScore
|
|
from app.prompts.ats import SYSTEM_PROMPT, build_input
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Reasons the provider can return on an incomplete response.
|
|
_TRUNCATED = "max_output_tokens"
|
|
_FILTERED = "content_filter"
|
|
|
|
|
|
class Scorer(Protocol):
|
|
"""The seam tests replace with a fake. Nothing else may talk to the provider."""
|
|
|
|
async def score(self, job_description: str, resume_text: str) -> ATSScore: ...
|
|
|
|
|
|
def _first_refusal(response: Any) -> str | None:
|
|
"""Return the refusal text if the model declined, else ``None``.
|
|
|
|
A refusal arrives as a content part inside an output message, not as an error, so
|
|
it has to be walked for explicitly before the parsed output is trusted.
|
|
"""
|
|
for item in getattr(response, "output", None) or []:
|
|
for part in getattr(item, "content", None) or []:
|
|
if getattr(part, "type", None) == "refusal":
|
|
refusal = getattr(part, "refusal", None)
|
|
return str(refusal) if refusal else "refused"
|
|
return None
|
|
|
|
|
|
class OpenAIScorer:
|
|
def __init__(
|
|
self,
|
|
client: AsyncOpenAI,
|
|
*,
|
|
model: str,
|
|
max_output_tokens: int,
|
|
effort: str,
|
|
enable_cache: bool = True,
|
|
) -> None:
|
|
self._client = client
|
|
self._model = model
|
|
self._max_output_tokens = max_output_tokens
|
|
self._effort = effort
|
|
self._enable_cache = enable_cache
|
|
self._supports_reasoning = supports_reasoning(model)
|
|
|
|
def _cache_key(self, job_description: str) -> str:
|
|
"""Stable per job description, so a batch routes to one cache.
|
|
|
|
OpenAI caching is automatic; this is only a routing hint that raises the hit
|
|
rate by steering identical prefixes to the same machine. It must stay low
|
|
cardinality -- one value per batch, never per candidate.
|
|
"""
|
|
digest = hashlib.sha256(job_description.encode("utf-8")).hexdigest()[:32]
|
|
return f"ats-{digest}"
|
|
|
|
async def score(self, job_description: str, resume_text: str) -> ATSScore:
|
|
kwargs: dict[str, Any] = {
|
|
"model": self._model,
|
|
"instructions": SYSTEM_PROMPT,
|
|
"input": build_input(job_description, resume_text),
|
|
"text_format": ATSScore,
|
|
"max_output_tokens": self._max_output_tokens,
|
|
}
|
|
if self._supports_reasoning:
|
|
kwargs["reasoning"] = {"effort": self._effort}
|
|
if self._enable_cache:
|
|
kwargs["prompt_cache_key"] = self._cache_key(job_description)
|
|
|
|
response = await self._client.responses.parse(**kwargs)
|
|
|
|
status = getattr(response, "status", None)
|
|
self._log_usage(response, status)
|
|
|
|
# Branch on delivery status before trusting any output.
|
|
if status == "failed":
|
|
raise ModelUnavailableError("provider reported a failed response")
|
|
|
|
if status == "incomplete":
|
|
reason = getattr(getattr(response, "incomplete_details", None), "reason", None)
|
|
if reason == _FILTERED:
|
|
raise ModelRefusedError("content filter blocked the response")
|
|
if reason == _TRUNCATED:
|
|
raise ModelResponseInvalidError("response truncated at max_output_tokens")
|
|
raise ModelResponseInvalidError(f"incomplete response: {reason}")
|
|
|
|
refusal = _first_refusal(response)
|
|
if refusal is not None:
|
|
raise ModelRefusedError("model declined to score this document")
|
|
|
|
parsed = getattr(response, "output_parsed", None)
|
|
if not isinstance(parsed, ATSScore):
|
|
raise ModelResponseInvalidError("response did not parse into ATSScore")
|
|
return parsed
|
|
|
|
def _log_usage(self, response: Any, status: object) -> None:
|
|
usage = getattr(response, "usage", None)
|
|
input_details = getattr(usage, "input_tokens_details", None)
|
|
output_details = getattr(usage, "output_tokens_details", None)
|
|
logger.info(
|
|
"candidate_scored_upstream",
|
|
extra={
|
|
"model": self._model,
|
|
"stop_reason": status,
|
|
"provider_request_id": getattr(response, "id", None),
|
|
"input_tokens": getattr(usage, "input_tokens", None),
|
|
"output_tokens": getattr(usage, "output_tokens", None),
|
|
"cached_tokens": getattr(input_details, "cached_tokens", None),
|
|
"reasoning_tokens": getattr(output_details, "reasoning_tokens", None),
|
|
},
|
|
)
|