utopia-ai
/
HR-ATS-Portal
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

agent coide simplifgication

pull/6/head
ahmed.mujtaba 2026-08-10 16:24:47 +05:00
parent 08068b2b18
commit 4d340fd6d5
10 changed files with 148 additions and 219 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
backend/agent/agent_setup.py
View File

@ -13,45 +13,39 @@ from __future__ import annotations
import logging
from langgraph.graph import END, START, StateGraph
from langgraph.graph import END,START,StateGraph
from agent.models import AgentState
from agent.views import finalize, match_jobs, prepare_context, route_after_prepare
from agent.views import match_jobs,prepare_context,route_after_prepare
logger = logging.getLogger("agent")
logger=logging.getLogger("agent")
_graph = None
_graph=None
def build_graph():
"""Construct and compile the HR-ATS candidate matching graph."""
graph = StateGraph(AgentState)
graph.add_node("prepare", prepare_context)
graph.add_node("match_jobs", match_jobs)
graph.add_node("finalize", finalize)
graph.add_edge(START, "prepare")
graph.add_conditional_edges("prepare", route_after_prepare)
graph.add_edge("match_jobs", "finalize")
graph.add_edge("finalize", END)
graph=StateGraph(AgentState)
graph.add_node("prepare",prepare_context)
graph.add_node("match_jobs",match_jobs)
graph.add_edge(START,"prepare")
graph.add_conditional_edges("prepare",route_after_prepare)
graph.add_edge("match_jobs",END)
return graph.compile()
def get_graph():
"""Return the cached compiled graph, building it on first use."""
global _graph
if _graph is None:
_graph = build_graph()
_graph=build_graph()
logger.info("langgraph compiled")
return _graph
async def init_agent():
"""Warm the compiled graph. LLM init stays on llm_setup.init_llm()."""
get_graph()
async def close_agent():
"""Drop the cached graph."""
global _graph
_graph = None
_graph=None
logger.info("agent graph closed")

72
backend/agent/decorators.py
View File

@ -11,45 +11,41 @@ import uuid
def normalize_job_posts(job_posts) -> list[dict]:
"""Keep only dict items with an id field; stringify ids for the LLM."""
if not job_posts:
return []
normalized: list[dict] = []
normalized=[]
for item in job_posts:
if not isinstance(item, dict):
if not isinstance(item,dict):
continue
job_id = item.get("id")
job_id=item.get("id")
if job_id is None:
continue
normalized.append(
{
"id": str(job_id),
"title": item.get("title") or "",
"description": item.get("description") or "",
"post_text": item.get("post_text") or "",
"requirements": item.get("requirements") or [],
"optional_skills": item.get("optional_skills") or [],
"location": item.get("location") or "",
"employment_type": item.get("employment_type") or "",
}
)
normalized.append({
"id":str(job_id),
"title":item.get("title") or "",
"description":item.get("description") or "",
"post_text":item.get("post_text") or "",
"requirements":item.get("requirements") or [],
"optional_skills":item.get("optional_skills") or [],
"location":item.get("location") or "",
"employment_type":item.get("employment_type") or "",
})
return normalized
def parse_match_response(data, allowed_ids) -> tuple[list[str], str, str]:
"""Filter model JSON ids to the allowed job-post set."""
if not isinstance(data, dict):
def parse_match_response(data,allowed_ids) -> tuple[list[str],str,str,str]:
if not isinstance(data,dict):
raise RuntimeError(f"model did not return a JSON object: {data!r}")
allowed = set(allowed_ids or [])
raw_ids = data.get("suggested_job_post_ids") or []
if not isinstance(raw_ids, list):
raw_ids = []
allowed=set(allowed_ids or [])
raw_ids=data.get("suggested_job_post_ids") or []
if not isinstance(raw_ids,list):
raw_ids=[]
suggested: list[str] = []
seen: set[str] = set()
suggested=[]
seen=set()
for raw_id in raw_ids:
job_id = str(raw_id).strip()
job_id=str(raw_id).strip()
if not job_id or job_id not in allowed or job_id in seen:
continue
try:
@ -59,18 +55,18 @@ def parse_match_response(data, allowed_ids) -> tuple[list[str], str, str]:
seen.add(job_id)
suggested.append(job_id)
summary = data.get("summary")
if not isinstance(summary, str):
summary = ""
summary=data.get("summary")
if not isinstance(summary,str):
summary=""
reasoning = data.get("reasoning")
if isinstance(reasoning, list):
reasoning = "\n".join(str(item) for item in reasoning)
if not isinstance(reasoning, str):
reasoning = ""
reasoning=data.get("reasoning")
if isinstance(reasoning,list):
reasoning="\n".join(str(item) for item in reasoning)
if not isinstance(reasoning,str):
reasoning=""
experience = data.get("experience")
if not isinstance(experience, str):
experience = ""
experience=data.get("experience")
if not isinstance(experience,str):
experience=""
return suggested, summary.strip(), reasoning.strip(), experience.strip()
return suggested,summary.strip(),reasoning.strip(),experience.strip()

