Candidate Management Pipeline

id: candidate_management_pipeline
name: Candidate Management Pipeline
tasks:
- id: source_candidates
  name: Source Candidates
  type: script
  script: "#!/usr/bin/env python3\nimport json\nimport os\nimport random\nfrom datetime\
    \ import datetime\n\njob_req = json.loads(os.environ.get('job_requirements', '{}'))\n\
    sources = json.loads(os.environ.get('candidate_sources', '[\"database\"]'))\n\n\
    print(f\"Sourcing candidates for: {job_req.get('title')}\")\nprint(f\"From sources:\
    \ {sources}\")\n\n# Simulate candidate sourcing (in production, would connect\
    \ to real sources)\ncandidates = []\n\n# Simulate database sourcing\nif \"database\"\
    \ in sources:\n    db_candidates = [\n        {\n            \"id\": f\"DB{i:03d}\"\
    ,\n            \"name\": f\"Candidate {i}\",\n            \"email\": f\"candidate{i}@example.com\"\
    ,\n            \"source\": \"database\",\n            \"resume_url\": f\"https://resumes.example.com/candidate{i}.pdf\"\
    ,\n            \"skills\": random.sample([\"Python\", \"Java\", \"APIs\", \"SQL\"\
    , \"AWS\", \"React\"], k=3),\n            \"experience_years\": random.randint(1,\
    \ 10),\n            \"location\": random.choice([\"New York\", \"San Francisco\"\
    , \"Remote\"])\n        }\n        for i in range(1, 6)\n    ]\n    candidates.extend(db_candidates)\n\
    \n# Simulate LinkedIn sourcing\nif \"linkedin\" in sources:\n    linkedin_candidates\
    \ = [\n        {\n            \"id\": f\"LI{i:03d}\",\n            \"name\": f\"\
    LinkedIn Candidate {i}\",\n            \"email\": f\"linkedin{i}@example.com\"\
    ,\n            \"source\": \"linkedin\",\n            \"profile_url\": f\"https://linkedin.com/in/candidate{i}\"\
    ,\n            \"skills\": random.sample([\"Python\", \"JavaScript\", \"Docker\"\
    , \"Kubernetes\", \"CI/CD\"], k=4),\n            \"experience_years\": random.randint(2,\
    \ 8),\n            \"location\": random.choice([\"Austin\", \"Seattle\", \"Boston\"\
    ])\n        }\n        for i in range(1, 4)\n    ]\n    candidates.extend(linkedin_candidates)\n\
    \n# Save candidates for next step\nwith open('/tmp/sourced_candidates.json', 'w')\
    \ as f:\n    json.dump(candidates, f, indent=2)\n\nprint(f\"\u2705 Sourced {len(candidates)}\
    \ candidates\")\nprint(f\"__OUTPUTS__ {json.dumps({'candidates_count': len(candidates),\
    \ 'sources_used': sources})}\")\n"
  description: Fetch candidates from multiple sources based on job requirements
  timeout_seconds: 30
