Praktische Übungen: Hands-on KI-Automatisierung

Überblick

Diese Übungen führen Sie schrittweise durch die Implementierung einer vollständigen KI-Automatisierungslösung für das Finanzwesen. Von der ersten Analyse bis zur produktiven Lösung.

---

🎯 Workshop 1: Use Case Bewertung & Priorisierung

Aufgabe: Bewerten Sie Ihre eigenen Prozesse

Ziel: Identifizieren Sie den besten Automatisierungskandidaten in Ihrem Unternehmen.

Schritt 1: Prozessinventar erstellen

Füllen Sie diese Tabelle für 5 Prozesse aus Ihrem Arbeitsbereich aus:

Prozess	Häufigkeit/Monat	Zeit pro Vorgang	Fehlerrate	Regelbasiert (1-5)
Rechnungsverarbeitung	500	20 Min	5%	4
Zahlungsfreigaben	200	15 Min	2%	5
Monatsabschluss	1	8 Std	10%	3
Mahnwesen	150	10 Min	3%	4
Kostenstellen-Zuordnung	300	5 Min	8%	2

Schritt 2: Bewertungsmatrix anwenden

def calculate_automation_score(process_data):
    """Berechnet Automatisierungs-Score für Prozess"""

    weights = {
        'rule_based': 0.25,
        'volume': 0.20,
        'time_intensity': 0.20,
        'error_rate': 0.15,
        'standardization': 0.10,
        'data_quality': 0.10
    }

    # Normalisierte Scores (1-5)
    scores = {
        'rule_based': process_data['rule_based_score'],
        'volume': min(5, process_data['monthly_volume'] / 100),
        'time_intensity': min(5, process_data['time_per_process'] / 10),
        'error_rate': 5 - (process_data['error_rate'] * 50),  # Höhere Fehlerrate = höhere Priorität
        'standardization': process_data['standardization_score'],
        'data_quality': process_data['data_quality_score']
    }

    weighted_score = sum(scores[key] * weights[key] for key in scores)
    return round(weighted_score, 2)

# Beispiel-Berechnung
rechnungsverarbeitung = {
    'rule_based_score': 4,
    'monthly_volume': 500,
    'time_per_process': 20,
    'error_rate': 0.05,
    'standardization_score': 4,
    'data_quality_score': 4
}

score = calculate_automation_score(rechnungsverarbeitung)
print(f"Automatisierungs-Score: {score}/5")

Übung 1.1: Ihre Bewertung

1. Wählen Sie 3 Prozesse aus Ihrem Unternehmen
2. Bewerten Sie jeden Prozess mit obiger Formel
3. Erstellen Sie eine Prioritätenliste

Lösung: Der Prozess mit dem höchsten Score ist Ihr bester Automatisierungskandidat.

---

🛠️ Workshop 2: Minimaler Rechnungs-Extraktor

Aufgabe: Bauen Sie Ihren ersten KI-Extraktor

Ziel: Erstellen Sie einen funktionsfähigen PDF-Rechnungsextraktor.

Schritt 1: Setup & Dependencies

# Neue virtuelle Umgebung
python -m venv workshop_env
source workshop_env/bin/activate

# Dependencies installieren
pip install streamlit PyPDF2 openai python-dotenv

Schritt 2: Basis-Extraktor implementieren

# simple_extractor.py
import streamlit as st
import PyPDF2
import openai
import os
from dotenv import load_dotenv
import re

load_dotenv()

class SimpleInvoiceExtractor:
    def __init__(self):
        openai.api_key = os.getenv("OPENAI_API_KEY")

    def extract_text_from_pdf(self, pdf_file):
        """Extrahiert Text aus PDF"""
        reader = PyPDF2.PdfReader(pdf_file)
        text = ""
        for page in reader.pages:
            text += page.extract_text()
        return text

    def extract_invoice_data(self, text):
        """Extrahiert Rechnungsdaten mit OpenAI"""

        prompt = f"""
        Analysiere die folgende deutsche Rechnung und extrahiere diese Informationen:

        - Rechnungsnummer
        - Lieferant/Firma
        - Rechnungsdatum  
        - Nettobetrag
        - MwSt-Betrag
        - Gesamtbetrag

        Text: {text[:2000]}

        Antwort als JSON:
        """

        try:
            response = openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=[{"role": "user", "content": prompt}],
                temperature=0
            )

            return response.choices[0].message.content

        except Exception as e:
            return f"Fehler: {str(e)}"

