← Back to DevBytes

Building a HR Screening Agent with LangChain: Complete Guide

What is an HR Screening Agent?

An HR Screening Agent is an AI-powered system that automates the initial stages of candidate evaluation in the recruitment pipeline. Built with LangChain, this agent can review resumes, compare qualifications against job requirements, ask clarifying questions to candidates, and provide structured recommendations to human recruiters. Think of it as an intelligent first-pass filter that handles the repetitive, high-volume screening work while surfacing the most promising applicants for human review.

The agent leverages large language models (LLMs) combined with tools for document parsing, structured data extraction, and decision logic. Using LangChain's agent framework, you can create a system that reasons about candidate profiles, uses external tools like resume parsers and skills databases, and maintains conversation context across multiple screening interactions.

Why an AI Screening Agent Matters

Recruitment teams face three fundamental challenges that an AI screening agent directly addresses:

Volume Overload

A single job posting can attract hundreds or thousands of applications. Human recruiters typically spend only 6-8 seconds on an initial resume scan, which leads to qualified candidates being overlooked due to rushed reviews. An AI agent processes every application with consistent attention to detail.

Unconscious Bias Reduction

Human screeners bring implicit biases to the review process. While AI systems require careful design to avoid perpetuating bias, a well-configured agent can evaluate candidates based purely on job-relevant criteria, masking demographic indicators and focusing on skills and experience.

Candidate Experience

Applicants often wait weeks for any response. An automated agent can provide immediate acknowledgment, conduct preliminary screening conversations, and keep candidates informed about their status, dramatically improving the candidate experience while reducing the recruiter's administrative burden.

Core Architecture Overview

Our HR screening agent uses LangChain's ReAct (Reasoning + Acting) pattern, combining an LLM for reasoning with specific tools for resume processing and candidate evaluation. The architecture consists of:

Prerequisites and Setup

Install the required dependencies before building the agent:

pip install langchain langchain-openai langchain-community chromadb pypdf unstructured python-dotenv streamlit

Set up your environment variables for API access:

# .env file
OPENAI_API_KEY=your-openai-api-key
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=your-langsmith-key  # optional, for debugging

Step 1: Building the Resume Parser Tool

First, we create a tool that extracts structured information from resume files. This tool wraps LangChain's document loaders and adds custom extraction logic:

from langchain_community.document_loaders import PyPDFLoader, UnstructuredWordDocumentLoader
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from typing import Optional, Dict
from pydantic import BaseModel, Field
import os

class ResumeExtractionInput(BaseModel):
    file_path: str = Field(description="Path to the resume file (PDF or DOCX)")

class ResumeData(BaseModel):
    full_name: str
    email: str
    phone: str
    skills: list[str]
    years_experience: Dict[str, int]
    education: list[Dict[str, str]]
    recent_roles: list[Dict[str, str]]

@tool("resume-parser", args_schema=ResumeExtractionInput)
def parse_resume(file_path: str) -> Dict:
    """
    Parse a resume file and extract structured candidate information.
    Use this tool whenever you need to analyze a candidate's resume.
    """
    # Determine file type and load
    if file_path.endswith('.pdf'):
        loader = PyPDFLoader(file_path)
    elif file_path.endswith('.docx'):
        loader = UnstructuredWordDocumentLoader(file_path)
    else:
        return {"error": "Unsupported file format. Use PDF or DOCX."}
    
    documents = loader.load()
    full_text = "\n".join([doc.page_content for doc in documents])
    
    # Use LLM for structured extraction
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    
    extraction_prompt = f"""
    Extract the following information from this resume text.
    Return ONLY valid JSON, no other text.
    
    Required fields:
    - full_name: string
    - email: string (email address)
    - phone: string (phone number)
    - skills: array of strings (technical and soft skills mentioned)
    - years_experience: object mapping skill areas to years (e.g., {{"python": 5, "management": 3}})
    - education: array of {{"degree": "", "institution": "", "year": ""}}
    - recent_roles: array of {{"title": "", "company": "", "duration": "", "achievements": []}}
    
    Resume text:
    {full_text[:4000]}
    """
    
    response = llm.invoke(extraction_prompt)
    
    # Parse the JSON response
    import json
    try:
        # Clean response in case LLM wraps in markdown
        content = response.content.strip()
        if content.startswith("json"):
            content = content[7:]
        if content.startswith(""):
            content = content[3:]
        if content.endswith(""):
            content = content[:-3]
        return json.loads(content.strip())
    except json.JSONDecodeError:
        return {"error": "Failed to parse resume data", "raw_response": response.content}
    
