Hiring Engineers Who Ship: A Practical Framework for AI Teams

## Senior ML Engineer - Recommendation Systems

### Daily Work (80% of time):
- Debug why model accuracy dropped 3% overnight
- Optimize feature pipeline reducing latency from 200ms to 50ms
- Review PRs for data quality checks
- Investigate data drift in production models
- Write integration tests for new features

### Weekly Work (15% of time):
- Experiment with new model architectures
- Analyze A/B test results
- Mentor junior engineers
- Participate in on-call rotation

### Monthly Work (5% of time):
- Present findings to stakeholders
- Contribute to technical strategy
- Evaluate new tools and frameworks

"""
ML Engineer Assessment: Feature Pipeline Optimization

Context: You have a feature pipeline that processes user events
for a recommendation system. It's currently too slow for production.

Task: Optimize this pipeline to run in under 100ms for 95% of requests.
You may refactor anything, but must maintain the same output format.

Time: 2 hours (paid at $250/hour)
"""

import pandas as pd
import numpy as np
from typing import Dict, List
import time

class FeaturePipeline:
    def __init__(self):
        # Simulated model embeddings (in production, loaded from model file)
        self.user_embeddings = np.random.randn(100000, 128)
        self.item_embeddings = np.random.randn(50000, 128)
        
    def process_request(self, user_id: int, 
                        recent_items: List[int],
                        context: Dict) -> Dict:
        """
        Current implementation - YOUR TASK: Optimize this
        """
        start = time.time()
        
        # Fetch user embedding (simulated DB call)
        time.sleep(0.05)  # Simulate network latency
        user_emb = self.user_embeddings[user_id]
        
        # Process recent items sequentially
        item_features = []
        for item_id in recent_items:
            time.sleep(0.01)  # Simulate DB call
            item_emb = self.item_embeddings[item_id]
            similarity = np.dot(user_emb, item_emb)
            item_features.append({
                'item_id': item_id,
                'similarity': similarity,
                'interaction_time': context.get('timestamp')
            })
        
        # Aggregate features (inefficient implementation)
        df = pd.DataFrame(item_features)
        aggregated = {
            'mean_similarity': df['similarity'].mean(),
            'max_similarity': df['similarity'].max(),
            'item_count': len(df),
            'user_id': user_id
        }
        
        # More processing...
        time.sleep(0.02)
        
        return {
            'features': aggregated,
            'latency_ms': (time.time() - start) * 1000
        }

# Test harness
def evaluate_solution(pipeline: FeaturePipeline):
    """We'll use this to evaluate your optimized version"""
    latencies = []
    for _ in range(100):
        user_id = np.random.randint(0, 100000)
        recent_items = np.random.randint(0, 50000, size=20).tolist()
        context = {'timestamp': time.time()}
        
        result = pipeline.process_request(user_id, recent_items, context)
        latencies.append(result['latency_ms'])
    
    p95 = np.percentile(latencies, 95)
    print(f"P95 Latency: {p95:.2f}ms")
    return p95 < 100  # Target: under 100ms

# Your optimized implementation goes here...

def calculate_offer(candidate_profile):
    """
    Data-driven compensation calculation
    Fair market value based on role, skills, and location
    """
    # Start with market rate for level and location
    base_salary = get_market_rate(
        level=candidate_profile['level'],  # junior, mid, senior, staff, principal
        location=candidate_profile['location'],
        market_data_source='levels.fyi'  # or salary surveys
    )

    # Adjust for specific high-demand skills
    skill_multipliers = {
        'production_ml': 1.1,
        'distributed_systems': 1.05,
        'deep_learning': 1.15,
        'mlops': 1.1,
        'published_research': 1.05
    }

    for skill in candidate_profile['verified_skills']:
        if skill in skill_multipliers:
            base_salary *= skill_multipliers[skill]

    # Equity scales with level
    equity_percentage = {
        'junior': 0.05,
        'mid': 0.1,
        'senior': 0.2,
        'staff': 0.35,
        'principal': 0.5
    }[candidate_profile['level']]

    return {
        'base': int(base_salary),
        'equity': equity_percentage,
        'signing_bonus': int(base_salary * 0.1),
        'annual_bonus_target': 0.15
    }

## Week 1: Environment Setup and Context
- Day 1: Machine setup, accounts, documentation access
- Day 2: Run existing models locally, understand inference pipeline
- Day 3: Deploy a small change to production (with guidance)
- Day 4: Shadow on-call engineer, review recent incidents
- Day 5: First code review, first PR

## Week 2: First Contribution
- Small, well-defined task that touches the full stack
- Pair with senior engineer for guidance
- Goal: Merged PR that adds value

## Week 4: Independent Contribution
- Own a small feature or improvement
- Present findings at team meeting
- Begin participating in on-call rotation

## Week 8: Full Productivity
- Leading small projects
- Mentoring newer team members
- Contributing to technical decisions

class HiringMetrics:
    def __init__(self):
        self.candidates = []
        
    def track_candidate(self, candidate):
        return {
            'time_to_fill': days_from_posting_to_accept,
            'assessment_score': technical_assessment_result,
            'interview_score': average_interviewer_rating,
            'offer_acceptance_rate': offers_accepted / offers_made,
            '90_day_retention': still_employed_after_90_days,
            'performance_rating': manager_rating_at_6_months,
            'referral_generation': referrals_from_this_hire
        }
    
    def analyze_channel_effectiveness(self):
        """Which sourcing channels produce best hires?"""
        channels = defaultdict(list)
        for candidate in self.candidates:
            channels[candidate.source].append(candidate.outcome)
        
        return {
            channel: {
                'conversion_rate': sum(1 for o in outcomes if o == 'hired') / len(outcomes),
                'quality_score': np.mean([c.performance_rating for c in outcomes if c.performance_rating]),
                'cost_per_hire': total_cost / hires
            }
            for channel, outcomes in channels.items()
        }

def calculate_turnover_cost(role, salary):
    """
    Real cost of losing an AI engineer
    """
    recruitment_cost = salary * 0.25  # Recruiter fees
    opportunity_cost = salary * 0.5   # Lost productivity during search
    onboarding_cost = salary * 0.33   # 4 months to full productivity
    knowledge_loss = salary * 0.5     # Institutional knowledge
    
    total_cost = recruitment_cost + opportunity_cost + onboarding_cost + knowledge_loss
    return total_cost  # Often 1.5-2x annual salary