# Streamlit App
def main():
    st.title("🧾 Einfacher Rechnungsextraktor")

    extractor = SimpleInvoiceExtractor()

    uploaded_file = st.file_uploader("PDF-Rechnung hochladen", type="pdf")

    if uploaded_file:
        st.success(f"Datei hochgeladen: {uploaded_file.name}")

        if st.button("Rechnung analysieren"):
            with st.spinner("Analysiere Rechnung..."):
                # Text extrahieren
                text = extractor.extract_text_from_pdf(uploaded_file)

                # Daten extrahieren
                extracted_data = extractor.extract_invoice_data(text)

                st.subheader("Extrahierte Daten:")
                st.code(extracted_data)

                st.subheader("Roher PDF-Text:")
                st.text_area("Text", text[:1000], height=200)

if __name__ == "__main__":
    main()

Übung 2.1: Testen Sie den Extraktor

1. Erstellen Sie eine .env Datei mit Ihrem OpenAI API Key
2. Starten Sie die App: streamlit run simple_extractor.py
3. Testen Sie mit einer echten PDF-Rechnung
4. Dokumentieren Sie die Genauigkeit

Übung 2.2: Verbesserungen implementieren

Erweitern Sie den Extraktor um: - Validierung der extrahierten Daten - Confidence Score Berechnung - Fehlerbehandlung

---

📊 Workshop 3: Validierungs-Pipeline

Aufgabe: Implementieren Sie robuste Validierung

Ziel: Bauen Sie eine mehrstufige Validierungspipeline.

# validation_workshop.py
import re
from datetime import datetime, timedelta
from decimal import Decimal
from typing import Dict, List, Tuple

class InvoiceValidator:
    def __init__(self):
        self.validation_rules = {
            'max_amount': 50000,
            'valid_vat_rates': [0.0, 0.07, 0.19],
            'max_age_days': 365
        }

    def validate_invoice(self, data: Dict) -> Tuple[bool, List[str]]:
        """Vollständige Validierung"""
        errors = []

        # Syntaktische Validierung
        errors.extend(self._validate_syntax(data))

        # Semantische Validierung  
        errors.extend(self._validate_semantics(data))

        # Geschäftsregeln
        errors.extend(self._validate_business_rules(data))

        return len(errors) == 0, errors

    def _validate_syntax(self, data: Dict) -> List[str]:
        """Prüft Datenformate"""
        errors = []

        # Rechnungsnummer Format
        invoice_number = data.get('invoice_number', '')
        if not re.match(r'^[A-Z0-9\-_]{3,20}$', invoice_number):
            errors.append("Ungültiges Rechnungsnummer-Format")

        # Beträge müssen numerisch sein
        try:
            total = float(data.get('total_amount', 0))
            if total <= 0:
                errors.append("Gesamtbetrag muss positiv sein")
        except (ValueError, TypeError):
            errors.append("Gesamtbetrag ist nicht numerisch")

        return errors

    def _validate_semantics(self, data: Dict) -> List[str]:
        """Prüft Geschäftslogik"""
        errors = []

        try:
            net = float(data.get('net_amount', 0))
            vat = float(data.get('vat_amount', 0))
            total = float(data.get('total_amount', 0))

            # Summenprüfung
            if abs((net + vat) - total) > 0.01:
                errors.append(f"Summe stimmt nicht: {net} + {vat} ≠ {total}")

            # MwSt-Satz prüfen
            if net > 0 and vat > 0:
                vat_rate = vat / net
                valid_rates = self.validation_rules['valid_vat_rates']

                if not any(abs(vat_rate - rate) < 0.01 for rate in valid_rates):
                    errors.append(f"Ungewöhnlicher MwSt-Satz: {vat_rate:.1%}")

        except (ValueError, TypeError):
            errors.append("Numerische Validierung fehlgeschlagen")

        return errors

    def _validate_business_rules(self, data: Dict) -> List[str]:
        """Prüft Geschäftsregeln"""
        errors = []

        # Maximalbetrag
        total = float(data.get('total_amount', 0))
        if total > self.validation_rules['max_amount']:
            errors.append(f"Betrag über Limit: {total} > {self.validation_rules['max_amount']}")

        # Rechnungsalter
        invoice_date = data.get('invoice_date')
        if invoice_date:
            try:
                date_obj = datetime.strptime(invoice_date, '%Y-%m-%d')
                age = (datetime.now() - date_obj).days

                if age > self.validation_rules['max_age_days']:
                    errors.append(f"Rechnung zu alt: {age} Tage")
            except ValueError:
                errors.append("Ungültiges Datumsformat")

        return errors