parse_resume.invoke({"file_path": "candidate_resume.pdf"})

Step 2: Creating the Job Requirements Vector Store

We store job descriptions in a vector database to enable semantic matching between candidate profiles and job requirements. This allows the agent to find relevant jobs for a candidate or find relevant candidates for a job:

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document
import json

# Sample job descriptions
job_descriptions = [
    {
        "title": "Senior Python Developer",
        "department": "Engineering",
        "required_skills": ["Python", "Django", "AWS", "PostgreSQL", "Docker"],
        "min_years": 5,
        "description": "We need a senior Python developer to build scalable backend systems. Experience with Django REST framework, AWS infrastructure, and database optimization required. Must have led technical teams and mentored junior developers.",
        "nice_to_have": ["Kubernetes", "Machine Learning", "GraphQL"]
    },
    {
        "title": "Product Manager",
        "department": "Product",
        "required_skills": ["Product Strategy", "Data Analysis", "User Research", "Agile"],
        "min_years": 3,
        "description": "Seeking a product manager to own the roadmap for our analytics platform. You'll conduct user research, define requirements, and work closely with engineering and design teams. Strong analytical skills and experience with A/B testing required.",
        "nice_to_have": ["SQL", "Figma", "MBA"]
    }
]

# Create embeddings and store
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)

documents = []
for job in job_descriptions:
    # Create a rich document with metadata
    doc_text = f"""
    Title: {job['title']}
    Department: {job['department']}
    Required Skills: {', '.join(job['required_skills'])}
    Minimum Years Experience: {job['min_years']}
    Nice to Have: {', '.join(job.get('nice_to_have', []))}
    Description: {job['description']}
    """
    doc = Document(
        page_content=doc_text,
        metadata={
            "title": job['title'],
            "department": job['department'],
            "required_skills": json.dumps(job['required_skills']),
            "min_years": job['min_years']
        }
    )
    documents.append(doc)

# Split and store
splits = text_splitter.split_documents(documents)
vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=embeddings,
    collection_name="job_descriptions",
    persist_directory="./job_store"
)

# Create a retriever tool
@tool("job-search")
def search_jobs(query: str) -> str:
    """
    Search for relevant job descriptions matching the query.
    Use this to find jobs that match a candidate's profile or to find specific job requirements.
    """
    retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
    results = retriever.invoke(query)
    
    output = []
    for doc in results:
        output.append({
            "title": doc.metadata.get("title"),
            "department": doc.metadata.get("department"),
            "relevance_score": round(doc.metadata.get("score", 0), 3) if hasattr(doc, 'metadata') else "N/A",
            "content": doc.page_content[:300]
        })
    
    return json.dumps(output, indent=2)

Step 3: Building the Screening Logic Tool

The screening tool compares structured candidate data against job requirements and produces a scored evaluation. This is the core decision-support component:

from langchain.tools import tool
from typing import Dict
import json

