KI Prozessoptimierung: 35% Effizienz in 60 Tagen [ROI-Methodik 2025]

Process Mining Stack:
  Tool: Celonis (Open Source Alternative: pm4py)
  Datenquellen:
    - ERP-System (Aufträge, Lieferungen)
    - CRM (Kundenanfragen, Angebote)
    - Email-System (Kommunikation)
    - Excel-Dateien (manuelle Nachverfolgung)

Analysierte Prozesse:
  - Auftragserfassung (1.200 Aufträge/Monat)
  - Angebotserstellung (3.500 Anfragen/Monat)
  - Reklamationsbearbeitung (180 Fälle/Monat)
  - Lieferantenabwicklung (450 Bestellungen/Monat)

# Process Mining Ergebnisse (vereinfacht)

import pm4py

# Event-Log aus ERP laden
event_log = pm4py.read_xes('auftragsabwicklung.xes')

# Prozessmodell entdecken
process_model = pm4py.discover_petri_net_inductive(event_log)

# Bottleneck-Analyse
bottlenecks = pm4py.analysis.bottleneck_analysis(event_log)
# Output:
# 1. Manuelle Bonitätsprüfung: 4,2 Stunden Wartezeit (35% aller Fälle)
# 2. Verfügbarkeitscheck: 3,8 Stunden (manuell per Email)
# 3. Preiskalkulation: 2,1 Stunden (Excel-basiert)
# 4. Auftragsbestätigung: 1,5 Stunden (manuelle Erstellung)

# Prozessvarianten-Analyse
variants = pm4py.get_variants(event_log)
# Ergebnis: 47 verschiedene Varianten (!) statt Standard-Prozess
# → 62% aller Aufträge laufen NICHT nach Standard
# → Ursache: Ausnahmen nicht systematisch behandelt

# Automatische Auftragserfassung mit NLP

from transformers import pipeline
import re

class IntelligentOrderProcessor:
    def __init__(self):
        self.nlp = pipeline("ner", model="deepset/gbert-large")
        self.classifier = pipeline("zero-shot-classification")
        
    def process_email_order(self, email_body, sender):
        """
        Extrahiert Auftragsdaten aus Email automatisch
        """
        # 1. Named Entity Recognition
        entities = self.nlp(email_body)
        
        # 2. Strukturierte Daten extrahieren
        order_data = {
            'artikel': self.extract_products(email_body),
            'menge': self.extract_quantities(email_body),
            'lieferdatum': self.extract_delivery_date(email_body),
            'lieferadresse': self.extract_address(entities),
            'sonderwünsche': self.extract_special_requests(email_body)
        }
        
        # 3. Priorität klassifizieren
        priority = self.classifier(
            email_body,
            candidate_labels=["eilig", "standard", "termingebunden"]
        )
        
        # 4. Bonitätsprüfung automatisch
        credit_check = self.automated_credit_check(sender)
        
        # 5. Verfügbarkeit prüfen
        availability = self.check_availability(order_data['artikel'])
        
        # 6. Auftrag direkt anlegen (wenn alle Checks OK)
        if credit_check['approved'] and availability['all_available']:
            order_id = self.create_order_in_erp(order_data)
            self.send_confirmation_email(sender, order_id, order_data)
            return {'status': 'auto_processed', 'order_id': order_id}
        else:
            # An Sachbearbeiter mit Vorschlag
            return {
                'status': 'needs_review',
                'order_data': order_data,
                'issues': credit_check['issues'] + availability['missing_items']
            }

# ML-basierte Preisoptimierung

from sklearn.ensemble import RandomForestRegressor
import pandas as pd

class PredictivePricingEngine:
    def __init__(self):
        self.model = self.load_trained_model()
        
    def calculate_optimal_price(self, product, customer, order_context):
        """
        Berechnet optimalen Preis basierend auf:
        - Historische Deals
        - Kundenprofil (Volumen, Zahlungsmoral)
        - Marktbedingungen
        - Wettbewerberpreise
        """
        features = pd.DataFrame({
            'produkt_kategorie': [product['category']],
            'menge': [order_context['quantity']],
            'kunde_umsatz_12m': [customer['revenue_12m']],
            'kunde_zahlungsziel': [customer['payment_terms']],
            'wettbewerb_preis_avg': [self.get_competitor_price(product)],
            'saison': [self.get_season()],
            'lagerbestand': [product['stock_level']]
        })
        
        # ML-Vorhersage
        predicted_price = self.model.predict(features)[0]
        
        # Profitabilität prüfen
        min_price = product['cost'] * 1.15  # 15% Mindestmarge
        
        if predicted_price < min_price:
            return {
                'price': min_price,
                'confidence': 'adjusted_for_margin',
                'discount': 0
            }
        else:
            return {
                'price': predicted_price,
                'confidence': self.model.score_,
                'discount': (product['list_price'] - predicted_price) / product['list_price']
            }