# Streamlit Interface für Validierung
def validation_interface():
    st.title("🔍 Rechnungsvalidierung")

    validator = InvoiceValidator()

    # Testdaten eingeben
    st.subheader("Rechnungsdaten eingeben:")

    col1, col2 = st.columns(2)

    with col1:
        invoice_number = st.text_input("Rechnungsnummer", "RE-2024-001")
        supplier = st.text_input("Lieferant", "Musterfirma GmbH")
        invoice_date = st.date_input("Rechnungsdatum").strftime('%Y-%m-%d')

    with col2:
        net_amount = st.number_input("Nettobetrag", value=100.0, step=0.01)
        vat_amount = st.number_input("MwSt-Betrag", value=19.0, step=0.01)
        total_amount = st.number_input("Gesamtbetrag", value=119.0, step=0.01)

    if st.button("Validierung durchführen"):
        test_data = {
            'invoice_number': invoice_number,
            'supplier_name': supplier,
            'invoice_date': invoice_date,
            'net_amount': net_amount,
            'vat_amount': vat_amount,
            'total_amount': total_amount
        }

        is_valid, errors = validator.validate_invoice(test_data)

        if is_valid:
            st.success("✅ Validierung erfolgreich!")
        else:
            st.error("❌ Validierungsfehler gefunden:")
            for error in errors:
                st.write(f"• {error}")

Übung 3.1: Validierungsregeln testen

1. Testen Sie verschiedene Eingabedaten
2. Provozieren Sie bewusst Validierungsfehler
3. Erweitern Sie die Regeln um weitere Geschäftslogik

---

🚀 Workshop 4: End-to-End Automatisierung

Aufgabe: Bauen Sie eine vollständige Pipeline

Ziel: Kombinieren Sie alle Komponenten zu einer funktionsfähigen Lösung.

# complete_pipeline.py
import streamlit as st
from dataclasses import dataclass
from typing import Optional, List
import json
import time

@dataclass
class ProcessingResult:
    success: bool
    extracted_data: dict
    validation_errors: List[str]
    confidence_score: float
    processing_time: float
    recommendation: str

