From 4d340fd6d5b2a133fb42a387333fcc077c45a2eb Mon Sep 17 00:00:00 2001 From: "ahmed.mujtaba" Date: Mon, 10 Aug 2026 16:24:47 +0500 Subject: [PATCH] agent coide simplifgication --- backend/agent/agent_setup.py | 30 +++++------- backend/agent/decorators.py | 72 +++++++++++++-------------- backend/agent/execute_agent.py | 17 +++---- backend/agent/models.py | 24 +++++---- backend/agent/serializers.py | 15 +++--- backend/agent/views.py | 90 ++++++++++++---------------------- backend/inbox/tasks.py | 54 ++++++-------------- backend/inbox/views.py | 13 ++--- backend/job/app.py | 36 +++++++------- backend/job/candidate/views.py | 16 +++--- 10 files changed, 148 insertions(+), 219 deletions(-) diff --git a/backend/agent/agent_setup.py b/backend/agent/agent_setup.py index 5fe000a..2bd9bba 100644 --- a/backend/agent/agent_setup.py +++ b/backend/agent/agent_setup.py @@ -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") diff --git a/backend/agent/decorators.py b/backend/agent/decorators.py index 53bd62c..5cdbbeb 100644 --- a/backend/agent/decorators.py +++ b/backend/agent/decorators.py @@ -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() diff --git a/backend/agent/execute_agent.py b/backend/agent/execute_agent.py index e7ef8d0..f7e0c03 100644 --- a/backend/agent/execute_agent.py +++ b/backend/agent/execute_agent.py @@ -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) diff --git a/backend/agent/models.py b/backend/agent/models.py index 51fb99a..01d15de 100644 --- a/backend/agent/models.py +++ b/backend/agent/models.py @@ -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"] diff --git a/backend/agent/serializers.py b/backend/agent/serializers.py index 347f7e0..833aac7 100644 --- a/backend/agent/serializers.py +++ b/backend/agent/serializers.py @@ -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 "", } diff --git a/backend/agent/views.py b/backend/agent/views.py index 8520014..a128e70 100644 --- a/backend/agent/views.py +++ b/backend/agent/views.py @@ -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 "", - } diff --git a/backend/inbox/tasks.py b/backend/inbox/tasks.py index 554b8f6..2d59cc9 100644 --- a/backend/inbox/tasks.py +++ b/backend/inbox/tasks.py @@ -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 []} diff --git a/backend/inbox/views.py b/backend/inbox/views.py index e7d88b1..8c6092d 100644 --- a/backend/inbox/views.py +++ b/backend/inbox/views.py @@ -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): diff --git a/backend/job/app.py b/backend/job/app.py index cdeb38c..682bb75 100644 --- a/backend/job/app.py +++ b/backend/job/app.py @@ -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)) \ No newline at end of file diff --git a/backend/job/candidate/views.py b/backend/job/candidate/views.py index 0307168..6b10aa3 100644 --- a/backend/job/candidate/views.py +++ b/backend/job/candidate/views.py @@ -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)) \ No newline at end of file +# 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)) \ No newline at end of file