17
backend/agent/execute_agent.py
View File

@ -9,14 +9,11 @@ from agent.agent_setup import get_graph
from agent.serializers import serialize_agent_result
async def run_agent(*, subject="", resume_text="", job_posts=None) -> dict:
"""Run the default graph and return a serialized result dict."""
final_state = await get_graph().ainvoke(
{
"subject": subject or "",
"resume_text": resume_text or "",
"job_posts": job_posts or [],
"status": "pending",
}
)
async def run_agent(*,subject="",resume_text="",job_posts=None) -> dict:
final_state=await get_graph().ainvoke({
"subject":subject or "",
"resume_text":resume_text or "",
"job_posts":job_posts or [],
"status":"pending",
})
return serialize_agent_result(final_state)

24
backend/agent/models.py
View File

@ -5,18 +5,16 @@ Pure module: no FastAPI imports and no HTTPException.
from __future__ import annotations
from typing import Literal, TypedDict
from typing import Literal,TypedDict
class AgentState(TypedDict, total=False):
"""Shared state passed between graph nodes."""
subject: str
resume_text: str
experience: str
job_posts: list[dict]
suggested_job_post_ids: list[str]
summary: str
reasoning: str
error: str
status: Literal["pending", "ready", "matched", "skipped", "failed"]
class AgentState(TypedDict,total=False):
subject:str
resume_text:str
experience:str
job_posts:list[dict]
suggested_job_post_ids:list[str]
summary:str
reasoning:str
error:str
status:Literal["pending","ready","matched","skipped","failed"]

15
backend/agent/serializers.py
View File

@ -6,13 +6,12 @@ Pure module: no FastAPI imports and no HTTPException.
from __future__ import annotations
def serialize_agent_result(state: dict) -> dict:
"""Plain dict for services/serializers — no ORM objects."""
def serialize_agent_result(state:dict) -> dict:
return {
"suggested_job_post_ids": state.get("suggested_job_post_ids") or [],
"summary": state.get("summary") or "",
"reasoning": state.get("reasoning") or "",
"experience": state.get("experience") or "",
"status": state.get("status") or "failed",
"error": state.get("error") or "",
"suggested_job_post_ids":state.get("suggested_job_post_ids") or [],
"summary":state.get("summary") or "",
"reasoning":state.get("reasoning") or "",
"experience":state.get("experience") or "",
"status":state.get("status") or "failed",
"error":state.get("error") or "",
}

90
backend/agent/views.py
View File

@ -11,84 +11,58 @@ from typing import Literal
from langgraph.graph import END
from agent.decorators import normalize_job_posts, parse_match_response
from agent.decorators import normalize_job_posts,parse_match_response
from agent.models import AgentState
from agent.prompt import prompt, user_prompt
from agent.prompt import prompt,user_prompt
from llm_setup import llm_call
logger = logging.getLogger("agent")
logger=logging.getLogger("agent")
async def prepare_context(state: AgentState) -> dict:
"""Validate inputs and decide whether matching should run."""
subject = (state.get("subject") or "").strip()
resume_text = (state.get("resume_text") or "").strip()
job_posts = normalize_job_posts(state.get("job_posts"))
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": [],
"summary": "",
"reasoning": "",
}
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": [],
"summary": "",
"reasoning": "",
}
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": "",
"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":
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:
"""Ask the LLM (via llm_setup.llm_call) to map the candidate to job posts."""
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)
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,
"status":"matched",
"suggested_job_post_ids":suggested,
"summary":summary,
"reasoning":reasoning,
"experience":experience,
}
except Exception as exc:
except Exception as e:
logger.exception("agent match_jobs failed")
return {
"status": "failed",
"error": str(exc),
"suggested_job_post_ids": [],
"summary": "",
"reasoning": "",
"experience": "",
"status":"failed",
"error":str(e),
"suggested_job_post_ids":[],
"summary":"",
"reasoning":"",
"experience":"",
}
async def finalize(state: AgentState) -> dict:
"""Normalize terminal state for callers."""
return {
"suggested_job_post_ids": state.get("suggested_job_post_ids") or [],
"summary": state.get("summary") or "",
"reasoning": state.get("reasoning") or "",
"experience": state.get("experience") or "",
"status": state.get("status") or "failed",
"error": state.get("error") or "",
}