@tool("screening-evaluator")
def evaluate_candidate(candidate_data: Dict, job_id: str) -> str:
    """
    Evaluate a candidate's fit for a specific job based on structured data.
    Input candidate_data should be the parsed resume dictionary.
    Input job_id should be the job title or identifier to evaluate against.
    """
    # Retrieve job requirements from vector store
    retriever = vectorstore.as_retriever(search_kwargs={"k": 1})
    job_docs = retriever.invoke(job_id)
    
    if not job_docs:
        return json.dumps({"error": f"Job '{job_id}' not found in database"})
    
    job_doc = job_docs[0]
    required_skills = json.loads(job_doc.metadata.get("required_skills", "[]"))
    min_years = job_doc.metadata.get("min_years", 0)
    
    candidate_skills = [s.lower() for s in candidate_data.get("skills", [])]
    required_skills_lower = [s.lower() for s in required_skills]
    
    # Calculate skill match
    matched_skills = [s for s in required_skills_lower if s in candidate_skills]
    missing_skills = [s for s in required_skills_lower if s not in candidate_skills]
    skill_match_percentage = len(matched_skills) / len(required_skills_lower) * 100 if required_skills_lower else 0
    
    # Calculate experience match
    total_relevant_years = sum(
        candidate_data.get("years_experience", {}).get(skill, 0)
        for skill in matched_skills
    )
    experience_score = min(total_relevant_years / (min_years * len(matched_skills)) * 100 if matched_skills else 0, 100)
    
    # Composite score
    composite_score = (skill_match_percentage * 0.6) + (experience_score * 0.4)
    
    result = {
        "job_title": job_doc.metadata.get("title"),
        "composite_score": round(composite_score, 1),
        "skill_match": f"{len(matched_skills)}/{len(required_skills)}",
        "matched_skills": matched_skills,
        "missing_skills": missing_skills,
        "relevant_experience_years": total_relevant_years,
        "recommendation": "STRONG_HIRE" if composite_score > 80 else "POTENTIAL" if composite_score > 60 else "SCREEN_OUT"
    }
    
    return json.dumps(result, indent=2)

Step 4: Assembling the LangChain Agent

Now we combine all tools into a ReAct agent using LangChain's agent framework. The agent can reason about when to parse resumes, search jobs, and evaluate candidates:

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain.prompts import PromptTemplate
from langchain.memory import ConversationBufferMemory
from langchain.callbacks import StreamlitCallbackHandler
from typing import List

# Define the agent's tools
tools = [parse_resume, search_jobs, evaluate_candidate]

# Create the LLM with reasoning capabilities
llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0.2,
    max_tokens=2000
)

# Custom system prompt for HR screening context
system_prompt = """You are an expert HR Screening Agent. Your role is to help recruiters evaluate candidates efficiently and fairly.

You have access to the following tools:
- resume-parser: Extract structured data from resume files (PDF/DOCX)
- job-search: Find relevant job descriptions from the company's database
- screening-evaluator: Compare a candidate against job requirements and produce a scored evaluation

Workflow Guidelines:
1. When given a resume file path, ALWAYS parse it first using resume-parser
2. When asked to evaluate a candidate, first parse their resume, then search for the relevant job, then run the evaluation
3. Present results clearly with scores, matched skills, and missing skills
4. For borderline cases (scores 55-65), suggest areas where the candidate could improve
5. NEVER make decisions based on gender, age, ethnicity, or other protected characteristics
6. Flag candidates with strong potential even if they don't match 100% — highlight transferable skills

Output Format:
Always structure your response with:
- Candidate Summary
- Job Match Analysis  
- Detailed Scoring Breakdown
- Recommendation with reasoning
- Next Steps for the recruiter

Be concise but thorough. Your goal is to save the recruiter time while ensuring no qualified candidate is overlooked."""

# Create the prompt template
prompt = PromptTemplate.from_template(
    """You are an HR Screening Agent.
    
{system_prompt}

You have access to the following tools:
{tools}

Use the following format for your reasoning:
Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Begin!

Question: {input}
Thought: {agent_scratchpad}"""
)

# Create the agent
agent = create_react_agent(
    llm=llm,
    tools=tools,
    prompt=prompt
)

