Разработка 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 in Russian 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('Низкий engagement score')

        if employee.get('months_in_current_role', 0) > 24:
            risk_score += 0.15
            risk_factors.append('Долго в одной роли без роста')

        if employee.get('performance_rating', 3) > 4 and employee.get('comp_percentile', 50) < 60:
            risk_score += 0.20
            risk_factors.append('Высокая производительность, недооплата рынка')

        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'Ниже рынка на {market_salary_gap:.0%}')

        if employee.get('years_at_company', 0) in [2, 3]:
            risk_score += 0.10
            risk_factors.append('Типичное время смены работы (2-3 года)')

        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': (
                'Немедленная встреча с менеджером + оффер пересмотра' if risk_score > 0.6
                else 'Карьерный разговор + план развития' if risk_score > 0.4
                else 'Плановый check-in'
            )
        }

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