# MULTI-AGENT SYSTEM SPECIFICATION ## System Overview **Name:** Intelligent Recruitment Automation System (IRAS) **Purpose:** End-to-end hiring pipeline automation from candidate sourcing to interview scheduling and reporting **Complexity:** 6 specialized agents, 12 integrated tools, 4 external platforms (LinkedIn, Google Drive, Google Calendar/Meet, ATS) --- ## Architecture ### Agent Roster | Agent | Role | Tools | Inputs | Outputs | |-------|------|-------|--------|---------| | **Sourcing Agent** | Find and extract candidate profiles from LinkedIn | LinkedIn API, web scraper, profile parser | Job requirements, keywords, filters | Structured candidate profiles (JSON) | | **Screening Agent** | Evaluate resumes against job criteria | Resume parser, NLP matcher, scoring algorithm | Candidate profiles, job description | Ranked candidate list with scores | | **ATS Sync Agent** | Maintain candidate records in ATS | ATS API, data validator, deduplication | Screened candidates, status updates | ATS candidate IDs, sync confirmations | | **Communication Agent** | Send personalized outreach and scheduling requests | Email API, template engine, tracking | Candidate contact info, interview slots | Sent confirmations, response tracking | | **Scheduling Agent** | Coordinate interview times and create Meet links | Google Calendar API, Meet API, conflict resolver | Candidate availability, interviewer calendars | Scheduled events with Meet links | | **Reporting Agent** | Generate hiring pipeline analytics | Data aggregator, chart generator, report builder | ATS data, agent logs, metrics | Weekly/monthly hiring reports (PDF) | ### Workflow Diagram ``` [Job Posting] → [Sourcing Agent] → [Screening Agent] → [ATS Sync Agent] ↓ ↓ ↓ (LinkedIn) (Resume scoring) (ATS Platform) ↓ [Human Review Gate] ↓ [Communication Agent] ← → [Scheduling Agent] ↓ ↓ (Email outreach) (Google Meet/Calendar) ↓ [Reporting Agent] ↓ (Analytics Dashboard) ``` **Execution Flow:** 1. **Parallel Phase (Days 1-2):** Sourcing Agent continuously finds candidates → Screening Agent evaluates in batches 2. **Sequential Phase (Day 2):** ATS Sync Agent updates records → Human approves top candidates 3. **Parallel Phase (Days 3-4):** Communication Agent sends outreach → Scheduling Agent finds slots concurrently 4. **Final Phase (Day 5):** Reporting Agent generates metrics and insights ### Communication Protocol ```json { "message_type": "candidate_handoff | scheduling_request | status_update | error_report", "timestamp": "ISO 8601", "from_agent": "agent_name", "to_agent": "agent_name", "payload": { "candidate_id": "unique_id", "data": {}, "metadata": { "priority": "high | medium | low", "retry_count": 0, "expires_at": "ISO 8601" } }, "status": "success | pending | failed" } ``` --- ## Agent Implementations ### 1. Sourcing Agent **Role Prompt:** ``` You are a Sourcing Agent specialized in finding qualified candidates on LinkedIn. RESPONSIBILITIES: - Search LinkedIn using job-relevant keywords and filters - Extract candidate profiles (name, experience, skills, education, contact) - Validate profile completeness (minimum 70% fields populated) - Output structured JSON for downstream processing REASONING APPROACH (ReAct): Thought: Analyze job requirements to determine search parameters Action: Execute LinkedIn search with optimized filters Observation: Review search results quality and relevance Thought: Assess if results meet quality threshold (>5 qualified candidates) Action: Extract profile data or refine search if needed Observation: Validate extracted data completeness CONSTRAINTS: - Maximum 50 profiles per job posting (cost control) - Respect LinkedIn rate limits (100 searches/day) - Flag profiles missing contact information for manual review SUCCESS CRITERIA: - 80%+ of sourced candidates meet minimum qualifications - <10% duplicate profiles - All extracted data passes validation schema ``` **ReAct Loop Implementation:** ```python class SourcingAgent: def __init__(self, linkedin_api, profile_parser): self.linkedin = linkedin_api self.parser = profile_parser self.max_profiles = 50 def execute(self, job_requirements): """Main ReAct loop for candidate sourcing""" candidates = [] attempt = 0 while len(candidates) < self.max_profiles and attempt < 3: # THOUGHT search_params = self._analyze_requirements(job_requirements, attempt) self.log(f"Thought: Searching with params: {search_params}") # ACTION try: results = self.linkedin.search( keywords=search_params['keywords'], location=search_params['location'], experience_level=search_params['experience'], limit=20 ) self.log(f"Action: Executed search, found {len(results)} profiles") except RateLimitError: self.log("Observation: Rate limit hit, waiting 60s") time.sleep(60) continue # OBSERVATION qualified = self._evaluate_results(results, job_requirements) self.log(f"Observation: {len(qualified)} qualified candidates found") # THOUGHT if len(qualified) < 5: self.log("Thought: Results too sparse, refining search parameters") attempt += 1 continue # ACTION for profile in qualified: parsed = self.parser.extract(profile) if self._validate_profile(parsed): candidates.append(parsed) self.log(f"Action: Added candidate {parsed['name']}") break return { "candidates": candidates, "metadata": { "total_found": len(candidates), "search_attempts": attempt + 1, "timestamp": datetime.utcnow().isoformat() } } def _validate_profile(self, profile): """Validate profile has minimum required fields""" required = ['name', 'email', 'experience', 'skills'] completeness = sum(1 for field in required if profile.get(field)) / len(required) return completeness >= 0.7 ``` **Example Execution:** ``` Input: {"title": "Senior Python Developer", "skills": ["Python", "Django", "AWS"], "experience": "5+ years"} Output: { "candidates": [ { "name": "Jane Smith", "email": "jane.smith@example.com", "experience": "7 years", "skills": ["Python", "Django", "PostgreSQL", "AWS"], "education": "BS Computer Science", "linkedin_url": "linkedin.com/in/janesmith" }, ... ], "metadata": {"total_found": 42, "search_attempts": 1} } ``` --- ### 2. Screening Agent **Role Prompt:** ``` You are a Screening Agent that evaluates candidate qualifications against job requirements. RESPONSIBILITIES: - Parse resumes and extract structured data - Score candidates on multiple dimensions (skills match, experience, education) - Rank candidates by overall fit score - Flag edge cases for human review REASONING APPROACH (ReAct + CoT): Thought: Break down job requirements into scoreable criteria Action: Parse candidate resume and extract relevant sections Observation: Identify which requirements are met/unmet Thought: Calculate weighted scores across dimensions Action: Generate ranking with justification Observation: Validate scores are within expected ranges SCORING RUBRIC: - Skills Match (40%): Keyword overlap + semantic similarity - Experience (30%): Years in role + relevant projects - Education (15%): Degree level + field relevance - Culture Fit (15%): Values alignment from cover letter/profile EDGE CASES: - Overqualified candidates (>150% experience): Flag for retention risk - Career gaps >2 years: Flag for discussion - Non-traditional backgrounds: Flag for diversity consideration SUCCESS CRITERIA: - Top 20% of ranked candidates advance to human review - <5% false negatives (qualified candidates scored too low) - All scores include human-readable justification ``` **ReAct Loop Implementation:** ```python class ScreeningAgent: def __init__(self, resume_parser, nlp_model): self.parser = resume_parser self.nlp = nlp_model def execute(self, candidates, job_description): """Main ReAct loop for candidate screening""" scored_candidates = [] for candidate in candidates: # THOUGHT self.log(f"Thought: Evaluating {candidate['name']} against requirements") criteria = self._extract_criteria(job_description) # ACTION resume_data = self.parser.parse(candidate) self.log(f"Action: Parsed resume into structured format") # OBSERVATION matches = self._calculate_matches(resume_data, criteria) self.log(f"Observation: Found {matches['skills_matched']}/{matches['skills_total']} skill matches") # THOUGHT self.log("Thought: Calculating weighted scores") # ACTION scores = { 'skills': self._score_skills(matches, criteria) * 0.4, 'experience': self._score_experience(resume_data, criteria) * 0.3, 'education': self._score_education(resume_data, criteria) * 0.15, 'culture': self._score_culture(resume_data, job_description) * 0.15 } total_score = sum(scores.values()) # OBSERVATION flags = self._check_edge_cases(resume_data, criteria) self.log(f"Observation: Total score {total_score:.2f}, Flags: {flags}") scored_candidates.append({ "candidate_id": candidate['id'], "name": candidate['name'], "total_score": total_score, "dimension_scores": scores, "flags": flags, "justification": self._generate_justification(scores, matches) }) # Rank and filter ranked = sorted(scored_candidates, key=lambda x: x['total_score'], reverse=True) cutoff = int(len(ranked) * 0.2) or 1 # Top 20% return { "qualified_candidates": ranked[:cutoff], "all_scores": ranked, "metadata": { "total_screened": len(candidates), "qualified_count": len(ranked[:cutoff]), "avg_score": sum(c['total_score'] for c in ranked) / len(ranked) } } def _score_skills(self, matches, criteria): """Calculate skills match score (0-100)""" required_match = matches['required_skills_matched'] / max(matches['required_skills_total'], 1) preferred_match = matches['preferred_skills_matched'] / max(matches['preferred_skills_total'], 1) return (required_match * 0.7 + preferred_match * 0.3) * 100 def _check_edge_cases(self, resume_data, criteria): """Identify candidates needing special review""" flags = [] if resume_data['years_experience'] > criteria['experience_years'] * 1.5: flags.append("overqualified") if self._has_career_gap(resume_data['work_history']): flags.append("career_gap") if resume_data['education']['degree'] not in ['BS', 'MS', 'PhD']: flags.append("non_traditional_background") return flags ``` **Example Execution:** ``` Input: 42 candidates + job description Output: { "qualified_candidates": [ { "candidate_id": "cand_001", "name": "Jane Smith", "total_score": 87.5, "dimension_scores": {"skills": 35, "experience": 27, "education": 13, "culture": 12.5}, "flags": [], "justification": "Strong skills match (9/10 required), 7 years relevant experience with Django/AWS projects, BS in CS from accredited university" }, ... ], "metadata": {"total_screened": 42, "qualified_count": 8, "avg_score": 62.3} } ``` --- ### 3. ATS Sync Agent **Role Prompt:** ``` You are an ATS Sync Agent responsible for maintaining candidate records in the company's Applicant Tracking System. RESPONSIBILITIES: - Create/update candidate profiles in ATS - Prevent duplicate entries (fuzzy matching on name+email) - Track candidate status transitions (Applied → Screening → Interview → Offer) - Ensure data consistency across systems REASONING APPROACH (ReAct): Thought: Check if candidate already exists in ATS Action: Query ATS by email and name (fuzzy match) Observation: Determine if new record or update needed Thought: Validate data meets ATS schema requirements Action: Create or update record with all fields Observation: Confirm successful sync and capture ATS ID ERROR HANDLING: - Duplicate detection: Merge records if >90% similarity - API failures: Retry 3x with exponential backoff (2s, 4s, 8s) - Validation errors: Log details and flag for manual review - Orphaned records: Weekly reconciliation job SUCCESS CRITERIA: - 100% of qualified candidates synced to ATS - <1% duplicate records created - All status transitions logged with timestamps ``` **ReAct Loop Implementation:** ```python class ATSSyncAgent: def __init__(self, ats_api, deduplicator): self.ats = ats_api self.dedup = deduplicator self.retry_policy = [2, 4, 8] # Exponential backoff def execute(self, candidates): """Main ReAct loop for ATS synchronization""" sync_results = [] for candidate in candidates: # THOUGHT self.log(f"Thought: Syncing {candidate['name']} to ATS") # ACTION existing = self._check_existing(candidate) self.log(f"Action: Checked for existing record") # OBSERVATION if existing: self.log(f"Observation: Found existing record {existing['id']}") operation = "update" target_id = existing['id'] else: self.log("Observation: No existing record, will create new") operation = "create" target_id = None # THOUGHT self.log(f"Thought: Validating data for {operation}") # ACTION validated_data = self._validate_and_transform(candidate, operation) # Execute with retry logic for attempt, delay in enumerate(self.retry_policy + [None]): try: if operation == "create": result = self.ats.create_candidate(validated_data) else: result = self.ats.update_candidate(target_id, validated_data) # OBSERVATION self.log(f"Observation: {operation} succeeded, ATS ID: {result['id']}") sync_results.append({ "candidate_id": candidate['id'], "ats_id": result['id'], "operation": operation, "status": "success", "timestamp": datetime.utcnow().isoformat() }) break except ATSValidationError as e: self.log(f"Error: Validation failed - {str(e)}") sync_results.append({ "candidate_id": candidate['id'], "operation": operation, "status": "failed", "error": str(e), "requires_manual_review": True }) break except ATSAPIError as e: if delay is None: # Last attempt self.log(f"Error: API failure after all retries - {str(e)}") sync_results.append({ "candidate_id": candidate['id'], "operation": operation, "status": "failed", "error": str(e), "retry_count": attempt + 1 }) else: self.log(f"Observation: API error, retrying in {delay}s") time.sleep(delay) return { "sync_results": sync_results, "metadata": { "total_processed": len(candidates), "successful": sum(1 for r in sync_results if r['status'] == 'success'), "failed": sum(1 for r in sync_results if r['status'] == 'failed') } } def _check_existing(self, candidate): """Check for existing ATS records using fuzzy matching""" # Exact email match by_email = self.ats.search(email=candidate['email']) if by_email: return by_email[0] # Fuzzy name match by_name = self.ats.search(name=candidate['name']) for record in by_name: similarity = self.dedup.calculate_similarity(candidate, record) if similarity > 0.9: return record return None ``` --- ### 4. Communication Agent **Role Prompt:** ``` You are a Communication Agent responsible for candidate outreach and interview coordination. RESPONSIBILITIES: - Generate personalized email templates based on candidate profile - Send scheduling requests with available time slots - Track email opens, clicks, and responses - Handle common candidate questions via pre-approved templates - Escalate complex questions to human recruiters REASONING APPROACH (ReAct): Thought: Determine communication type (initial outreach, reminder, rejection) Action: Generate personalized email from template + candidate data Observation: Review email for tone, accuracy, missing fields Thought: Validate candidate contact information is current Action: Send email via API and log tracking metadata Observation: Monitor delivery status and engagement metrics PERSONALIZATION STRATEGY: - Reference specific skills/projects from candidate's profile - Mention mutual connections if available - Adjust tone based on seniority level (formal for executives, casual for junior) - Include role-specific details (team, technologies, benefits) SUCCESS CRITERIA: - 90%+ emails delivered successfully - 40%+ response rate within 48 hours - <2% marked as spam - All communications logged in ATS ``` **ReAct Loop Implementation:** ```python class CommunicationAgent: def __init__(self, email_api, template_engine, tracker): self.email = email_api self.templates = template_engine self.tracker = tracker def execute(self, candidates, interview_slots, communication_type="initial_outreach"): """Main ReAct loop for candidate communication""" sent_emails = [] for candidate in candidates: # THOUGHT self.log(f"Thought: Crafting {communication_type} email for {candidate['name']}") # ACTION email_content = self._generate_personalized_email( candidate, interview_slots, communication_type ) self.log("Action: Generated personalized email") # OBSERVATION validation = self._validate_email(email_content, candidate) if not validation['valid']: self.log(f"Observation: Email validation failed - {validation['errors']}") sent_emails.append({ "candidate_id": candidate['id'], "status": "failed", "error": "validation_failed", "details": validation['errors'] }) continue # THOUGHT self.log("Thought: Email looks good, validating contact info") # ACTION if not self._verify_email_address(candidate['email']): self.log("Observation: Email address appears invalid") sent_emails.append({ "candidate_id": candidate['id'], "status": "failed", "error": "invalid_email" }) continue # ACTION try: result = self.email.send( to=candidate['email'], subject=email_content['subject'], body=email_content['body'], tracking_enabled=True ) # OBSERVATION self.log(f"Observation: Email sent successfully, message_id: {result['message_id']}") sent_emails.append({ "candidate_id": candidate['id'], "email": candidate['email'], "message_id": result['message_id'], "status": "sent", "sent_at": datetime.utcnow().isoformat(), "tracking_url": result['tracking_url'] }) except EmailAPIError as e: self.log(f"Error: Failed to send email - {str(e)}") sent_emails.append({ "candidate_id": candidate['id'], "status": "failed", "error": str(e) }) return { "sent_emails": sent_emails, "metadata": { "total_attempted": len(candidates), "successful": sum(1 for e in sent_emails if e['status'] == 'sent'), "failed": sum(1 for e in sent_emails if e['status'] == 'failed') } } def _generate_personalized_email(self, candidate, slots, comm_type): """Create personalized email using templates and candidate data""" template = self.templates.get(comm_type) # Extract personalization tokens tokens = { "candidate_name": candidate['name'], "candidate_skills": ", ".join(candidate['skills'][:3]), "specific_project": candidate.get('notable_project', 'your experience'), "role_title": candidate['applied_role'], "company_name": "Acme Corp", "available_slots": self._format_slots(slots[:3]), "recruiter_name": "Sarah Johnson", "recruiter_title": "Senior Technical Recruiter" } # Adjust tone based on seniority if "senior" in candidate['applied_role'].lower() or candidate['years_experience'] > 8: template = self.templates.get(f"{comm_type}_formal") return { "subject": template['subject'].format(**tokens), "body": template['body'].format(**tokens) } ``` --- ### 5. Scheduling Agent **Role Prompt:** ``` You are a Scheduling Agent that coordinates interview times between candidates and interviewers. RESPONSIBILITIES: - Find optimal interview slots considering all participant calendars - Create Google Meet links for virtual interviews - Send calendar invitations to all participants - Handle rescheduling requests with minimal disruption - Respect time zones and working hours REASONING APPROACH (ReAct + Constraint Solving): Thought: Identify hard constraints (interviewer availability, candidate preferences, time zones) Action: Query Google Calendar for all participants Observation: Identify free time blocks that satisfy constraints Thought: Rank time slots by optimization criteria (fewest conflicts, preferred times) Action: Select best slot and create Meet event Observation: Validate all participants received invitations OPTIMIZATION CRITERIA: 1. Business hours in all participants' time zones 2. Minimum 30-minute buffer between back-to-back interviews 3. Prefer interviewer's stated "best times" (stored in preferences) 4. Avoid Mondays before 10am and Fridays after 3pm 5. Group panel interviews to minimize calendar fragmentation SUCCESS CRITERIA: - 95%+ of interviews scheduled on first attempt - <10% rescheduling rate - All Meet links functional at scheduled time - Zero double-bookings ``` **ReAct Loop Implementation:** ```python class SchedulingAgent: def __init__(self, gcal_api, meet_api, conflict_resolver): self.calendar = gcal_api self.meet = meet_api self.resolver = conflict_resolver def execute(self, candidates, interviewers, interview_duration=60): """Main ReAct loop for interview scheduling""" scheduled_interviews = [] for candidate in candidates: # THOUGHT self.log(f"Thought: Scheduling interview for {candidate['name']}") assigned_interviewers = self._assign_interviewers(candidate, interviewers) # ACTION self.log(f"Action: Querying calendars for {len(assigned_interviewers)} interviewers") availability = self._get_availability( assigned_interviewers, start_date=datetime.now(), end_date=datetime.now() + timedelta(days=5) ) # OBSERVATION free_slots = self._find_free_slots(availability, interview_duration) self.log(f"Observation: Found {len(free_slots)} potential time slots") if not free_slots: self.log("Observation: No available slots found, escalating to manual scheduling") scheduled_interviews.append({ "candidate_id": candidate['id'], "status": "failed", "reason": "no_available_slots", "requires_manual_scheduling": True }) continue # THOUGHT self.log("Thought: Ranking slots by optimization criteria") # ACTION ranked_slots = self._rank_slots( free_slots, assigned_interviewers, candidate['timezone'] ) best_slot = ranked_slots[0] # ACTION try: # Create Google Meet event meet_link = self.meet.create_meeting( title=f"Interview: {candidate['name']} - {candidate['applied_role']}", start_time=best_slot['start'], duration_minutes=interview_duration ) # Create calendar events for all participants event = self.calendar.create_event( summary=f"Interview: {candidate['name']}", start=best_slot['start'], end=best_slot['end'], attendees=[candidate['email']] + [i['email'] for i in assigned_interviewers], description=f"Google Meet: {meet_link['url']}\n\nCandidate: {candidate['name']}\nRole: {candidate['applied_role']}", conference_data=meet_link ) # OBSERVATION self.log(f"Observation: Interview scheduled successfully, event ID: {event['id']}") scheduled_interviews.append({ "candidate_id": candidate['id'], "candidate_name": candidate['name'], "interviewers": [i['name'] for i in assigned_interviewers], "scheduled_time": best_slot['start'].isoformat(), "duration_minutes": interview_duration, "meet_link": meet_link['url'], "calendar_event_id": event['id'], "status": "scheduled", "timezone": candidate['timezone'] }) except (CalendarAPIError, MeetAPIError) as e: self.log(f"Error: Scheduling failed - {str(e)}") scheduled_interviews.append({ "candidate_id": candidate['id'], "status": "failed", "error": str(e) }) return { "scheduled_interviews": scheduled_interviews, "metadata": { "total_candidates": len(candidates), "successfully_scheduled": sum(1 for s in scheduled_interviews if s['status'] == 'scheduled'), "failed": sum(1 for s in scheduled_interviews if s['status'] == 'failed') } } def _find_free_slots(self, availability, duration): """Find time blocks where all participants are free""" # Merge all calendars to find common free time all_busy_times = [] for person in availability: all_busy_times.extend(person['busy_times']) # Sort and merge overlapping busy periods merged_busy = self.resolver.merge_intervals(all_busy_times) # Find gaps >= duration free_slots = [] current_time = datetime.now().replace(hour=9, minute=0) # Start at 9am end_time = current_time + timedelta(days=5) while current_time < end_time: # Check if current_time overlaps with any busy period is_free = not any( busy['start'] <= current_time < busy['end'] for busy in merged_busy ) if is_free and self._is_business_hours(current_time): # Check if we have duration minutes free slot_end = current_time + timedelta(minutes=duration) slot_free = not any( busy['start'] < slot_end and busy['end'] > current_time for busy in merged_busy ) if slot_free: free_slots.append({ 'start': current_time, 'end': slot_end }) current_time += timedelta(minutes=30) # Check every 30 min return free_slots def _rank_slots(self, slots, interviewers, candidate_tz): """Rank slots by optimization criteria""" scored_slots = [] for slot in slots: score = 100 # Start with perfect score # Penalty for Monday morning or Friday afternoon if slot['start'].weekday() == 0 and slot['start'].hour < 10: score -= 15 if slot['start'].weekday() == 4 and slot['start'].hour >= 15: score -= 15 # Bonus for interviewer preferred times for interviewer in interviewers: if 'preferred_hours' in interviewer: if slot['start'].hour in interviewer['preferred_hours']: score += 10 # Penalty for early/late in candidate's timezone candidate_local_hour = self._convert_to_timezone(slot['start'], candidate_tz).hour if candidate_local_hour < 9 or candidate_local_hour > 17: score -= 20 scored_slots.append({'slot': slot, 'score': score}) return sorted(scored_slots, key=lambda x: x['score'], reverse=True) ``` --- ### 6. Reporting Agent **Role Prompt:** ``` You are a Reporting Agent that generates hiring pipeline analytics and insights. RESPONSIBILITIES: - Aggregate data from all agents and ATS - Calculate key metrics (time-to-hire, conversion rates, cost-per-hire) - Generate visualizations (funnel charts, trend lines, heat maps) - Produce weekly summary reports and monthly deep-dives - Identify bottlenecks and recommend process improvements REASONING APPROACH (CoT + Data Analysis): Thought: Determine reporting period and stakeholder needs Action: Query ATS and agent logs for raw data Observation: Assess data completeness and quality Thought: Calculate metrics and identify trends Action: Generate visualizations and narrative insights Observation: Validate calculations and check for anomalies KEY METRICS: - Pipeline velocity: Candidates at each stage, conversion rates - Time-to-hire: Days from application to offer acceptance - Cost-per-hire: (Sourcing + Screening + Interview costs) / Hires - Quality-of-hire: 90-day retention rate, performance ratings - Diversity metrics: Demographic distribution at each stage SUCCESS CRITERIA: - Reports delivered within 2 hours of request - All metrics accurate to within 2% of manual calculation - Visualizations render correctly in PDF export