# Add memory for conversation context
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

# Create the executor
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    max_iterations=8,
    max_execution_time=120,
    handle_parsing_errors=True,
    verbose=True
)

# Test the agent
response = agent_executor.invoke({
    "input": "Evaluate candidate John Smith from the file 'resumes/john_smith.pdf' for the Senior Python Developer position. Give me a complete screening report.",
    "system_prompt": system_prompt
})

print(response["output"])

Step 5: Adding a Candidate Chat Interface

For interactive screening, we add a conversational component that allows the agent to ask follow-up questions directly to candidates:

from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.schema import SystemMessage, HumanMessage
import streamlit as st

class ScreeningConversation:
    """
    Manages a screening conversation with a candidate.
    The agent asks clarifying questions based on resume gaps.
    """
    
    def __init__(self, candidate_data: Dict, job_data: Dict):
        self.candidate_data = candidate_data
        self.job_data = job_data
        self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
        self.conversation_history = []
        
        # Identify gaps to explore
        self.gaps = self._identify_gaps()
    
    def _identify_gaps(self) -> List[str]:
        """Find areas where resume is unclear or missing details"""
        required = [s.lower() for s in self.job_data.get("required_skills", [])]
        candidate_skills = [s.lower() for s in self.candidate_data.get("skills", [])]
        
        gaps = []
        for skill in required:
            if skill not in candidate_skills:
                gaps.append(f"No mention of {skill} in resume")
        
        # Check for missing experience details
        if not self.candidate_data.get("years_experience"):
            gaps.append("Missing years of experience breakdown")
        
        return gaps
    
    def generate_question(self) -> str:
        """Generate the next screening question based on gaps"""
        if not self.gaps:
            return None
        
        gap = self.gaps[0]
        
        question_prompt = f"""
        You are conducting a screening interview for the {self.job_data.get('title')} position.
        
        Candidate profile summary:
        - Skills: {', '.join(self.candidate_data.get('skills', []))}
        - Recent roles: {json.dumps(self.candidate_data.get('recent_roles', [])[:2])}
        
        Gap identified: {gap}
        
        Generate ONE natural, conversational question to clarify this gap.
        The question should be friendly but professional, and allow the candidate to elaborate.
        Keep it to 1-2 sentences.
        """
        
        response = self.llm.invoke(question_prompt)
        return response.content
    
    def process_answer(self, question: str, answer: str):
        """Process candidate's answer and update their profile"""
        self.conversation_history.append({"question": question, "answer": answer})
        
        # Extract new information from the answer
        extraction_prompt = f"""
        Extract any new skills, experience details, or qualifications mentioned in this answer.
        Return a JSON object with:
        - new_skills: array of any skills mentioned
        - experience_update: object with skill areas and years if mentioned
        - clarification: string summarizing the key insight
        
        Answer: {answer}
        """
        
        response = self.llm.invoke(extraction_prompt)
        try:
            update = json.loads(response.content)
            if update.get("new_skills"):
                self.candidate_data["skills"].extend(update["new_skills"])
            if update.get("experience_update"):
                self.candidate_data.setdefault("years_experience", {}).update(update["experience_update"])
        except:
            pass
        
        # Remove the addressed gap
        if self.gaps:
            self.gaps.pop(0)

