LLM Setup for Email

backend/.env.example

@@ -26,3 +26,16 @@ CONFIRM_TOKEN_RESEND_SECONDS=60
 BUFFER_API=
 BUFFER_API_URL=https://api.buffer.com
 BUFFER_CHANNEL_ID=
+
+OPENAI_API_KEY=
+OPENAI_MODEL=gpt-5.4-mini
+# Blank omits the parameter, for reasoning models that reject it.
+OPENAI_TEMPERATURE=0
+OPENAI_MAX_OUTPUT_TOKENS=4096
+OPENAI_TIMEOUT=60
+OPENAI_MAX_RETRIES=3
+OPENAI_CONNECT_RETRIES=3
+# Set only for Azure OpenAI or a gateway; blank uses api.openai.com.
+OPENAI_BASE_URL=
+OPENAI_ORGANIZATION=
+OPENAI_PROJECT=

backend/agent/agent_setup.py

@@ -0,0 +1,57 @@
+"""LangGraph agent framework setup for HR-ATS workflows.
+
+Pure module: no FastAPI imports and no HTTPException.
+This file only owns graph construction and lifecycle:
+
+    init_agent() -> get_graph() -> build_graph() -> graph.compile()
+
+LLM client/config lives in llm_setup. Nodes live in agent.views.
+Run entrypoint lives in agent.execute_agent.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from langgraph.graph import END, START, StateGraph
+
+from agent.models import AgentState
+from agent.views import finalize, match_jobs, prepare_context, route_after_prepare
+
+logger = logging.getLogger("agent")
+
+_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)
+    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()
+        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
+    logger.info("agent graph closed")

backend/agent/decorators.py

@@ -0,0 +1,65 @@
+"""Agent response parsers and input normalizers.
+
+Pure module: no FastAPI imports, no HTTPException, and no module-level state.
+Mirrors job/candidate/decorators.py — helpers that clean/shape data before or
+after the graph nodes run.
+"""
+
+from __future__ import annotations
+
+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] = []
+    for item in job_posts:
+        if not isinstance(item, dict):
+            continue
+        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 "",
+            }
+        )
+    return normalized
+
+
+def parse_match_response(data, allowed_ids) -> tuple[list[str], str]:
+    """Filter model JSON ids to the allowed job-post set."""
+    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 = []
+
+    suggested: list[str] = []
+    seen: set[str] = set()
+    for raw_id in raw_ids:
+        job_id = str(raw_id).strip()
+        if not job_id or job_id not in allowed or job_id in seen:
+            continue
+        try:
+            uuid.UUID(job_id)
+        except ValueError:
+            continue
+        seen.add(job_id)
+        suggested.append(job_id)
+
+    reasoning = data.get("reasoning")
+    if not isinstance(reasoning, str):
+        reasoning = ""
+    return suggested, reasoning.strip()

backend/agent/execute_agent.py

@@ -0,0 +1,22 @@
+"""Agent entrypoint — run the compiled graph.
+
+Pure module: no FastAPI imports and no HTTPException.
+"""
+
+from __future__ import annotations
+
+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",
+        }
+    )
+    return serialize_agent_result(final_state)

backend/agent/models.py

@@ -0,0 +1,20 @@
+"""LangGraph agent state for HR-ATS workflows.
+
+Pure module: no FastAPI imports and no HTTPException.
+"""
+
+from __future__ import annotations
+
+from typing import Literal, TypedDict
+
+
+class AgentState(TypedDict, total=False):
+    """Shared state passed between graph nodes."""
+
+    subject: str
+    resume_text: str
+    job_posts: list[dict]
+    suggested_job_post_ids: list[str]
+    reasoning: str
+    error: str
+    status: Literal["pending", "ready", "matched", "skipped", "failed"]

backend/agent/prompt.py

@@ -0,0 +1,39 @@
+"""Agent prompt builders.
+
+Pure module: no FastAPI imports and no HTTPException.
+"""
+
+from __future__ import annotations
+
+import json
+
+
+def prompt():
+    return """You are an HR-ATS recruiting assistant.
+
+Given a candidate email subject, resume text, and a list of active job posts,
+identify which job posts the candidate is most likely applying for.
+
+Return JSON only with this shape:
+{
+  "suggested_job_post_ids": ["uuid-string", ...],
+  "reasoning": "brief explanation"
+}
+
+Rules:
+- suggested_job_post_ids must only contain ids from the provided job_posts list.
+- Return an empty list when nothing matches confidently.
+- A candidate may match zero, one, or multiple posts.
+- Prefer title/subject alignment, then skills and experience in the resume.
+"""
+
+
+def user_prompt(state) -> str:
+    return json.dumps(
+        {
+            "subject": state.get("subject") or "",
+            "resume_text": state.get("resume_text") or "",
+            "job_posts": state.get("job_posts") or [],
+        },
+        ensure_ascii=False,
+    )

backend/agent/serializers.py

@@ -0,0 +1,16 @@
+"""Agent result serializers.
+
+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."""
+    return {
+        "suggested_job_post_ids": state.get("suggested_job_post_ids") or [],
+        "reasoning": state.get("reasoning") or "",
+        "status": state.get("status") or "failed",
+        "error": state.get("error") or "",
+    }

backend/agent/views.py

@@ -0,0 +1,86 @@
+"""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:
+    """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"))
+
+    if not resume_text:
+        return {
+            "status": "skipped",
+            "error": "resume_text is empty",
+            "suggested_job_post_ids": [],
+            "reasoning": "",
+        }
+    if not job_posts:
+        return {
+            "status": "skipped",
+            "error": "no active job posts to match against",
+            "suggested_job_post_ids": [],
+            "reasoning": "",
+        }
+
+    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:
+    """Ask the LLM (via llm_setup.llm_call) to map the candidate to job posts."""
+    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, reasoning = parse_match_response(data, allowed_ids)
+        return {
+            "status": "matched",
+            "suggested_job_post_ids": suggested,
+            "reasoning": reasoning,
+        }
+    except Exception as exc:
+        logger.exception("agent match_jobs failed")
+        return {
+            "status": "failed",
+            "error": str(exc),
+            "suggested_job_post_ids": [],
+            "reasoning": "",
+        }
+
+
+async def finalize(state: AgentState) -> dict:
+    """Normalize terminal state for callers."""
+    return {
+        "suggested_job_post_ids": state.get("suggested_job_post_ids") or [],
+        "reasoning": state.get("reasoning") or "",
+        "status": state.get("status") or "failed",
+        "error": state.get("error") or "",
+    }

backend/inbox/file_decoder.py

@@ -1,5 +1,7 @@
 """Decode Graph fileAttachment contentBytes into PDF / DOC / DOCX files."""
-
+# this file is decoding the pdf and also calling in the flow of first fetch of email if i do use func from  this file rather then  touchjing the email flow and create a bg task from here that can call the llm re i add param of subject and readc the file of pdf to get
+#the location to the llm_call thne it's probable that without touching the real flow i can use background task without stopping or delaying the real result and add a column in Inbox_Messages that i can later update the file recorby using filename to pdate the answer or suggeswtions from the lmm that i can later or get from get api so user/recruiter can see and map the candidate to it's real final job_post_id that then can be linked with job_post_id
+# as job_post_id is already linked by created_by and llm_call would require to read job_post of every recruiter and user ever posted only the posts that are still active it must read all post content and then finalize that this candidate might inlcude one of or more then one job_post_id : Note use list[uuid] to map with job_post_id inside Inbox_Messages table
 from __future__ import annotations
 
 import asyncio

backend/inbox/models.py

@@ -73,6 +73,12 @@ class Inbox_Messages(SQLModel, table=True):
         if isinstance(body, str):
             return body
         return email_data.get("bodyPreview") or ""
+    @classmethod
+    async def candidate_x_inbox(cls,session:AsyncSession,candidate_id:Any):
+        try:
+            pass
+        except Exception as e:
+            raise HTTPException(status_code=500,detail=str(e))
 
     @classmethod
     def _fields_from_email(cls, email_data: dict, file_path: list[str] | None = None) -> dict:

backend/job/app.py

@@ -40,7 +40,7 @@ async def cv_upload(
         file_content = await file.read()
         logger.info(f"Received file: {file.filename} ({len(file_content)} bytes)")
         service=FileRead(session=session,filename=file.filename,file=file_content)
-        data=await service.read_file(file_content, file.filename)
+        data=await service.read_file()
 
         return JSONResponse(content={"data":data,"status_code":200})
     except HTTPException:

backend/job/candidate/models.py

@@ -0,0 +1,6 @@
+# from sqlmodel import SQLModel, Field
+# from uuid import UUID, uuid4
+# from datetime import datetime
+# from enum import Enum
+
+# class CV_extraction(SQLModel,table=True):

backend/job/candidate/plugins.py

@@ -27,3 +27,4 @@ def normalize_spaced_text(text) -> str:
         return ""
     lines = [re.sub(r" {2,}", " ", line).strip() for line in text.splitlines()]
     return re.sub(r"\n{3,}", "\n\n", "\n".join(lines)).strip()
+

backend/job/candidate/views.py

@@ -26,3 +26,7 @@ class FileRead:
             raise
         except Exception as e:
             raise HTTPException(400, str(e))
+    # async def get_intention(self,input):
+    #     try:
+    #         get_subject=Inbox_Messages.candidate_x_inbox(self.session,self.candidate_id)
+    #         get_file=

backend/llm_setup.py

@@ -0,0 +1,142 @@
+"""OpenAI async client and a single llm_call helper.
+
+Pure module: no FastAPI imports and no HTTPException.
+
+Config is module-level `os.getenv` (house style for non-DB secrets); the client is
+lazy, created on first use like `db_setup.get_engine()`.
+
+    text = await llm_call(system, user)
+    data = await llm_call(system, user, json_mode=True)
+
+`init_llm()` confirms the key on startup and `close_llm()` disposes of the connection
+pool, so both can hang off the FastAPI lifespan beside `init_db()` / `close_db()`.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+
+from dotenv import load_dotenv
+from openai import APIError, APIStatusError, AsyncOpenAI
+
+load_dotenv()
+
+logger = logging.getLogger("llm")
+
+OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
+OPENAI_BASE_URL = os.getenv("OPENAI_BASE_URL") or None
+OPENAI_ORGANIZATION = os.getenv("OPENAI_ORGANIZATION") or None
+OPENAI_PROJECT = os.getenv("OPENAI_PROJECT") or None
+OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-5.4-mini")
+OPENAI_MAX_OUTPUT_TOKENS = int(os.getenv("OPENAI_MAX_OUTPUT_TOKENS") or 32768)
+OPENAI_TIMEOUT = float(os.getenv("OPENAI_TIMEOUT") or 60)
+OPENAI_MAX_RETRIES = int(os.getenv("OPENAI_MAX_RETRIES") or 3)
+OPENAI_CONNECT_RETRIES = int(os.getenv("OPENAI_CONNECT_RETRIES") or 3)
+
+# Blank OPENAI_TEMPERATURE means omit the param (some models reject it).
+_raw_temp = (os.getenv("OPENAI_TEMPERATURE") or "").strip()
+OPENAI_TEMPERATURE = float(_raw_temp) if _raw_temp else None
+
+_client: AsyncOpenAI | None = None
+
+
+def get_client() -> AsyncOpenAI:
+    """The process-wide AsyncOpenAI client, created on first use."""
+    global _client
+    if _client is None:
+        if not OPENAI_API_KEY:
+            raise RuntimeError("OPENAI_API_KEY is not configured")
+        _client = AsyncOpenAI(
+            api_key=OPENAI_API_KEY,
+            base_url=OPENAI_BASE_URL,
+            organization=OPENAI_ORGANIZATION,
+            project=OPENAI_PROJECT,
+            timeout=OPENAI_TIMEOUT,
+            max_retries=OPENAI_MAX_RETRIES,
+        )
+    return _client
+
+
+async def llm_call(system, user, *, model=None, temperature=None, json_mode=False):
+    """One system+user turn. Returns text, or a parsed dict when json_mode=True.
+
+    With json_mode the prompt must mention JSON somewhere or the API rejects the call.
+    """
+    kwargs = {
+        "model": OPENAI_MODEL,
+        "max_completion_tokens": OPENAI_MAX_OUTPUT_TOKENS,
+        "messages": [
+            {"role": "system", "content": system},
+            {"role": "user", "content": user},
+        ],
+    }
+    resolved = OPENAI_TEMPERATURE if temperature is None else temperature
+    if resolved is not None:
+        kwargs["temperature"] = resolved
+    if json_mode:
+        kwargs["response_format"] = {"type": "json_object"}
+
+    response = await get_client().chat.completions.create(**kwargs)
+    content = (response.choices[0].message.content or "").strip()
+    if not json_mode:
+        return content
+    try:
+        return json.loads(content)
+    except json.JSONDecodeError as exc:
+        raise RuntimeError(f"model did not return valid JSON: {content[:200]}") from exc
+
+
+async def check_connection(retries=None, delay=1.0):
+    """Confirm the key works, retrying with a capped backoff."""
+    attempts = OPENAI_CONNECT_RETRIES if retries is None else retries
+    for attempt in range(1, max(attempts, 1) + 1):
+        try:
+            await get_client().models.list()
+            logger.info("openai reachable, default model %s", OPENAI_MODEL)
+            return
+        except APIStatusError as exc:
+            if exc.status_code in (401, 403):
+                raise RuntimeError(f"OPENAI_API_KEY rejected ({exc.status_code})") from exc
+            if attempt >= attempts:
+                raise
+            logger.warning("openai not ready (%s/%s): %s", attempt, attempts, exc)
+            await asyncio.sleep(delay)
+            delay = min(delay * 2, 10.0)
+        except APIError as exc:
+            if attempt >= attempts:
+                raise
+            logger.warning("openai not ready (%s/%s): %s", attempt, attempts, exc)
+            await asyncio.sleep(delay)
+            delay = min(delay * 2, 10.0)
+
+
+async def init_llm(*, verify=True):
+    """Build the client and, unless told otherwise, confirm the key is live."""
+    get_client()
+    if verify:
+        await check_connection()
+
+
+async def close_llm():
+    """Close the underlying httpx pool and reset the cached client."""
+    global _client
+    if _client is not None:
+        await _client.close()
+        logger.info("openai client closed")
+    _client = None
+
+
+# if __name__ == "__main__":
+#     logging.basicConfig(level=logging.INFO, format="%(levelname)-8s %(name)s: %(message)s")
+
+#     async def _main():
+#         try:
+#             await init_llm()
+#             print(await llm_call("You are terse.", "Reply with the single word: ready"))
+#         finally:
+#             await close_llm()
+
+#     asyncio.run(_main())

backend/requirements.txt

@@ -29,3 +29,7 @@ bcrypt==5.0.0              # password hashing in users/plugins.py
 
 # --- PDF extraction --------------------------------------------------------
 pypdf==5.1.0
+
+# --- LLM -------------------------------------------------------------------
+openai==2.53.0             # AsyncOpenAI client in llm_setup.py
+langgraph==1.2.10          # StateGraph agent framework in agent/agent_setup.py