54
backend/inbox/tasks.py
View File

@ -3,19 +3,19 @@
from __future__ import annotations
import logging
from datetime import datetime, timezone
from datetime import datetime,timezone
from agent.execute_agent import run_agent
from db_setup import session_scope
from inbox.models import Inbox_Messages
from inbox.plugins import extract_resume_text
from job.job_post.models import JobPosts
from job.job_post.serializers import serialize_job_post
from taskiq_management.broker_setup import MAX_RETRIES, RETRY_DELAY, broker
from taskiq_management.broker_setup import MAX_RETRIES,RETRY_DELAY,broker
from taskiq_management.middleware import PermanentTaskError
logger=logging.getLogger("inbox.tasks")
_DONE_STATUSES=frozenset({"matched","skipped","no_text","failed","dlq"})
_DONE=frozenset({"matched","skipped","no_text","failed","dlq"})
@broker.task(
@ -27,60 +27,41 @@ _DONE_STATUSES=frozenset({"matched","skipped","no_text","failed","dlq"})
async def match_inbox_message(record_id:str,force:bool=False) -> dict:
if not record_id or not str(record_id).strip():
raise PermanentTaskError("record_id is required")
record_id=str(record_id).strip()
async with session_scope() as session:
row=await Inbox_Messages.get_inbox_message_by_id(session,record_id)
if not row:
raise PermanentTaskError(f"inbox message {record_id} not found")
if not force and row.match_status in _DONE_STATUSES:
logger.info("skip %s — already %s",record_id,row.match_status)
if not force and row.match_status in _DONE:
return {"status":row.match_status,"skipped":True}
if not row.attachment or not row.file_path:
raise PermanentTaskError("message has no attachment to match")
paths=[p.strip() for p in row.file_path.split(",") if p.strip()]
subject=row.message_subject or ""
row.match_status="processing"
row.match_error=None
row.matched_at=datetime.now(timezone.utc)
session.add(row)
await session.commit()
await session.refresh(row)
paths=[p.strip() for p in (row.file_path or "").split(",") if p.strip()]
subject=row.message_subject or ""
posts=await JobPosts.get_active_job_posts(session)
job_posts=[serialize_job_post(p) for p in posts]
text,extract_err=await extract_resume_text(paths)
if not text:
async with session_scope() as session:
await Inbox_Messages.set_match_result(
session,
record_id,
status="no_text",
error=extract_err or "no text extracted",
session,record_id,status="no_text",error=extract_err or "no text extracted",
)
return {"status":"no_text","error":extract_err}
from agent.execute_agent import run_agent
async with session_scope() as session:
posts=await JobPosts.get_active_job_posts(session)
job_posts=[serialize_job_post(p) for p in posts]
try:
result=await run_agent(subject=subject,resume_text=text,job_posts=job_posts)
except Exception as exc:
logger.exception("agent failed for %s",record_id)
raise RuntimeError(f"agent matching failed: {exc}") from exc
result=await run_agent(subject=subject,resume_text=text,job_posts=job_posts)
status=result.get("status") or "failed"
error=result.get("error") or ""
if status=="failed":
raise RuntimeError(error or "agent returned failed status")
logger.info("agent result: %s,%s",result,result.get("experience"))
raise RuntimeError(result.get("error") or "agent returned failed status")
async with session_scope() as session:
await Inbox_Messages.set_match_result(
session,
@ -91,11 +72,6 @@ async def match_inbox_message(record_id:str,force:bool=False) -> dict:
summary=result.get("summary") or "",
reasoning=result.get("reasoning") or "",
status=status,
error=error,
error=result.get("error") or "",
)
logger.info("matched inbox %s status=%s ids=%s",record_id,status,result.get("suggested_job_post_ids"))
return {
"status":status,
"suggested_job_post_ids":result.get("suggested_job_post_ids") or [],
}
return {"status":status,"suggested_job_post_ids":result.get("suggested_job_post_ids") or []}

13
backend/inbox/views.py
View File

@ -51,7 +51,7 @@ class Email:
except Exception as e:
raise HTTPException(status_code=500,detail=str(e))
async def get_email_by_id(self,message_id,test_on=True):
async def get_email_by_id(self,message_id):
async with httpx.AsyncClient() as client:
try:
response=await client.get(f"{self.get_url}/emails/{message_id}",
@ -63,8 +63,6 @@ class Email:
row,new_user_email=await Inbox_Messages.insert_email(session=self.session,email_data=data,file_path=re_create_file)
if row.attachment and row.file_path and row.match_status is None:
self.pending_match_ids.append(str(row.id))
if test_on:
return data
if new_user_email:
self.pending_confirmation_emails.append(new_user_email)
return data
@ -131,17 +129,14 @@ class Email:
return task_ids
async def send_account_setup(self,emails):
"""Mail a confirmation link per freshly created sender. Best effort: a failed
mail must not fail a fetch whose messages are already stored."""
results=[]
for email in emails or []:
try:
status=await request_email_confirmation(email)
except Exception as exc:
logger.warning("confirmation request failed for %s: %s",email,exc)
results.append({"email":email,"sent":status==200})
except Exception as e:
logger.warning("confirmation request failed for %s: %s",email,e)
results.append({"email":email,"sent":False})
continue
results.append({"email":email,"sent":status==200})
return results
async def count_inbox_messages(self,search=None,isread:bool=True,application_status:Candidate_application_Status=Candidate_application_Status.CLOSED):

36
backend/job/app.py
View File

@ -2,7 +2,7 @@ from fastapi import APIRouter,Depends,Query
from fastapi.responses import JSONResponse
from fastapi import HTTPException
from db_setup import get_session
from job.candidate.views import FileRead,CandidateView
from job.candidate.views import FileRead
from sqlalchemy.ext.asyncio import AsyncSession
from users.permissions import PermissionTag, require_permission
from job.job_post.views import JobPost,JobPostCreate
@ -102,21 +102,21 @@ async def buffer_channels(
except Exception as e:
raise HTTPException(status_code=500,detail=str(e))
@router.get("/candidate/fetch")
async def fetch_candidate(
user_id:str=Query(None),
current_user: dict = Depends(require_permission(PermissionTag.CANDIDATES_VIEW)),
session: AsyncSession = Depends(get_session),
):
try:
service=CandidateView(session=session)
if user_id:
data=await service.get_candidate(user_id=user_id)
else:
data=await service.get_candidate()
return JSONResponse(content={"data":data,"status_code":200})
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500,detail=str(e))
# @router.get("/candidate/fetch")
# async def fetch_candidate(
# user_id:str=Query(None),
# current_user: dict = Depends(require_permission(PermissionTag.CANDIDATES_VIEW)),
# session: AsyncSession = Depends(get_session),
# ):
# try:
# service=CandidateView(session=session)
# if user_id:
# data=await service.get_candidate(user_id=user_id)
# else:
# data=await service.get_candidate()
# return JSONResponse(content={"data":data,"status_code":200})
# except HTTPException:
# raise
# except Exception as e:
# raise HTTPException(status_code=500,detail=str(e))

16
backend/job/candidate/views.py
View File

@ -68,12 +68,12 @@ class FileRead:
# get_subject=Inbox_Messages.candidate_x_inbox(self.session,self.candidate_id)
# get_file=
class CandidateView:
def __init__(self,session:AsyncSession):
self.session=session
# class CandidateView:
# def __init__(self,session:AsyncSession):
# self.session=session
async def get_candidate(self,user_id=None):
try:
call_func=Inbox_Messages.get_candidate_profile(user_id=user_id)
except Exception as e:
raise HTTPException(status_code=500,detail=str(e))
# async def get_candidate(self,user_id=None):
# try:
# call_func=Inbox_Messages.get_candidate_profile(user_id=user_id)
# except Exception as e:
# raise HTTPException(status_code=500,detail=str(e))