# Streamlit UI for interactive screening
def run_screening_ui():
    st.title("🤖 AI HR Screening Agent")
    
    # Initialize session state
    if "screening" not in st.session_state:
        st.session_state.screening = None
    if "messages" not in st.session_state:
        st.session_state.messages = []
    
    # Upload resume
    uploaded_file = st.file_uploader("Upload Candidate Resume", type=["pdf", "docx"])
    
    if uploaded_file and st.button("Start Screening"):
        # Save file temporarily
        with open(f"temp_{uploaded_file.name}", "wb") as f:
            f.write(uploaded_file.getbuffer())
        
        # Parse resume
        candidate = parse_resume.invoke({"file_path": f"temp_{uploaded_file.name}"})
        
        # Get job data
        job_query = st.selectbox("Select Position", ["Senior Python Developer", "Product Manager"])
        job_results = search_jobs.invoke(job_query)
        job_data = json.loads(job_results)[0]
        
        # Initialize conversation
        st.session_state.screening = ScreeningConversation(candidate, job_data)
        st.session_state.messages = []
        st.rerun()
    
    # Display conversation
    if st.session_state.screening:
        screening = st.session_state.screening
        
        for msg in st.session_state.messages:
            with st.chat_message(msg["role"]):
                st.write(msg["content"])
        
        # Generate next question
        if screening.gaps:
            question = screening.generate_question()
            if question:
                with st.chat_message("assistant"):
                    st.write(question)
                
                answer = st.chat_input("Your response...")
                if answer:
                    st.session_state.messages.append({"role": "assistant", "content": question})
                    st.session_state.messages.append({"role": "user", "content": answer})
                    screening.process_answer(question, answer)
                    st.rerun()
        else:
            st.success("Screening complete! All gaps addressed.")
            st.json(screening.candidate_data)

Step 6: Batch Processing Pipeline

For high-volume recruiting, you need a batch processor that evaluates multiple candidates against multiple jobs simultaneously:

import asyncio
from typing import List, Dict
from datetime import datetime
import pandas as pd

class BatchScreeningPipeline:
    """
    Processes multiple resumes against multiple job descriptions
    and produces ranked candidate lists.
    """
    
    def __init__(self, resume_directory: str, agent_executor: AgentExecutor):
        self.resume_directory = resume_directory
        self.agent = agent_executor
        self.results = []
    
    def collect_resumes(self) -> List[str]:
        """Gather all resume files from directory"""
        import glob
        return glob.glob(f"{self.resume_directory}/**/*", recursive=True)
    
    async def process_single(self, resume_path: str, job_title: str) -> Dict:
        """Process one resume against one job"""
        try:
            result = await self.agent.ainvoke({
                "input": f"Evaluate the candidate from resume file '{resume_path}' for the '{job_title}' position. Provide a complete screening report with scores."
            })
            return {
                "resume": resume_path,
                "job": job_title,
                "result": result["output"],
                "timestamp": datetime.now().isoformat()
            }
        except Exception as e:
            return {
                "resume": resume_path,
                "job": job_title,
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            }
    
    async def run_pipeline(self, job_titles: List[str]):
        """Run full batch screening pipeline"""
        resumes = self.collect_resumes()
        
        tasks = []
        for resume in resumes:
            for job in job_titles:
                tasks.append(self.process_single(resume, job))
        
        # Process in batches of 5 to avoid rate limits
        batch_size = 5
        for i in range(0, len(tasks), batch_size):
            batch = tasks[i:i+batch_size]
            batch_results = await asyncio.gather(*batch)
            self.results.extend(batch_results)
            print(f"Processed batch {i//batch_size + 1}/{(len(tasks) + batch_size - 1)//batch_size}")
        
        return self.generate_report()
    
    def generate_report(self) -> pd.DataFrame:
        """Generate a ranked report DataFrame"""
        rows = []
        for r in self.results:
            if "error" not in r:
                # Extract score from result text (simplified)
                result_text = r["result"]
                score = 0
                if "composite_score" in result_text.lower():
                    # Try to extract numeric score
                    import re
                    matches = re.findall(r'composite_score[:\s]+(\d+\.?\d*)', result_text, re.IGNORECASE)
                    if matches:
                        score = float(matches[0])
                
                rows.append({
                    "Candidate": r["resume"].split("/")[-1],
                    "Job": r["job"],
                    "Score": score,
                    "Status": "STRONG_HIRE" if score > 80 else "POTENTIAL" if score > 60 else "SCREEN_OUT"
                })
        
        df = pd.DataFrame(rows)
        df = df.sort_values("Score", ascending=False)
        return df

