HR-ATS-Portal/backend/job_assist/execute_agent.py

73 lines
2.9 KiB
Python

"""Field-assist entrypoint — llm_setup.llm_call only.
Pure module: no FastAPI imports and no HTTPException.
Called from job.app's POST /job/assist-field route; no HTTP surface of its own.
ValueError means the request itself is bad (unknown field/action, nothing to
fix) and maps to 422 at the route. RuntimeError means the model could not be
consulted or returned something unusable and maps to 503 — the route must not
echo its message to the client, since it can carry provider detail.
"""
from __future__ import annotations
import logging
from job_assist.decorators import parse_assist_response
from job_assist.prompt import ACTIONS, FIELD_RULES, SYSTEM_PROMPT, user_prompt
from llm_setup import llm_call
logger = logging.getLogger("job_assist")
# A suggestion is only writable once these context fields exist — without the
# role and its seniority the model can only produce generic filler.
# "experience" is satisfied by either end of the range.
SUGGEST_ANCHORS = {
"department": ("title",),
"location": ("title",),
"salary": ("title", "experience"),
"requirements": ("title", "experience"),
"optional_skills": ("title", "experience"),
"description": ("title", "experience"),
}
_ANCHOR_LABELS = {"title": "job title", "experience": "experience range"}
def _has_anchor(context, key) -> bool:
ctx = context or {}
if key == "experience":
return any(str(ctx.get(k) or "").strip() for k in ("experience_min", "experience_max"))
return bool(str(ctx.get(key) or "").strip())
async def run_field_assist(*, field, action, text="", context=None) -> str:
if field not in FIELD_RULES:
raise ValueError(f"unsupported field: {field}")
if action not in ACTIONS:
raise ValueError(f"unsupported action: {action}")
if action == "fix" and not (text or "").strip():
raise ValueError("nothing to fix: the field is empty")
if action == "suggest":
if field == "title":
ctx_values = (context or {}).values()
if not (text or "").strip() and not any(str(v or "").strip() for v in ctx_values):
raise ValueError("type a draft title or fill in another field first")
else:
missing = [a for a in SUGGEST_ANCHORS[field] if not _has_anchor(context, a)]
if missing:
names = " and ".join(_ANCHOR_LABELS[a] for a in missing)
raise ValueError(f"fill in the {names} first")
try:
data = await llm_call(
SYSTEM_PROMPT,
user_prompt(field, action, text, context),
json_mode=True,
)
return parse_assist_response(data, field)
except Exception as e:
# Log the type only — the message can carry prompt or form content.
logger.warning("field assist failed: field=%s action=%s exc=%s", field, action, type(e).__name__)
raise RuntimeError("field assist failed") from e