class InvoiceAutomationPipeline:
    def __init__(self):
        self.extractor = SimpleInvoiceExtractor()
        self.validator = InvoiceValidator()
        self.processing_history = []

    def process_invoice(self, pdf_file) -> ProcessingResult:
        """Vollständige Verarbeitung einer Rechnung"""
        start_time = time.time()

        try:
            # 1. Text extrahieren
            text = self.extractor.extract_text_from_pdf(pdf_file)

            # 2. Daten extrahieren
            raw_data = self.extractor.extract_invoice_data(text)
            extracted_data = self._parse_extracted_data(raw_data)

            # 3. Validierung
            is_valid, validation_errors = self.validator.validate_invoice(extracted_data)

            # 4. Confidence Score berechnen
            confidence_score = self._calculate_confidence(extracted_data, validation_errors)

            # 5. Empfehlung generieren
            recommendation = self._generate_recommendation(
                confidence_score, 
                is_valid, 
                extracted_data.get('total_amount', 0)
            )

            processing_time = time.time() - start_time

            result = ProcessingResult(
                success=True,
                extracted_data=extracted_data,
                validation_errors=validation_errors,
                confidence_score=confidence_score,
                processing_time=processing_time,
                recommendation=recommendation
            )

            self.processing_history.append(result)
            return result

        except Exception as e:
            return ProcessingResult(
                success=False,
                extracted_data={},
                validation_errors=[f"Verarbeitungsfehler: {str(e)}"],
                confidence_score=0.0,
                processing_time=time.time() - start_time,
                recommendation="MANUAL_PROCESSING"
            )

    def _parse_extracted_data(self, raw_data: str) -> dict:
        """Parst extrahierte Daten"""
        try:
            # Versuche JSON zu parsen
            return json.loads(raw_data)
        except:
            # Fallback: Regex-basierte Extraktion
            return self._regex_fallback_extraction(raw_data)

    def _regex_fallback_extraction(self, text: str) -> dict:
        """Fallback-Extraktion mit Regex"""
        patterns = {
            'invoice_number': r'(?:Rechnung|Invoice).*?(\d+)',
            'total_amount': r'(?:Gesamt|Total).*?(\d+[,.]?\d*)',
            'supplier_name': r'(.*?GmbH|.*?AG|.*?KG)'
        }

        result = {}
        for field, pattern in patterns.items():
            match = re.search(pattern, text, re.IGNORECASE)
            result[field] = match.group(1) if match else "N/A"

        return result

    def _calculate_confidence(self, data: dict, validation_errors: List[str]) -> float:
        """Berechnet Confidence Score"""
        base_score = 1.0

        # Abzug für fehlende Pflichtfelder
        required_fields = ['invoice_number', 'total_amount', 'supplier_name']
        missing_fields = sum(1 for field in required_fields if not data.get(field))
        base_score -= (missing_fields * 0.2)

        # Abzug für Validierungsfehler
        base_score -= (len(validation_errors) * 0.1)

        return max(0.0, min(1.0, base_score))

    def _generate_recommendation(self, confidence: float, is_valid: bool, amount: float) -> str:
        """Generiert Verarbeitungsempfehlung"""
        if not is_valid:
            return "MANUAL_REVIEW_REQUIRED"
        elif confidence < 0.7:
            return "VALIDATION_RECOMMENDED"
        elif amount > 10000:
            return "APPROVAL_REQUIRED"
        else:
            return "AUTO_PROCESS"

# Streamlit Dashboard
def main_dashboard():
    st.title("🤖 KI-Rechnungsautomatisierung - Vollständige Pipeline")

    pipeline = InvoiceAutomationPipeline()

    # Sidebar für Statistiken
    st.sidebar.title("📊 Statistiken")
    if pipeline.processing_history:
        total_processed = len(pipeline.processing_history)
        successful = sum(1 for r in pipeline.processing_history if r.success)
        avg_confidence = sum(r.confidence_score for r in pipeline.processing_history) / total_processed

        st.sidebar.metric("Verarbeitete Rechnungen", total_processed)
        st.sidebar.metric("Erfolgsrate", f"{successful/total_processed:.1%}")
        st.sidebar.metric("Ø Confidence", f"{avg_confidence:.1%}")

    # Hauptbereich
    uploaded_file = st.file_uploader("📄 PDF-Rechnung hochladen", type="pdf")

    if uploaded_file:
        col1, col2 = st.columns([2, 1])

        with col1:
            st.info(f"**Datei:** {uploaded_file.name}")

        with col2:
            if st.button("🚀 Verarbeiten", type="primary"):
                with st.spinner("Verarbeite Rechnung..."):
                    result = pipeline.process_invoice(uploaded_file)
                    display_result(result)

def display_result(result: ProcessingResult):
    """Zeigt Verarbeitungsergebnis an"""

    # Status
    if result.success:
        st.success("✅ Verarbeitung erfolgreich!")
    else:
        st.error("❌ Verarbeitung fehlgeschlagen")

    # Metriken
    col1, col2, col3 = st.columns(3)

    with col1:
        st.metric("Confidence Score", f"{result.confidence_score:.1%}")

    with col2:
        st.metric("Verarbeitungszeit", f"{result.processing_time:.2f}s")

    with col3:
        validation_status = "✅" if not result.validation_errors else "❌"
        st.metric("Validierung", validation_status)

    # Extrahierte Daten
    st.subheader("📋 Extrahierte Daten")
    st.json(result.extracted_data)

    # Validierungsfehler
    if result.validation_errors:
        st.subheader("⚠️ Validierungsfehler")
        for error in result.validation_errors:
            st.write(f"• {error}")

    # Empfehlung
    recommendation_colors = {
        "AUTO_PROCESS": "🟢",
        "VALIDATION_RECOMMENDED": "🟡",
        "APPROVAL_REQUIRED": "🟠",
        "MANUAL_REVIEW_REQUIRED": "🔴"
    }

    recommendation_text = {
        "AUTO_PROCESS": "Automatische Verarbeitung möglich",
        "VALIDATION_RECOMMENDED": "Zusätzliche Validierung empfohlen",
        "APPROVAL_REQUIRED": "Manuelle Genehmigung erforderlich",
        "MANUAL_REVIEW_REQUIRED": "Vollständige manuelle Überprüfung nötig"
    }

    color = recommendation_colors.get(result.recommendation, "⚪")
    text = recommendation_text.get(result.recommendation, result.recommendation)

    st.info(f"{color} **Empfehlung:** {text}")