Reklamations-Workflow (vorher):
  1. Email kommt rein → Postfach (2-48h Wartezeit)
  2. Mitarbeiter liest → kategorisiert manuell (15 Min)
  3. Prüfung in verschiedenen Systemen (25 Min)
  4. Rücksprache mit Kollegen (30 Min)
  5. Entscheidung & Antwort (20 Min)
  Total: 90 Minuten + bis zu 48h Wartezeit

Reklamations-Workflow (nachher - KI):
  1. Email kommt rein → Automatische Kategorisierung (5 Sek)
  2. Prüfung gegen Auftrag/Lieferschein/Zahlungen (10 Sek)
  3. Entscheidungslogik (ML-basiert):
     - Wert < €500 + bestehender Kunde → Auto-Kulanz
     - Wert €500-2000 + klare Beweislage → Auto-Erstattung
     - Wert > €2000 oder unklar → An Mitarbeiter mit Empfehlung
  4. Automatische Antwort + Prozessauslösung (Gutschrift/Retoure)
  Total: 45 Sekunden (bei 78% der Fälle) oder 12 Min (mit Mitarbeiter)

Open Source Stack:
  - pm4py (Process Mining in Python)
  - ProM (Process Mining Framework)
  - Celonis Free Trial (30 Tage)

Data Sources:
  - ERP: Aufträge, Rechnungen, Lieferungen
  - CRM: Leads, Opportunities, Support-Tickets
  - Email: Outlook/Gmail via API
  - Excel/CSV: Manuelle Prozesse dokumentieren

# Beispiel: Automatische Rechnungsprüfung

from rpaframework.windows import Windows
import pandas as pd

class InvoiceAutomation:
    def __init__(self):
        self.windows = Windows()
        
    def process_incoming_invoices(self, invoice_folder):
        """
        Automatisiert Rechnungsprüfung:
        1. PDF aus Email-Anhang extrahieren
        2. OCR + Datenextraktion
        3. Abgleich mit Bestellung im ERP
        4. Bei Match: Auto-Buchung
        5. Bei Differenz: Email an Einkauf mit Übersicht
        """
        invoices = self.scan_folder(invoice_folder)
        
        for invoice_pdf in invoices:
            # OCR mit Tesseract oder Cloud-API
            invoice_data = self.extract_invoice_data(invoice_pdf)
            
            # ERP-Abgleich
            po_data = self.lookup_purchase_order(invoice_data['po_number'])
            
            if self.validate_invoice(invoice_data, po_data):
                # Automatische Buchung
                self.post_to_accounting(invoice_data)
                self.send_confirmation(invoice_data['supplier'])
            else:
                # Abweichung → manuell
                self.create_exception_ticket(invoice_data, po_data)

# Lead-Scoring mit Random Forest

from sklearn.ensemble import RandomForestClassifier
import pandas as pd

# Training Data (historische Leads mit Outcome)
df_leads = pd.read_csv('leads_historical.csv')

features = [
    'industrie', 'mitarbeiter_anzahl', 'umsatz_geschätzt',
    'web_visits_30d', 'downloads', 'demo_requests',
    'email_engagement_score', 'linkedin_activity'
]

X = df_leads[features]
y = df_leads['converted']  # 1 = Deal, 0 = Lost

# Model trainieren
model = RandomForestClassifier(n_estimators=100, max_depth=10)
model.fit(X, y)

# Feature Importance analysieren
importance = pd.DataFrame({
    'feature': features,
    'importance': model.feature_importances_
}).sort_values('importance', ascending=False)

# Output:
# 1. demo_requests: 0.35 (wichtigster Faktor!)
# 2. email_engagement_score: 0.22
# 3. umsatz_geschätzt: 0.18
# ... 

# Prediction für neue Leads
new_leads = pd.read_csv('leads_this_week.csv')
new_leads['score'] = model.predict_proba(new_leads[features])[:, 1]
new_leads['priority'] = pd.cut(
    new_leads['score'],
    bins=[0, 0.3, 0.7, 1.0],
    labels=['low', 'medium', 'high']
)

# Ergebnis: Vertrieb fokussiert auf High-Priority Leads
# → 45% höhere Conversion-Rate!

# Kündigungsrisiko vorhersagen

from xgboost import XGBClassifier

# Features für Churn-Modell
churn_features = [
    'kunde_seit_monaten',
    'umsatz_12m', 'umsatz_trend',
    'anzahl_reklamationen', 'support_tickets',
    'zahlungsverzug_tage', 'rabatt_prozent',
    'letzte_bestellung_tage', 'produkt_diversität'
]

# Modell trainieren
churn_model = XGBClassifier(n_estimators=200, max_depth=6)
churn_model.fit(X_train, y_train)

# Prognose für aktuelle Kunden
customers = pd.read_sql('SELECT * FROM customers', con=db_connection)
customers['churn_risk'] = churn_model.predict_proba(customers[churn_features])[:, 1]

# Automatische Aktionen
high_risk = customers[customers['churn_risk'] > 0.7]

for idx, customer in high_risk.iterrows():
    # Automatisch: Key-Account-Manager informieren
    send_alert(customer['account_manager'], customer)
    
    # Automatisch: Rabatt-Angebot vorbereiten
    if customer['umsatz_12m'] > 50000:
        prepare_retention_offer(customer, discount=10%)