- id: screen_candidates
  name: Screen Candidates
  type: script
  script: "#!/usr/bin/env python3\nimport json\nimport os\nimport re\n\njob_req =\
    \ json.loads(os.environ.get('job_requirements', '{}'))\nrequired_skills = job_req.get('skills',\
    \ [])\nmin_experience = job_req.get('experience', 0)\n\n# Load sourced candidates\n\
    with open('/tmp/sourced_candidates.json', 'r') as f:\n    candidates = json.load(f)\n\
    \nprint(f\"Screening {len(candidates)} candidates\")\nprint(f\"Required skills:\
    \ {required_skills}\")\nprint(f\"Minimum experience: {min_experience} years\"\
    )\n\nscreened_candidates = []\n\nfor candidate in candidates:\n    # Calculate\
    \ skill match\n    candidate_skills = [s.lower() for s in candidate.get('skills',\
    \ [])]\n    matched_skills = [s for s in required_skills if s.lower() in candidate_skills]\n\
    \    skill_match = len(matched_skills) / len(required_skills) * 100 if required_skills\
    \ else 0\n    \n    # Check experience\n    experience_match = candidate.get('experience_years',\
    \ 0) >= min_experience\n    \n    # Basic screening score\n    screening_score\
    \ = skill_match * 0.7 + (30 if experience_match else 0)\n    \n    candidate['screening_results']\
    \ = {\n        'skill_match_percentage': skill_match,\n        'matched_skills':\
    \ matched_skills,\n        'experience_qualified': experience_match,\n       \
    \ 'screening_score': screening_score,\n        'passed_screening': screening_score\
    \ >= 50\n    }\n    \n    if candidate['screening_results']['passed_screening']:\n\
    \        screened_candidates.append(candidate)\n        print(f\"\u2713 {candidate['name']}:\
    \ Score {screening_score:.1f}\")\n    else:\n        print(f\"\u2717 {candidate['name']}:\
    \ Score {screening_score:.1f} (Failed)\")\n\n# Save screened candidates\nwith\
    \ open('/tmp/screened_candidates.json', 'w') as f:\n    json.dump(screened_candidates,\
    \ f, indent=2)\n\nprint(f\"\\n\u2705 Screening complete: {len(screened_candidates)}/{len(candidates)}\
    \ passed\")\nprint(f\"__OUTPUTS__ {json.dumps({'total_screened': len(candidates),\
    \ 'passed_screening': len(screened_candidates)})}\")\n"
  depends_on:
  - source_candidates
  description: Parse resumes and perform initial qualification checks
  timeout_seconds: 30
- id: match_candidates
  name: Match & Recommend Candidates
  type: ai_agent
  config:
    model_client_id: gpt4_recruiting
  depends_on:
  - screen_candidates
  description: AI-powered matching and ranking of candidates
  user_message: 'Analyze these ${screen_candidates.passed_screening} candidates for
    the position:

    Job Title: ${job_requirements.title}

    Required Skills: ${job_requirements.skills}

    Experience: ${job_requirements.experience} years


    Candidates data is stored in /tmp/screened_candidates.json


    For each candidate, provide:

    1. Match score (0-100)

    2. Key strengths

    3. Potential gaps

    4. Recommendation (Highly Recommended/Recommended/Consider/Not Recommended)


    Return as JSON with candidate_id as keys.

    '
  system_message: You are an expert recruiter. Analyze candidates and provide matching
    scores with detailed reasoning.
- id: assess_candidates
  name: Assess Top Candidates
  type: script
  script: "#!/usr/bin/env python3\nimport json\nimport os\nimport random\nfrom datetime\
    \ import datetime, timedelta\n\nthreshold = int(os.environ.get('assessment_threshold',\
    \ 70))\n\n# Load matched candidates\nwith open('/tmp/screened_candidates.json',\
    \ 'r') as f:\n    candidates = json.load(f)\n\n# Simulate AI matching scores (in\
    \ production, would use actual AI output)\nfor candidate in candidates:\n    candidate['match_score']\
    \ = candidate['screening_results']['screening_score'] + random.randint(-10, 20)\n\
    \    candidate['match_score'] = max(0, min(100, candidate['match_score']))\n\n\
    # Filter candidates above threshold\nassessment_candidates = [c for c in candidates\
    \ if c['match_score'] >= threshold]\n\nprint(f\"Sending assessments to {len(assessment_candidates)}\
    \ candidates (threshold: {threshold})\")\n\nassessments = []\nfor candidate in\
    \ assessment_candidates:\n    assessment = {\n        'candidate_id': candidate['id'],\n\
    \        'candidate_name': candidate['name'],\n        'email': candidate['email'],\n\
    \        'assessment_url': f\"https://assessments.company.com/take/{candidate['id']}\"\
    ,\n        'deadline': (datetime.now() + timedelta(days=3)).isoformat(),\n   \
    \     'status': 'sent',\n        'test_type': 'technical_skills'\n    }\n    assessments.append(assessment)\n\
    \    print(f\"\U0001F4E7 Assessment sent to {candidate['name']} ({candidate['email']})\"\
    )\n\n# Save assessment data\nwith open('/tmp/assessments.json', 'w') as f:\n \
    \   json.dump(assessments, f, indent=2)\n\nprint(f\"\\n\u2705 Assessments sent:\
    \ {len(assessments)}\")\nprint(f\"__OUTPUTS__ {json.dumps({'assessments_sent':\
    \ len(assessments), 'threshold_used': threshold})}\")\n"
  depends_on:
  - match_candidates
  description: Send assessment tests to qualified candidates
  timeout_seconds: 30
