HR-ATS-Portal/backend/agent/views.py

69 lines
2.0 KiB
Python

"""Agent graph node logic.
Pure module: no FastAPI imports and no HTTPException.
LLM client/config lives in llm_setup — nodes call llm_call only.
"""
from __future__ import annotations
import logging
from typing import Literal
from langgraph.graph import END
from agent.decorators import normalize_job_posts,parse_match_response
from agent.models import AgentState
from agent.prompt import prompt,user_prompt
from llm_setup import llm_call
logger=logging.getLogger("agent")
async def prepare_context(state:AgentState) -> dict:
subject=(state.get("subject") or "").strip()
resume_text=(state.get("resume_text") or "").strip()
job_posts=normalize_job_posts(state.get("job_posts"))
if not resume_text:
return {"status":"skipped","error":"resume_text is empty","suggested_job_post_ids":[]}
if not job_posts:
return {"status":"skipped","error":"no active job posts to match against","suggested_job_post_ids":[]}
return {
"subject":subject,
"resume_text":resume_text,
"job_posts":job_posts,
"status":"ready",
"error":"",
}
def route_after_prepare(state:AgentState) -> Literal["match_jobs","__end__"]:
if state.get("status")=="ready":
return "match_jobs"
return END
async def match_jobs(state:AgentState) -> dict:
try:
data=await llm_call(prompt(),user_prompt(state),json_mode=True)
allowed_ids={item["id"] for item in state.get("job_posts") or []}
suggested,summary,reasoning,experience=parse_match_response(data,allowed_ids)
return {
"status":"matched",
"suggested_job_post_ids":suggested,
"summary":summary,
"reasoning":reasoning,
"experience":experience,
}
except Exception as e:
logger.exception("agent match_jobs failed")
return {
"status":"failed",
"error":str(e),
"suggested_job_post_ids":[],
"summary":"",
"reasoning":"",
"experience":"",
}