# Usage
pipeline = BatchScreeningPipeline("./resumes", agent_executor)
report = await pipeline.run_pipeline(["Senior Python Developer", "Product Manager"])
print(report.head(20))

Best Practices for HR Screening Agents

1. Bias Detection and Mitigation

Implement regular bias audits on your screening agent. Create a diverse test set of resumes with identical qualifications but different demographic indicators, and verify the agent produces consistent scores. Consider implementing a bias detection layer that flags potential disparate impact:

@tool("bias-check")
def bias_audit(candidate_data: Dict, evaluation_result: Dict) -> str:
    """
    Check if any protected characteristics influenced the evaluation.
    This tool should be run after every screening evaluation.
    """
    protected_terms = [
        "gender", "age", "race", "religion", "marital", "pregnancy",
        "disability", "veteran", "national origin"
    ]
    
    # Check if resume contained protected information
    resume_text = json.dumps(candidate_data).lower()
    flagged_terms = [term for term in protected_terms if term in resume_text]
    
    if flagged_terms:
        return json.dumps({
            "warning": f"Resume contained references to: {', '.join(flagged_terms)}",
            "action": "Review evaluation manually to ensure no bias",
            "automatic_hold": True
        })
    
    return json.dumps({"status": "No protected characteristics detected", "automatic_hold": False})

2. Human-in-the-Loop Design

Never let the agent make final hiring decisions autonomously. The agent should provide recommendations and evidence, but the final decision must rest with a human recruiter. Implement confidence thresholds that trigger mandatory human review:

# Confidence thresholds for human review triggers
HUMAN_REVIEW_TRIGGERS = {
    "score_range_40_70": "Borderline candidates require human judgment",
    "missing_critical_skills": "When >2 required skills are missing but candidate scores well on others",
    "unusual_career_gap": "Employment gaps >2 years require context",
    "overqualified": "Candidate exceeds requirements by >2x, risk of attrition"
}

def determine_review_requirement(evaluation_result: Dict) -> Dict:
    """Determine if human review is mandatory"""
    triggers = []
    
    score = evaluation_result.get("composite_score", 0)
    if 40 <= score <= 70:
        triggers.append(HUMAN_REVIEW_TRIGGERS["score_range_40_70"])
    
    missing = evaluation_result.get("missing_skills", [])
    if len(missing) > 2 and score > 50:
        triggers.append(HUMAN_REVIEW_TRIGGERS["missing_critical_skills"])
    
    return {
        "requires_human_review": len(triggers) > 0,
        "triggers": triggers,
        "auto_proceed": len(triggers) == 0 and score > 70
    }

3. Transparent Scoring Logic

Every recommendation should be explainable. Store the full chain of reasoning for each screening decision so recruiters can understand exactly why a candidate received a particular score. This builds trust and allows for debugging edge cases.

4. Regular Tool Calibration

Review your screening agent's performance monthly. Compare its recommendations against actual hiring outcomes. If candidates flagged as "STRONG_HIRE" are consistently rejected by human interviewers, your screening criteria need adjustment. Track metrics like:

5. Data Privacy and Compliance

HR data is highly sensitive. Implement strict data handling practices:

# Data retention and privacy configuration
from datetime import timedelta

class DataRetentionPolicy:
    """
    Ensures compliance with data privacy regulations (GDPR, CCPA, etc.)
    """
    
    def __init__(self):
        self.retention_days = 90  # Auto-delete after 90 days
        self.anonymize_after_days = 30  # Strip PII after 30 days
    
    def should_anonymize(self, record_date: datetime) -> bool:
        return (datetime.now() - record_date).days > self.anonymize_after_days
    
    def should_delete(self, record_date: datetime) -> bool:
        return (datetime.now() - record_date).days > self.retention_days
    
    def anonymize_candidate(self, candidate_data: Dict) -> Dict:
        """Remove personally identifiable information"""
        fields_to_remove = ["full_name", "email", "phone", "address"]
        for field in fields_to_remove:
            candidate_data.pop(field, None)
        candidate_data["anonymized"] = True
        candidate_data["candidate_id"] = f"ANON_{hash(candidate_data.get('full_name', ''))}"
        return candidate_data

