emmixam/ai_workflow_discovery_agent

1import pandas as pd2import numpy as np3from sentence_transformers import SentenceTransformer4from sklearn.cluster import HDBSCAN5from sklearn.metrics.pairwise import cosine_distances6from typing import List, Dict, Any, Tuple, Optional7 8TEXT_COLUMN_CANDIDATES = ['content', 'action', 'etape', 'category', 'description', 'titre', 'nom', 'label']9DURATION_COLUMN_CANDIDATES = ['duration_minutes', 'duree_moyenne_jours', 'resolution_time_hours', 'duree_estimee']10START_COLUMN_CANDIDATES = ['timestamp', 'timestamp_start', 'created', 'date_debut']11END_COLUMN_CANDIDATES = ['timestamp_end', 'resolved', 'date_fin']12 13class DataEngine:14    def __init__(self, embedding_model: str = 'all-MiniLM-L6-v2'):15        """16        Initialise le moteur avec un modèle d'embedding léger exécutable sur CPU.17        all-MiniLM-L6-v2 génère des vecteurs de 384 dimensions.18        """19        self.encoder = SentenceTransformer(embedding_model)20        self.current_embeddings = None21 22    @staticmethod23    def _extract_list(raw_data: Any) -> List[Dict[str, Any]]:24        """25        Détecte automatiquement la liste principale dans un JSON hétérogène.26        Parcourt les valeurs racine et retourne la première liste de dicts trouvée.27        """28        if isinstance(raw_data, list):29            return raw_data30        if isinstance(raw_data, dict):31            for value in raw_data.values():32                if isinstance(value, list) and len(value) > 0 and isinstance(value[0], dict):33                    return value34        return []35 36    def normalize(self, raw_data: Any, mapping_config: Dict[str, str] = {}) -> pd.DataFrame:37        """38        Transforme les données hétérogènes en DataFrame selon le Schéma Pivot.39        Détecte automatiquement les colonnes texte et durée.40        """41        if mapping_config is None:42            mapping_config = {}43 44        records = self._extract_list(raw_data)45        if not records:46            raise ValueError("Aucune liste de données détectable dans le JSON fourni.")47 48        df = pd.DataFrame(records)49 50        # Renommage optionnel si mapping fourni51        if mapping_config:52            df = df.rename(columns=mapping_config)53 54        # Détection automatique colonne texte → renommée 'content'55        if 'content' not in df.columns:56            for candidate in TEXT_COLUMN_CANDIDATES:57                if candidate in df.columns:58                    df = df.rename(columns={candidate: 'content'})59                    break60 61        if 'content' not in df.columns:62            raise ValueError(f"Aucune colonne texte détectée. Colonnes disponibles : {list(df.columns)}")63 64        # Détection automatique durée65        if 'duration_minutes' not in df.columns:66            for candidate in DURATION_COLUMN_CANDIDATES:67                if candidate in df.columns:68                    df['duration_minutes'] = pd.to_numeric(df[candidate], errors='coerce')69                    break70 71        # Calcul durée depuis timestamps si disponibles72        start_col = next((c for c in START_COLUMN_CANDIDATES if c in df.columns), None)73        end_col = next((c for c in END_COLUMN_CANDIDATES if c in df.columns), None)74        if start_col and end_col:75            df['timestamp_start'] = pd.to_datetime(df[start_col], errors='coerce')76            df['timestamp_end'] = pd.to_datetime(df[end_col], errors='coerce')77            mask = df['timestamp_start'].notna() & df['timestamp_end'].notna()78            df.loc[mask, 'duration_minutes'] = (79                df.loc[mask, 'timestamp_end'] - df.loc[mask, 'timestamp_start']80            ).dt.total_seconds() / 60.081 82        if 'timestamp' in df.columns:83            df['timestamp'] = pd.to_datetime(df['timestamp'], errors='coerce')84 85        df = df.dropna(subset=['content']).reset_index(drop=True)86        return df87 88    def vectorize_and_cluster(self, df: pd.DataFrame, min_cluster_size: Optional[int] = None, epsilon: float = 0.5) -> pd.DataFrame:89        textes = df['content'].tolist()90        embeddings = self.encoder.encode(textes, show_progress_bar=False)91 92        n_samples = len(df)93        if min_cluster_size is None:94            # Heuristique logarithmique pour ajustement dynamique de la densité95            # Prévient la fragmentation excessive sur de larges datasets96            min_cluster_size = max(2, int(np.log(n_samples) * 1.5)) if n_samples > 0 else 297 98        clusterer = HDBSCAN(99            min_cluster_size=min_cluster_size,100            metric='euclidean',101            cluster_selection_epsilon=epsilon,102            copy=True  # <-- AJOUT : Rend le code compatible avec Scikit-Learn 1.10+103        )104        df['cluster_id'] = clusterer.fit_predict(embeddings)105        self.current_embeddings = embeddings106 107        return df108 109    def extract_top_representatives(self, df_cluster: pd.DataFrame, cluster_idx: int) -> List[str]:110        """111        Trouve les 3 textes les plus proches du centre mathématique du cluster.112        """113        idx_in_cluster = df_cluster.index.tolist()114        cluster_embeddings = self.current_embeddings[idx_in_cluster]115 116        if len(cluster_embeddings) <= 3:117            return df_cluster['content'].tolist()118 119        # Calcul du centroïde (moyenne vectorielle du cluster)120        centroid = np.mean(cluster_embeddings, axis=0).reshape(1, -1)121 122        # Distances cosinus entre les points du cluster et le centroïde123        distances = cosine_distances(cluster_embeddings, centroid).flatten()124 125        # Récupère les index des 3 textes les plus proches (distance minimale)126        top_3_idx_local = np.argsort(distances)[:3]127 128        representatives = [df_cluster.iloc[i]['content'] for i in top_3_idx_local]129        return representatives130 131    def generate_payload(self, df: pd.DataFrame, source_name: str) -> Dict[str, Any]:132        """133        Agrège les métriques mathématiques et génère le payload minimaliste pour le LLM.134        """135        payload = {136            "source": source_name,137            "metriques_globales": {138                "volume_total_lignes": len(df),139                "temps_total_analyse_minutes": df['duration_minutes'].sum() if 'duration_minutes' in df.columns else 0140            },141            "clusters_repetitifs": []142        }143 144        # Ignorer le cluster -1 qui représente le bruit non répétitif selon HDBSCAN145        valid_clusters = df[df['cluster_id'] != -1]146        grouped = valid_clusters.groupby('cluster_id')147 148        for cluster_id, group in grouped:149            frequence = len(group)150            temps_perdu = group['duration_minutes'].sum() if 'duration_minutes' in df.columns else None151            exemples = self.extract_top_representatives(group, cluster_id)152 153            cluster_data = {154                "cluster_id": f"C{cluster_id}",155                "frequence": frequence,156                "temps_total_perdu_minutes": temps_perdu,157                "exemples_representatifs": exemples158            }159            payload["clusters_repetitifs"].append(cluster_data)160 161        return payload162 163    def compute_roi(self, payload: dict, taux_horaire: float) -> dict:164        raise RuntimeError(165            "DataEngine.compute_roi() est obsolète. "166            "Utilise compute_roi_from_time_report() dans business_metrics.py."167        )168 169    def process_pipeline(170        self,171        raw_data: Any,172        mapping_config: Dict[str, str] = None,173        source_name: str = "",174        taux_horaire: float = 25.0,175        min_cluster_size: Optional[int] = None,176        epsilon: float = 0.5177    ) -> Tuple[Dict[str, Any], Dict[str, Any]]:178        raise RuntimeError(179            "DataEngine.process_pipeline() est obsolète. "180            "Le pipeline doit appeler séparément: normalize() -> vectorize_and_cluster() "181            "-> generate_payload(), puis AgentTemps + business_metrics."182        )183