- id: schedule_interviews
  name: Schedule Interviews
  type: script
  script: "#!/usr/bin/env python3\nimport json\nimport os\nimport random\nfrom datetime\
    \ import datetime\n\ninterview_slots = json.loads(os.environ.get('interview_slots',\
    \ '[]'))\n\n# Load assessment data\nwith open('/tmp/assessments.json', 'r') as\
    \ f:\n    assessments = json.load(f)\n\n# Simulate assessment completion (in production,\
    \ would check actual results)\ncompleted_assessments = []\nfor assessment in assessments:\n\
    \    if random.random() > 0.3:  # 70% completion rate\n        assessment['status']\
    \ = 'completed'\n        assessment['score'] = random.randint(60, 95)\n      \
    \  if assessment['score'] >= 75:  # Passed assessment\n            completed_assessments.append(assessment)\n\
    \nprint(f\"Scheduling interviews for {len(completed_assessments)} candidates who\
    \ passed assessment\")\n\nscheduled_interviews = []\navailable_slots = interview_slots.copy()\n\
    \nfor assessment in completed_assessments:\n    if available_slots:\n        slot\
    \ = available_slots.pop(0)\n        interview = {\n            'candidate_id':\
    \ assessment['candidate_id'],\n            'candidate_name': assessment['candidate_name'],\n\
    \            'email': assessment['email'],\n            'interview_datetime':\
    \ slot,\n            'interview_type': 'technical_round_1',\n            'duration_minutes':\
    \ 60,\n            'meeting_link': f\"https://meet.company.com/interview/{assessment['candidate_id']}\"\
    ,\n            'calendar_invite_sent': True,\n            'assessment_score':\
    \ assessment['score']\n        }\n        scheduled_interviews.append(interview)\n\
    \        print(f\"\U0001F4C5 Interview scheduled: {assessment['candidate_name']}\
    \ - {slot}\")\n    else:\n        print(f\"\u26A0\uFE0F  No slots available for\
    \ {assessment['candidate_name']}\")\n\n# Save interview schedule\nwith open('/tmp/interview_schedule.json',\
    \ 'w') as f:\n    json.dump(scheduled_interviews, f, indent=2)\n\nprint(f\"\\\
    n\u2705 Interviews scheduled: {len(scheduled_interviews)}\")\nprint(f\"__OUTPUTS__\
    \ {json.dumps({'interviews_scheduled': len(scheduled_interviews), 'slots_remaining':\
    \ len(available_slots)})}\")\n"
  depends_on:
  - assess_candidates
  description: Automatically schedule interviews for candidates who complete assessment
  timeout_seconds: 30
- id: evaluate_candidates
  name: Final Candidate Evaluation
  type: ai_agent
  config:
    model_client_id: gpt4_recruiting
  depends_on:
  - schedule_interviews
  description: Comprehensive evaluation and hiring recommendations
  user_message: 'Provide final evaluation and hiring recommendations for candidates
    who have been interviewed.


    Interview data is in /tmp/interview_schedule.json

    Candidate data is in /tmp/screened_candidates.json


    For each candidate, provide:

    1. Overall rating (1-5 stars)

    2. Strengths summary

    3. Areas of concern

    4. Hiring recommendation (Strong Hire/Hire/Maybe/No Hire)

    5. Suggested next steps


    Also provide an overall summary with top 3 recommendations.

    '
  system_message: You are a senior hiring manager. Provide comprehensive candidate
    evaluations and hiring recommendations.