API-Gateway (FastAPI):
  Endpoints:
    - POST /api/process/order → Intelligente Auftragsverarbeitung
    - POST /api/score/lead → Lead-Scoring
    - GET /api/predict/churn/{customer_id} → Churn-Risk
    - POST /api/price/calculate → Predictive Pricing

Security:
  - OAuth 2.0 Authentication
  - Rate Limiting (100 req/min)
  - Audit-Logging (jede Request)

Monitoring:
  - Prometheus + Grafana
  - Alerting bei Prediction-Drift
  - Daily Model-Performance-Report

Kommunikation:
  - Kick-Off Workshop (alle Stakeholder)
  - Wöchentliche Updates (Email + Teams)
  - Success Stories teilen (Quick-Wins)

Training:
  - Admin-Schulung (IT): 4h
  - Power-User-Schulung (Schlüsselpersonen): 3h
  - End-User-Training (alle): 2h
  - Video-Tutorials (Self-Service)

Support:
  - Hotline 8-18 Uhr (erste 4 Wochen)
  - Confluence Wiki (Dokumentation)
  - Feedback-Formular (Verbesserungen)

Executive Dashboard (monatlich):
  KPIs:
    - Zeitersparnis in Stunden/Monat
    - Fehlerreduktion in %
    - ROI aktuell vs Target
    - Mitarbeiter-Akzeptanz (Survey-Score)
  
  Format: 1-Seiter mit Ampeln (Rot/Gelb/Grün)
  Review: 30 Min mit Geschäftsführung

# Data Quality Assessment

import pandas as pd
import numpy as np

def assess_data_quality(df, process_name):
    """
    Schneller DQ-Check für Process Mining/ML
    """
    report = {
        'process': process_name,
        'total_records': len(df),
        'completeness': {},
        'consistency': {},
        'timeliness': {}
    }
    
    # Vollständigkeit
    for col in df.columns:
        missing_pct = df[col].isnull().sum() / len(df) * 100
        report['completeness'][col] = {
            'missing_%': round(missing_pct, 2),
            'status': 'OK' if missing_pct < 5 else 'CRITICAL'
        }
    
    # Konsistenz (Duplikate)
    duplicates = df.duplicated().sum()
    report['consistency']['duplicates'] = {
        'count': duplicates,
        'pct': round(duplicates / len(df) * 100, 2)
    }
    
    # Aktualität (bei timestamp-column)
    if 'timestamp' in df.columns:
        latest = df['timestamp'].max()
        days_old = (pd.Timestamp.now() - latest).days
        report['timeliness'] = {
            'latest_record': str(latest),
            'days_old': days_old,
            'status': 'OK' if days_old < 7 else 'WARNING'
        }
    
    return report

# Ausführen
quality_report = assess_data_quality(orders_df, 'Auftragsabwicklung')

# Kritische Issues beheben BEVOR Process Mining!

Entscheidungs-Matrix:
  Automatisch (100%):
    - Wert < €500
    - Confidence > 95%
    - Standard-Fall (kein Compliance-Risiko)
  
  Mit Review (Mensch prüft):
    - Wert €500-5.000
    - Confidence 85-95%
    - Neue Kundensituation
  
  Manuell (KI schlägt vor):
    - Wert > €5.000
    - Confidence < 85%
    - Compliance-relevant (DSGVO, Verträge)

# Wöchentlicher ML-Health-Check

def weekly_model_health_check(model_name, predictions_df):
    """
    Automatischer Health-Check für ML-Modelle
    """
    health_report = {
        'model': model_name,
        'week': datetime.now().isocalendar()[1],
        'checks': {}
    }
    
    # 1. Prediction Distribution Drift
    current_dist = predictions_df['prediction'].describe()
    baseline_dist = load_baseline_distribution(model_name)
    
    drift_score = calculate_kl_divergence(current_dist, baseline_dist)
    health_report['checks']['distribution_drift'] = {
        'score': drift_score,
        'status': 'OK' if drift_score < 0.1 else 'ALERT'
    }
    
    # 2. Prediction Confidence
    avg_confidence = predictions_df['confidence'].mean()
    health_report['checks']['avg_confidence'] = {
        'value': round(avg_confidence, 3),
        'status': 'OK' if avg_confidence > 0.8 else 'WARNING'
    }
    
    # 3. Prediction Volume
    prediction_count = len(predictions_df)
    expected_count = get_expected_weekly_volume(model_name)
    
    volume_diff = abs(prediction_count - expected_count) / expected_count
    health_report['checks']['volume_anomaly'] = {
        'current': prediction_count,
        'expected': expected_count,
        'diff_%': round(volume_diff * 100, 1),
        'status': 'OK' if volume_diff < 0.2 else 'ALERT'
    }
    
    # Alert bei kritischen Issues
    if any(check['status'] == 'ALERT' for check in health_report['checks'].values()):
        send_alert_to_data_team(health_report)
    
    return health_report