Розробка AI-системи планування кар'єри співробітників

AI-система планування кар'єри працівників

Планування кар'єри без даних — це суб'єктивна розмова з менеджером один раз на рік. AI-система аналізує продуктивність, навички, моделі кар'єрного зростання колег та ринкові тренди, пропонуючи конкретні наступні ролі та план розвитку для кожного працівника.

Аналіз кар'єрних траєкторій

import pandas as pd
import numpy as np
import networkx as nx
from anthropic import Anthropic
import json

class CareerPathAnalyzer:
    """Аналіз реальних кар'єрних траєкторій в компанії"""

    def build_career_graph(self, historical_promotions: pd.DataFrame) -> nx.DiGraph:
        """
        Граф переходів між ролями на основі історії компанії.
        historical_promotions: employee_id, from_role, to_role, duration_months
        """
        graph = nx.DiGraph()

        transition_counts = historical_promotions.groupby(
            ['from_role', 'to_role']
        ).agg(
            count=('employee_id', 'count'),
            avg_duration_months=('duration_months', 'mean')
        ).reset_index()

        for _, row in transition_counts.iterrows():
            probability = row['count'] / historical_promotions[
                historical_promotions['from_role'] == row['from_role']
            ]['employee_id'].count()

            graph.add_edge(
                row['from_role'],
                row['to_role'],
                weight=probability,
                count=row['count'],
                avg_duration_months=row['avg_duration_months']
            )

        return graph

    def get_career_paths(self, current_role: str,
                          target_role: str,
                          graph: nx.DiGraph,
                          max_paths: int = 3) -> list[list[str]]:
        """Можливі шляхи від поточної до цільової ролі"""
        try:
            paths = list(nx.all_simple_paths(
                graph, current_role, target_role, cutoff=4
            ))
            # Сортуємо за середною тривалістю кожного кроку
            def path_duration(path):
                total = 0
                for i in range(len(path) - 1):
                    edge = graph.get_edge_data(path[i], path[i+1], {})
                    total += edge.get('avg_duration_months', 18)
                return total

            return sorted(paths, key=path_duration)[:max_paths]
        except (nx.NetworkXNoPath, nx.NodeNotFound):
            return []

    def find_similar_career_profiles(self, employee: dict,
                                      all_employees: pd.DataFrame,
                                      n: int = 10) -> pd.DataFrame:
        """Працівники з подібним кар'єрним профілем як приклади"""
        skill_cols = [c for c in all_employees.columns
                      if c.startswith('skill_')]

        if not skill_cols or employee.get('id') not in all_employees['id'].values:
            return pd.DataFrame()

        employee_row = all_employees[all_employees['id'] == employee['id']].iloc[0]
        emp_vector = employee_row[skill_cols].fillna(0).values

        similarities = []
        for _, row in all_employees.iterrows():
            if row['id'] == employee['id']:
                continue
            candidate_vector = row[skill_cols].fillna(0).values
            sim = np.dot(emp_vector, candidate_vector) / (
                np.linalg.norm(emp_vector) * np.linalg.norm(candidate_vector) + 1e-9
            )
            similarities.append({'id': row['id'], 'role': row.get('current_role'), 'similarity': sim})

        return pd.DataFrame(similarities).nlargest(n, 'similarity')


class CareerPlanGenerator:
    """Генерація персональних планів кар'єрного розвитку"""

    def __init__(self):
        self.llm = Anthropic()
        self.path_analyzer = CareerPathAnalyzer()

    def generate_idp(self, employee: dict,
                      skill_gaps: dict,
                      career_paths: list,
                      market_trends: dict) -> dict:
        """Individual Development Plan з AI-рекомендаціями"""
        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=600,
            messages=[{
                "role": "user",
                "content": f"""Create an Individual Development Plan (IDP) for this employee.

Employee:
- Current role: {employee.get('current_role')}
- Years at company: {employee.get('years_at_company', 0)}
- Performance rating: {employee.get('performance_rating', 3)}/5
- Career aspiration: {employee.get('target_role', 'senior level')}

Skill gaps to address: {list(skill_gaps.keys())[:5]}

Possible career paths: {career_paths[:2]}

Market trends in demand: {list(market_trends.get('growing_skills', []))[:5]}

Write IDP with:
1. Short-term goals (3-6 months)
2. Medium-term goals (6-18 months)
3. Long-term vision (2-3 years)
4. Specific actions (courses, projects, mentoring)
5. Success metrics

Be specific and realistic. 3-4 paragraphs."""
            }]
        )

        idp_text = response.content[0].text

        # Структурований план
        return {
            'employee_id': employee.get('id'),
            'created_at': pd.Timestamp.now().isoformat(),
            'target_role': employee.get('target_role'),
            'estimated_timeline_months': self._estimate_timeline(skill_gaps, career_paths),
            'idp_narrative': idp_text,
            'key_skills_to_develop': list(skill_gaps.keys())[:5],
            'next_review_date': (pd.Timestamp.now() + pd.DateOffset(months=3)).strftime('%Y-%m-%d')
        }

    def _estimate_timeline(self, skill_gaps: dict, paths: list) -> int:
        """Оцінка реалістичного горизонту"""
        if not paths:
            return 24
        # Середня тривалість шляху + допуск на закриття пропусків
        high_gaps = sum(1 for g in skill_gaps.values() if g.get('gap', 0) >= 2)
        base_months = 18
        return base_months + high_gaps * 3


class RetentionRiskPredictor:
    """Прогнозування ризику відходу працівника"""

    def predict_flight_risk(self, employee: dict,
                             engagement_data: dict,
                             market_data: dict) -> dict:
        """Ймовірність звільнення протягом 12 місяців"""
        risk_factors = []
        risk_score = 0.0

        # Фактори ризику
        if engagement_data.get('engagement_score', 3) < 3:
            risk_score += 0.25
            risk_factors.append('Low engagement score')

        if employee.get('months_in_current_role', 0) > 24:
            risk_score += 0.15
            risk_factors.append('Long time in one role without growth')

        if employee.get('performance_rating', 3) > 4 and employee.get('comp_percentile', 50) < 60:
            risk_score += 0.20
            risk_factors.append('High performance, below-market compensation')

        market_salary_gap = (
            market_data.get('median_salary', 0) - employee.get('salary', 0)
        ) / max(market_data.get('median_salary', 1), 1)

        if market_salary_gap > 0.15:
            risk_score += 0.20
            risk_factors.append(f'Below market by {market_salary_gap:.0%}')

        if employee.get('years_at_company', 0) in [2, 3]:
            risk_score += 0.10
            risk_factors.append('Typical tenure before job change (2-3 years)')

        risk_score = min(risk_score, 1.0)

        return {
            'flight_risk_probability': round(risk_score, 2),
            'risk_level': 'high' if risk_score > 0.5 else 'medium' if risk_score > 0.3 else 'low',
            'key_factors': risk_factors,
            'recommended_action': (
                'Immediate meeting with manager + compensation review offer' if risk_score > 0.6
                else 'Career conversation + development plan' if risk_score > 0.4
                else 'Routine check-in'
            )
        }

AI-система планування кар'єри зменшує добровільну плинність на 15-25% при активному використанні. Ключ до успіху: менеджери повинні регулярно працювати з рекомендаціями системи, а не просто мати доступ до них.