- id: generate_report
  name: Generate Hiring Report
  type: script
  script: "#!/usr/bin/env python3\nimport json\nimport os\nfrom datetime import datetime\n\
    \n# Compile all metrics\nreport = {\n    'report_id': f\"HIRE_{datetime.now().strftime('%Y%m%d_%H%M%S')}\"\
    ,\n    'job_title': json.loads(os.environ.get('job_requirements', '{}')).get('title'),\n\
    \    'pipeline_metrics': {\n        'total_sourced': int(os.environ.get('source_candidates.candidates_count',\
    \ 0)),\n        'passed_screening': int(os.environ.get('screen_candidates.passed_screening',\
    \ 0)),\n        'sent_assessments': int(os.environ.get('assess_candidates.assessments_sent',\
    \ 0)),\n        'scheduled_interviews': int(os.environ.get('schedule_interviews.interviews_scheduled',\
    \ 0))\n    },\n    'conversion_rates': {},\n    'recommendations': [],\n    'next_steps':\
    \ [\n        \"Conduct scheduled interviews\",\n        \"Perform reference checks\
    \ for top candidates\",\n        \"Prepare offer packages\",\n        \"Plan onboarding\
    \ process\"\n    ]\n}\n\n# Calculate conversion rates\nmetrics = report['pipeline_metrics']\n\
    if metrics['total_sourced'] > 0:\n    report['conversion_rates'] = {\n       \
    \ 'screening_pass_rate': f\"{metrics['passed_screening']/metrics['total_sourced']*100:.1f}%\"\
    ,\n        'assessment_rate': f\"{metrics['sent_assessments']/metrics['total_sourced']*100:.1f}%\"\
    ,\n        'interview_rate': f\"{metrics['scheduled_interviews']/metrics['total_sourced']*100:.1f}%\"\
    \n    }\n\n# Save final report\nwith open('/tmp/hiring_report.json', 'w') as f:\n\
    \    json.dump(report, f, indent=2)\n\nprint(\"\U0001F4CA HIRING PIPELINE REPORT\"\
    )\nprint(\"=\" * 50)\nprint(json.dumps(report, indent=2))\nprint(\"=\" * 50)\n\
    \nprint(f\"__OUTPUTS__ {json.dumps({'report_id': report['report_id'], 'status':\
    \ 'completed'})}\")\n"
  depends_on:
  - evaluate_candidates
  description: Create comprehensive hiring pipeline report
  timeout_seconds: 30
inputs:
  interview_slots:
    type: array
    default:
    - 2025-01-15 10:00
    - 2025-01-15 14:00
    - 2025-01-16 11:00
    required: true
    description: Available interview time slots
  job_requirements:
    type: string
    default: '{"title": "Software Engineer", "skills": ["Python", "APIs"], "experience":
      3}'
    required: true
    description: Job description and requirements in JSON format
  candidate_sources:
    type: array
    default:
    - database
    - linkedin
    required: true
    description: List of candidate sources (LinkedIn, Indeed, Database)
  assessment_threshold:
    type: integer
    default: 70
    required: true
    description: Minimum score to proceed to assessment (0-100)
outputs:
  report_status:
    type: object
    source: generate_report
    description: Hiring pipeline report
  sourcing_status:
    type: object
    source: source_candidates
    description: Candidate sourcing completion status
  screening_status:
    type: object
    source: screen_candidates
    description: Candidate screening results
  assessment_status:
    type: object
    source: assess_candidates
    description: Assessment distribution status
  matching_complete:
    type: object
    source: match_candidates
    description: AI matching and recommendations
  scheduling_status:
    type: object
    source: schedule_interviews
    description: Interview scheduling results
  evaluation_complete:
    type: object
    source: evaluate_candidates
    description: Final candidate evaluation
version: 1.0
namespace: recruitment
description: End-to-end candidate sourcing, screening, matching, assessment, scheduling,
  and evaluation workflow
model_clients:
- id: gpt4_recruiting
  config:
    model: gpt-4o-mini
    api_key: your-api-key-here
    max_tokens: 2000
    temperature: 0.3
  provider: openai
timeout_seconds: 600