6. Structured Output for Recruiter Integration

Format your agent's output to integrate seamlessly with existing ATS (Applicant Tracking Systems) and HR tools. Use consistent JSON schemas that downstream systems can consume:

# Standardized screening output schema
SCREENING_OUTPUT_SCHEMA = {
    "type": "object",
    "properties": {
        "candidate_id": {"type": "string"},
        "job_id": {"type": "string"},
        "evaluation_date": {"type": "string", "format": "date-time"},
        "scores": {
            "type": "object",
            "properties": {
                "overall_fit": {"type": "number", "minimum": 0, "maximum": 100},
                "skills_match": {"type": "number"},
                "experience_match": {"type": "number"},
                "education_match": {"type": "number"},
                "culture_fit_indicators": {"type": "number"}
            }
        },
        "skills_analysis": {
            "type": "object",
            "properties": {
                "matched": {"type": "array", "items": {"type": "string"}},
                "missing": {"type": "array", "items": {"type": "string"}},
                "transferable": {"type": "array", "items": {"type": "string"}}
            }
        },
        "recommendation": {
            "type": "object",
            "properties": {
                "level": {"enum": ["STRONG_HIRE", "HIRE", "POTENTIAL", "SCREEN_OUT"]},
                "confidence": {"type": "number"},
                "reasoning": {"type": "string"},
                "suggested_next_step": {"type": "string"}
            }
        },
        "flags": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "type": {"enum": ["INFO", "WARNING", "CRITICAL"]},
                    "message": {"type": "string"}
                }
            }
        }
    }
}

Common Pitfalls and How to Avoid Them

When building your HR screening agent, watch out for these common issues:

Production Deployment Considerations

When moving from prototype to production, consider these additional layers:

# Production-grade agent wrapper with monitoring
from langchain.callbacks import LangChainTracer
import logging

class ProductionScreeningAgent:
    """
    Production wrapper with monitoring, error handling, and rate limiting.
    """
    
    def __init__(self, agent_executor: AgentExecutor):
        self.executor = agent_executor
        self.tracer = LangChainTracer(project_name="hr-screening-prod")
        self.logger = logging.getLogger("hr-screening-agent")
        
        # Rate limiting configuration
        self.max_concurrent = 3
        self.semaphore = asyncio.Semaphore(self.max_concurrent)
    
    async def screen_candidate(self, resume_path: str, job_id: str) -> Dict:
        """Production screening with full error handling and tracing"""
        async with self.semaphore:
            try:
                self.logger.info(f"Starting screening: {resume_path} for {job_id}")
                
                result = await self.executor.ainvoke({
                    "input": f"Evaluate candidate from '{resume_path}' for '{job_id}'",
                }, config={"callbacks": [self.tracer]})
                
                # Validate output structure
                validated = self.validate_output(result["output"])
                
                self.logger.info(f"Screening complete: score={validated.get('scores', {}).get('overall_fit')}")
                return validated
                
            except Exception as e:
                self.logger.error(f"Screening failed: {str(e)}")
                return {
                    "error": str(e),
                    "status": "FAILED",
                    "requires_manual_review": True
                }
    
    def validate_output(self, output: str) -> Dict:
        """Ensure output meets production schema requirements"""
        # Attempt to parse structured data from agent output
        # Fall back to structured extraction if needed
        try:
            # Try direct JSON parse if agent returned structured data
            if "{" in output:
                start = output.index("{")
                end = output.rindex("}") + 1
                return json.loads(output[start:end])
        except:

— Ad —

Google AdSense will appear here after approval

← Back to all articles