if __name__ == "__main__":
    main_dashboard()

Übung 4.1: Vollständige Pipeline testen

1. Implementieren Sie die komplette Pipeline
2. Testen Sie mit verschiedenen PDF-Rechnungen
3. Dokumentieren Sie die Performance-Metriken
4. Identifizieren Sie Verbesserungsmöglichkeiten

---

📈 Workshop 5: ROI-Berechnung & Business Case

Aufgabe: Erstellen Sie einen Business Case

Ziel: Quantifizieren Sie den Nutzen Ihrer Automatisierung.

# roi_calculator.py
import streamlit as st
import pandas as pd
import plotly.express as px

class ROICalculator:
    def __init__(self):
        self.default_values = {
            'monthly_volume': 500,
            'minutes_per_manual': 20,
            'minutes_per_automated': 2,
            'hourly_rate': 35,
            'automation_accuracy': 0.85,
            'implementation_cost': 50000,
            'monthly_operating_cost': 2000
        }

    def calculate_roi(self, params: dict) -> dict:
        """Berechnet ROI für Automatisierung"""

        # Zeitersparnis berechnen
        manual_time_monthly = params['monthly_volume'] * params['minutes_per_manual'] / 60

        automated_time_monthly = (
            params['monthly_volume'] * params['automation_accuracy'] * params['minutes_per_automated'] / 60 +
            params['monthly_volume'] * (1 - params['automation_accuracy']) * params['minutes_per_manual'] / 60
        )

        time_saved_monthly = manual_time_monthly - automated_time_monthly

        # Kosteneinsparung
        cost_savings_monthly = time_saved_monthly * params['hourly_rate'] - params['monthly_operating_cost']
        cost_savings_annual = cost_savings_monthly * 12

        # ROI Metriken
        net_benefit = cost_savings_annual - params['implementation_cost']
        roi_percentage = (net_benefit / params['implementation_cost']) * 100
        payback_months = params['implementation_cost'] / cost_savings_monthly if cost_savings_monthly > 0 else float('inf')

        return {
            'time_saved_monthly_hours': time_saved_monthly,
            'cost_savings_monthly': cost_savings_monthly,
            'cost_savings_annual': cost_savings_annual,
            'net_benefit': net_benefit,
            'roi_percentage': roi_percentage,
            'payback_period_months': payback_months,
            'break_even_month': payback_months
        }

def roi_interface():
    st.title("💰 ROI-Rechner für KI-Automatisierung")

    calculator = ROICalculator()

    st.subheader("📊 Parameter eingeben")

    col1, col2 = st.columns(2)

    with col1:
        st.markdown("**Volumen & Zeit**")
        monthly_volume = st.number_input("Rechnungen pro Monat", value=500, step=50)
        minutes_per_manual = st.number_input("Minuten pro Rechnung (manuell)", value=20, step=1)
        minutes_per_automated = st.number_input("Minuten pro Rechnung (automatisiert)", value=2, step=1)

    with col2:
        st.markdown("**Kosten & Genauigkeit**")
        hourly_rate = st.number_input("Stundensatz (€)", value=35, step=5)
        automation_accuracy = st.slider("Automatisierungsrate", 0.0, 1.0, 0.85, 0.05)

    st.markdown("**Investition**")
    col3, col4 = st.columns(2)

    with col3:
        implementation_cost = st.number_input("Implementierungskosten (€)", value=50000, step=5000)

    with col4:
        monthly_operating_cost = st.number_input("Monatliche Betriebskosten (€)", value=2000, step=100)

    # Berechnung
    params = {
        'monthly_volume': monthly_volume,
        'minutes_per_manual': minutes_per_manual,
        'minutes_per_automated': minutes_per_automated,
        'hourly_rate': hourly_rate,
        'automation_accuracy': automation_accuracy,
        'implementation_cost': implementation_cost,
        'monthly_operating_cost': monthly_operating_cost
    }

    results = calculator.calculate_roi(params)

    # Ergebnisse anzeigen
    st.subheader("🎯 ROI-Ergebnisse")

    col1, col2, col3, col4 = st.columns(4)

    with col1:
        st.metric("Gesparte Zeit/Monat", f"{results['time_saved_monthly_hours']:.0f}h")

    with col2:
        st.metric("Einsparung/Monat", f"{results['cost_savings_monthly']:,.0f}€")

    with col3:
        st.metric("ROI", f"{results['roi_percentage']:.1f}%")

    with col4:
        payback_text = f"{results['payback_period_months']:.1f} Monate" if results['payback_period_months'] != float('inf') else "Nie"
        st.metric("Amortisation", payback_text)

    # Jahresübersicht
    if results['cost_savings_monthly'] > 0:
        months = list(range(1, 37))  # 3 Jahre
        cumulative_savings = [results['cost_savings_monthly'] * m - implementation_cost for m in months]

        df = pd.DataFrame({
            'Monat': months,
            'Kumulierte Einsparungen': cumulative_savings
        })

        fig = px.line(df, x='Monat', y='Kumulierte Einsparungen', 
                     title="Kumulierte Einsparungen über Zeit")
        fig.add_hline(y=0, line_dash="dash", line_color="red", annotation_text="Break-Even")

        st.plotly_chart(fig, use_container_width=True)

if __name__ == "__main__":
    roi_interface()

Übung 5.1: Business Case erstellen

1. Berechnen Sie den ROI für Ihren identifizierten Use Case
2. Variieren Sie die Parameter und analysieren Sie Sensitivitäten
3. Erstellen Sie eine Präsentation für das Management

---

🎓 Abschlussprojekt: Produktionsreife Lösung

Aufgabe: Implementieren Sie eine skalierbare Lösung

Ziel: Bauen Sie eine produktionsreife KI-Automatisierungslösung.

Anforderungen:

1. Vollständige Pipeline mit allen Komponenten
2. Robuste Validierung und Fehlerbehandlung
3. Monitoring & Logging für Produktionsbetrieb
4. Benutzerfreundliches Interface für Endanwender
5. Dokumentation für Betrieb und Wartung

Bewertungskriterien:

• Funktionalität (30%): Vollständigkeit und Korrektheit
• Qualität (25%): Code-Qualität und Architektur
• Benutzerfreundlichkeit (20%): Interface und User Experience
• Dokumentation (15%): Vollständigkeit und Klarheit
• Innovation (10%): Kreative Lösungsansätze

Abgabe:

• Vollständiger Source Code
• Deployment-Anleitung
• Benutzerhandbuch
• Technische Dokumentation
• Business Case Präsentation

---

📚 Zusätzliche Ressourcen

Nützliche Tools & Libraries

• LangChain: Framework für LLM-Anwendungen
• Streamlit: Rapid Prototyping für Web-Apps
• FastAPI: Production-ready APIs
• PyPDF2/pdfplumber: PDF-Verarbeitung
• Pydantic: Datenvalidierung
• Plotly: Interaktive Visualisierungen

Weiterführende Lektüre

• LangChain Documentation
• Streamlit Documentation
• OpenAI API Reference
• FastAPI Documentation

Community & Support

• LangChain Discord Community
• Streamlit Community Forum
• OpenAI Developer Community
• Local Python/AI Meetups

Glückwunsch!

Mit Abschluss dieser Übungen haben Sie die Grundlagen für erfolgreiche KI-Automatisierung im Finanzbereich gelegt. Nutzen Sie diese Erfahrungen als Sprungbrett für größere Automatisierungsprojekte in Ihrem Unternehmen!