Technische Implementierung: Von der Theorie zur Praxis¶
Überblick¶
Diese Seite bietet detaillierte technische Anleitungen zur Umsetzung von KI-Automatisierungslösungen im Finanzbereich. Von Low-Code-Ansätzen bis hin zu vollständigen Python-Implementierungen finden Sie hier praktische Beispiele und bewährte Praktiken.
🛠️ Tool-Stack & Architektur¶
Empfohlene Technologie-Kombination¶
Entwicklungsumgebung Setup¶
# Virtuelle Umgebung erstellen
python -m venv ai_finance_env
source ai_finance_env/bin/activate # Linux/Mac
# ai_finance_env\Scripts\activate # Windows
# Kern-Dependencies installieren
pip install langchain langchain-openai langgraph
pip install streamlit fastapi uvicorn
pip install pandas numpy scikit-learn
pip install python-multipart python-dotenv
pip install celery redis
📋 Vollständige Implementierung: Rechnungsverarbeitung¶
1. Projekt-Struktur¶
invoice_automation/
├── src/
│ ├── __init__.py
│ ├── processors/
│ │ ├── __init__.py
│ │ ├── document_processor.py
│ │ ├── validation_engine.py
│ │ └── erp_connector.py
│ ├── models/
│ │ ├── __init__.py
│ │ ├── invoice_model.py
│ │ └── validation_rules.py
│ ├── utils/
│ │ ├── __init__.py
│ │ ├── file_handler.py
│ │ └── logging_config.py
│ └── api/
│ ├── __init__.py
│ └── main.py
├── tests/
├── config/
│ └── settings.py
├── requirements.txt
└── docker-compose.yml
2. Datenmodell Definition¶
# src/models/invoice_model.py
from pydantic import BaseModel, Field, validator
from typing import Optional, List
from datetime import datetime
from decimal import Decimal
class InvoiceData(BaseModel):
"""Strukturiertes Datenmodell für Rechnungsinformationen"""
invoice_number: str = Field(..., description="Rechnungsnummer")
supplier_name: str = Field(..., description="Lieferantenname")
supplier_address: Optional[str] = Field(None, description="Lieferantenadresse")
invoice_date: datetime = Field(..., description="Rechnungsdatum")
due_date: Optional[datetime] = Field(None, description="Fälligkeitsdatum")
net_amount: Decimal = Field(..., description="Nettobetrag")
vat_amount: Decimal = Field(..., description="Mehrwertsteuerbetrag")
total_amount: Decimal = Field(..., description="Gesamtbetrag")
vat_rate: Optional[float] = Field(None, description="Mehrwertsteuersatz")
cost_center: Optional[str] = Field(None, description="Kostenstelle")
account_code: Optional[str] = Field(None, description="Kontierungscode")
line_items: List['LineItem'] = Field(default_factory=list, description="Rechnungspositionen")
confidence_score: float = Field(0.0, description="Vertrauensscore der Extraktion")
extraction_notes: List[str] = Field(default_factory=list, description="Extraktionshinweise")
@validator('total_amount')
def validate_total_amount(cls, v, values):
"""Validiert, dass Gesamtbetrag = Nettobetrag + MwSt"""
if 'net_amount' in values and 'vat_amount' in values:
expected_total = values['net_amount'] + values['vat_amount']
if abs(v - expected_total) > Decimal('0.01'):
raise ValueError('Gesamtbetrag stimmt nicht mit Netto + MwSt überein')
return v
class LineItem(BaseModel):
"""Einzelne Rechnungsposition"""
description: str
quantity: Optional[float] = None
unit_price: Optional[Decimal] = None
total_price: Decimal
account_code: Optional[str] = None
# Pydantic Model Update für Forward References
InvoiceData.model_rebuild()
3. Dokumentenverarbeitung mit LangChain¶
# src/processors/document_processor.py
import os
from typing import Dict, Any, Optional
from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain
from langchain.output_parsers import PydanticOutputParser
from src.models.invoice_model import InvoiceData
import logging
class InvoiceDocumentProcessor:
"""Intelligente Rechnungsverarbeitung mit LangChain"""
def __init__(self, openai_api_key: str = None):
self.logger = logging.getLogger(__name__)
# LLM Setup
api_key = openai_api_key or os.getenv("OPENAI_API_KEY")
self.llm = OpenAI(
temperature=0,
openai_api_key=api_key,
model_name="gpt-3.5-turbo-instruct"
)
# Output Parser für strukturierte Daten
self.output_parser = PydanticOutputParser(pydantic_object=InvoiceData)
# Prompt Template für deutsche Rechnungen
self.extraction_prompt = PromptTemplate(
input_variables=["document_text"],
template="""
Du bist ein Experte für die Extraktion von Rechnungsdaten aus deutschen Geschäftsdokumenten.
Analysiere den folgenden Rechnungstext und extrahiere alle relevanten Informationen:
{document_text}
Beachte dabei:
- Deutsche Datumsformate (DD.MM.YYYY oder DD/MM/YYYY)
- Deutsche Zahlenformate (Komma als Dezimaltrennzeichen)
- Typische deutsche Rechnungsbegriffe
- MwSt = Mehrwertsteuer, USt = Umsatzsteuer
Wenn Informationen nicht eindeutig erkennbar sind, setze den confidence_score entsprechend niedriger
und füge Hinweise in extraction_notes hinzu.
{format_instructions}
""",
partial_variables={"format_instructions": self.output_parser.get_format_instructions()}
)
# LLM Chain Setup
self.extraction_chain = LLMChain(
llm=self.llm,
prompt=self.extraction_prompt,
output_parser=self.output_parser
)
def process_pdf(self, pdf_path: str) -> InvoiceData:
"""
Verarbeitet eine PDF-Rechnung und extrahiert strukturierte Daten
Args:
pdf_path: Pfad zur PDF-Datei
Returns:
InvoiceData: Strukturierte Rechnungsdaten
"""
try:
# PDF laden und Text extrahieren
self.logger.info(f"Verarbeite PDF: {pdf_path}")
loader = PyPDFLoader(pdf_path)
documents = loader.load()
# Text zusammenfügen
full_text = "\n".join([doc.page_content for doc in documents])
# Bei sehr langen Dokumenten: Text aufteilen
if len(full_text) > 4000:
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=4000,
chunk_overlap=200
)
text_chunks = text_splitter.split_text(full_text)
# Ersten relevanten Chunk verwenden (meist Seite 1)
full_text = text_chunks[0]
# LLM-basierte Extraktion
self.logger.info("Starte LLM-basierte Datenextraktion")
extracted_data = self.extraction_chain.run(document_text=full_text)
# Confidence Score basierend auf Vollständigkeit berechnen
confidence = self._calculate_confidence(extracted_data)
extracted_data.confidence_score = confidence
self.logger.info(f"Extraktion abgeschlossen. Confidence: {confidence:.2f}")
return extracted_data
except Exception as e:
self.logger.error(f"Fehler bei PDF-Verarbeitung: {str(e)}")
# Fallback: Leeres InvoiceData-Objekt mit Fehlermeldung
return InvoiceData(
invoice_number="EXTRACTION_FAILED",
supplier_name="UNKNOWN",
invoice_date=datetime.now(),
net_amount=Decimal('0'),
vat_amount=Decimal('0'),
total_amount=Decimal('0'),
confidence_score=0.0,
extraction_notes=[f"Extraktionsfehler: {str(e)}"]
)
def _calculate_confidence(self, data: InvoiceData) -> float:
"""
Berechnet einen Confidence Score basierend auf Vollständigkeit der Daten
"""
required_fields = ['invoice_number', 'supplier_name', 'total_amount']
optional_fields = ['invoice_date', 'net_amount', 'vat_amount', 'cost_center']
# Basis-Score für Pflichtfelder
required_score = 0
for field in required_fields:
value = getattr(data, field)
if value and str(value) not in ['', '0', 'UNKNOWN']:
required_score += 1
# Bonus für optionale Felder
optional_score = 0
for field in optional_fields:
value = getattr(data, field)
if value and str(value) not in ['', '0', 'None']:
optional_score += 1
# Gesamtscore berechnen
base_confidence = (required_score / len(required_fields)) * 0.7
bonus_confidence = (optional_score / len(optional_fields)) * 0.3
return min(base_confidence + bonus_confidence, 1.0)
4. Validierungs-Engine¶
# src/processors/validation_engine.py
from typing import List, Dict, Any, Tuple
from decimal import Decimal
from datetime import datetime, timedelta
from src.models.invoice_model import InvoiceData
import re
class ValidationEngine:
"""Mehrstufige Validierung für Rechnungsdaten"""
def __init__(self):
self.validation_rules = self._load_validation_rules()
def validate_invoice(self, invoice: InvoiceData) -> Tuple[bool, List[str]]:
"""
Führt vollständige Validierung durch
Returns:
Tuple[bool, List[str]]: (is_valid, error_messages)
"""
errors = []
# Syntaktische Validierung
syntax_errors = self._validate_syntax(invoice)
errors.extend(syntax_errors)
# Semantische Validierung
semantic_errors = self._validate_semantics(invoice)
errors.extend(semantic_errors)
# Kontextuelle Validierung
context_errors = self._validate_context(invoice)
errors.extend(context_errors)
is_valid = len(errors) == 0
return is_valid, errors
def _validate_syntax(self, invoice: InvoiceData) -> List[str]:
"""Syntaktische Validierung: Datentypen und Formate"""
errors = []
# Rechnungsnummer Format
if not re.match(r'^[A-Z0-9\-_]+$', invoice.invoice_number):
errors.append("Rechnungsnummer enthält ungültige Zeichen")
# Beträge müssen positiv sein
if invoice.total_amount <= 0:
errors.append("Gesamtbetrag muss positiv sein")
if invoice.net_amount < 0:
errors.append("Nettobetrag darf nicht negativ sein")
# Datum Plausibilität
if invoice.invoice_date > datetime.now():
errors.append("Rechnungsdatum liegt in der Zukunft")
if invoice.invoice_date < datetime.now() - timedelta(days=365*2):
errors.append("Rechnungsdatum ist älter als 2 Jahre")
return errors
def _validate_semantics(self, invoice: InvoiceData) -> List[str]:
"""Semantische Validierung: Geschäftslogik"""
errors = []
# MwSt-Berechnung prüfen
if invoice.vat_amount > 0 and invoice.net_amount > 0:
calculated_vat_rate = float(invoice.vat_amount / invoice.net_amount)
# Deutsche Standard-MwSt-Sätze: 19%, 7%, 0%
valid_vat_rates = [0.0, 0.07, 0.19]
tolerance = 0.01
if not any(abs(calculated_vat_rate - rate) < tolerance for rate in valid_vat_rates):
errors.append(f"Ungewöhnlicher MwSt-Satz: {calculated_vat_rate:.1%}")
# Fälligkeitsdatum nach Rechnungsdatum
if invoice.due_date and invoice.due_date < invoice.invoice_date:
errors.append("Fälligkeitsdatum liegt vor Rechnungsdatum")
# Lieferantenname Plausibilität
if len(invoice.supplier_name) < 3:
errors.append("Lieferantenname zu kurz")
return errors
def _validate_context(self, invoice: InvoiceData) -> List[str]:
"""Kontextuelle Validierung: Historische Vergleiche"""
errors = []
# Hier könnten historische Daten verglichen werden
# Für Demo-Zwecke: Einfache Plausibilitätsprüfungen
# Sehr hohe Beträge markieren
if invoice.total_amount > Decimal('50000'):
errors.append("Ungewöhnlich hoher Rechnungsbetrag (>50.000€)")
# Confidence Score berücksichtigen
if invoice.confidence_score < 0.7:
errors.append(f"Niedrige Extraktions-Confidence: {invoice.confidence_score:.2f}")
return errors
def _load_validation_rules(self) -> Dict[str, Any]:
"""Lädt konfigurierbare Validierungsregeln"""
return {
"max_amount": Decimal('100000'),
"min_confidence": 0.8,
"allowed_vat_rates": [0.0, 0.07, 0.19],
"max_age_days": 730
}
5. FastAPI Web Service¶
# src/api/main.py
from fastapi import FastAPI, File, UploadFile, HTTPException, BackgroundTasks
from fastapi.responses import JSONResponse
from typing import List
import tempfile
import os
from src.processors.document_processor import InvoiceDocumentProcessor
from src.processors.validation_engine import ValidationEngine
from src.models.invoice_model import InvoiceData
import logging
# Logging Setup
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# FastAPI App
app = FastAPI(
title="KI-Rechnungsverarbeitung API",
description="Automatisierte Verarbeitung von Eingangsrechnungen mit KI",
version="1.0.0"
)
# Globale Instanzen
processor = InvoiceDocumentProcessor()
validator = ValidationEngine()
@app.post("/process-invoice/", response_model=dict)
async def process_invoice(
background_tasks: BackgroundTasks,
file: UploadFile = File(..., description="PDF-Rechnung zum Verarbeiten")
):
"""
Verarbeitet eine hochgeladene PDF-Rechnung
"""
if not file.filename.lower().endswith('.pdf'):
raise HTTPException(status_code=400, detail="Nur PDF-Dateien sind erlaubt")
try:
# Temporäre Datei erstellen
with tempfile.NamedTemporaryFile(delete=False, suffix='.pdf') as tmp_file:
content = await file.read()
tmp_file.write(content)
tmp_file_path = tmp_file.name
# Verarbeitung
logger.info(f"Verarbeite Datei: {file.filename}")
extracted_data = processor.process_pdf(tmp_file_path)
# Validierung
is_valid, validation_errors = validator.validate_invoice(extracted_data)
# Aufräumen
background_tasks.add_task(cleanup_temp_file, tmp_file_path)
# Response zusammenstellen
response = {
"filename": file.filename,
"extraction_successful": True,
"data": extracted_data.dict(),
"validation": {
"is_valid": is_valid,
"errors": validation_errors
},
"recommendation": get_processing_recommendation(extracted_data, is_valid)
}
return response
except Exception as e:
logger.error(f"Fehler bei Verarbeitung: {str(e)}")
# Aufräumen auch bei Fehlern
if 'tmp_file_path' in locals():
background_tasks.add_task(cleanup_temp_file, tmp_file_path)
raise HTTPException(status_code=500, detail=f"Verarbeitungsfehler: {str(e)}")
@app.post("/batch-process/", response_model=dict)
async def batch_process_invoices(
background_tasks: BackgroundTasks,
files: List[UploadFile] = File(..., description="Mehrere PDF-Rechnungen")
):
"""
Verarbeitet mehrere Rechnungen in einem Batch
"""
results = []
for file in files:
try:
# Einzelverarbeitung für jede Datei
result = await process_invoice(background_tasks, file)
results.append(result)
except Exception as e:
results.append({
"filename": file.filename,
"extraction_successful": False,
"error": str(e)
})
return {
"batch_size": len(files),
"successful": len([r for r in results if r.get("extraction_successful", False)]),
"failed": len([r for r in results if not r.get("extraction_successful", False)]),
"results": results
}
@app.get("/health")
async def health_check():
"""Gesundheitscheck für die API"""
return {"status": "healthy", "service": "KI-Rechnungsverarbeitung"}
def cleanup_temp_file(file_path: str):
"""Räumt temporäre Dateien auf"""
try:
os.unlink(file_path)
logger.info(f"Temporäre Datei gelöscht: {file_path}")
except Exception as e:
logger.warning(f"Konnte temporäre Datei nicht löschen: {e}")
def get_processing_recommendation(data: InvoiceData, is_valid: bool) -> str:
"""Gibt Empfehlung für weitere Verarbeitung"""
if not is_valid:
return "MANUAL_REVIEW_REQUIRED"
elif data.confidence_score < 0.8:
return "VALIDATION_RECOMMENDED"
elif data.total_amount > 10000:
return "APPROVAL_REQUIRED"
else:
return "AUTO_PROCESS"
# Anwendung starten mit: uvicorn src.api.main:app --reload
6. Streamlit Dashboard¶
# dashboard.py
import streamlit as st
import requests
import pandas as pd
from datetime import datetime
import plotly.express as px
from typing import Dict, Any
import json
# Konfiguration
API_BASE_URL = "http://localhost:8000"
st.set_page_config(
page_title="KI-Rechnungsverarbeitung Dashboard",
page_icon="📊",
layout="wide"
)
def main():
st.title("🤖 KI-Rechnungsverarbeitung Dashboard")
st.markdown("---")
# Sidebar für Navigation
st.sidebar.title("Navigation")
page = st.sidebar.selectbox(
"Seite auswählen",
["Rechnungen verarbeiten", "Batch-Verarbeitung", "Analytics", "Einstellungen"]
)
if page == "Rechnungen verarbeiten":
single_invoice_page()
elif page == "Batch-Verarbeitung":
batch_processing_page()
elif page == "Analytics":
analytics_page()
elif page == "Einstellungen":
settings_page()
def single_invoice_page():
st.header("📄 Einzelne Rechnung verarbeiten")
# File Upload
uploaded_file = st.file_uploader(
"PDF-Rechnung hochladen",
type=['pdf'],
help="Laden Sie eine PDF-Rechnung zur automatischen Verarbeitung hoch"
)
if uploaded_file is not None:
col1, col2 = st.columns(2)
with col1:
if st.button("Rechnung verarbeiten", type="primary"):
with st.spinner("Verarbeite Rechnung..."):
result = process_single_invoice(uploaded_file)
display_processing_result(result)
with col2:
st.info(f"**Datei:** {uploaded_file.name}\n**Größe:** {uploaded_file.size:,} Bytes")
def batch_processing_page():
st.header("📦 Batch-Verarbeitung")
uploaded_files = st.file_uploader(
"Mehrere PDF-Rechnungen hochladen",
type=['pdf'],
accept_multiple_files=True,
help="Laden Sie mehrere PDF-Rechnungen für die Batch-Verarbeitung hoch"
)
if uploaded_files:
st.write(f"**{len(uploaded_files)} Dateien** ausgewählt")
if st.button("Batch verarbeiten", type="primary"):
with st.spinner("Verarbeite alle Rechnungen..."):
results = process_batch_invoices(uploaded_files)
display_batch_results(results)
def analytics_page():
st.header("📈 Analytics & Übersicht")
# Demo-Daten für Analytics
demo_data = generate_demo_analytics()
col1, col2, col3, col4 = st.columns(4)
with col1:
st.metric("Verarbeitete Rechnungen", "1,234", "+123 (heute)")
with col2:
st.metric("Durchschnittliche Genauigkeit", "94.2%", "+2.1%")
with col3:
st.metric("Automatisierungsrate", "87%", "+5%")
with col4:
st.metric("Eingesparte Zeit", "156h", "+12h")
# Charts
col1, col2 = st.columns(2)
with col1:
st.subheader("Verarbeitungsvolumen")
fig = px.line(demo_data['volume'], x='date', y='count', title="Tägliche Rechnungsverarbeitung")
st.plotly_chart(fig, use_container_width=True)
with col2:
st.subheader("Genauigkeitsverteilung")
fig = px.histogram(demo_data['accuracy'], x='confidence_score', title="Confidence Score Verteilung")
st.plotly_chart(fig, use_container_width=True)
def settings_page():
st.header("⚙️ Einstellungen")
st.subheader("API-Konfiguration")
api_url = st.text_input("API Base URL", value=API_BASE_URL)
st.subheader("Validierungsregeln")
col1, col2 = st.columns(2)
with col1:
min_confidence = st.slider("Minimale Confidence", 0.0, 1.0, 0.8, 0.05)
max_amount = st.number_input("Maximalbetrag (€)", value=50000, step=1000)
with col2:
auto_approve_threshold = st.slider("Auto-Genehmigung bis", 0, 10000, 1000, 100)
manual_review_threshold = st.slider("Manuelle Prüfung ab", 0.0, 1.0, 0.7, 0.05)
if st.button("Einstellungen speichern"):
st.success("Einstellungen gespeichert!")
def process_single_invoice(uploaded_file) -> Dict[str, Any]:
"""Sendet einzelne Rechnung an API"""
try:
files = {"file": (uploaded_file.name, uploaded_file.getvalue(), "application/pdf")}
response = requests.post(f"{API_BASE_URL}/process-invoice/", files=files)
if response.status_code == 200:
return response.json()
else:
return {"error": f"API Fehler: {response.status_code}"}
except requests.exceptions.RequestException as e:
return {"error": f"Verbindungsfehler: {str(e)}"}
def process_batch_invoices(uploaded_files) -> Dict[str, Any]:
"""Sendet mehrere Rechnungen an API"""
try:
files = [("files", (f.name, f.getvalue(), "application/pdf")) for f in uploaded_files]
response = requests.post(f"{API_BASE_URL}/batch-process/", files=files)
if response.status_code == 200:
return response.json()
else:
return {"error": f"API Fehler: {response.status_code}"}
except requests.exceptions.RequestException as e:
return {"error": f"Verbindungsfehler: {str(e)}"}
def display_processing_result(result: Dict[str, Any]):
"""Zeigt Ergebnis der Einzelverarbeitung an"""
if "error" in result:
st.error(f"Fehler: {result['error']}")
return
data = result.get("data", {})
validation = result.get("validation", {})
# Erfolg/Fehler Status
if result.get("extraction_successful", False):
st.success("✅ Rechnung erfolgreich verarbeitet!")
else:
st.error("❌ Verarbeitung fehlgeschlagen")
# Extrahierte Daten anzeigen
col1, col2 = st.columns(2)
with col1:
st.subheader("Extrahierte Daten")
st.write(f"**Rechnungsnummer:** {data.get('invoice_number', 'N/A')}")
st.write(f"**Lieferant:** {data.get('supplier_name', 'N/A')}")
st.write(f"**Datum:** {data.get('invoice_date', 'N/A')}")
st.write(f"**Gesamtbetrag:** {data.get('total_amount', 'N/A')}€")
st.write(f"**Confidence Score:** {data.get('confidence_score', 0):.2%}")
with col2:
st.subheader("Validierung")
if validation.get("is_valid", False):
st.success("✅ Validierung erfolgreich")
else:
st.warning("⚠️ Validierungsfehler gefunden")
for error in validation.get("errors", []):
st.write(f"• {error}")
# Empfehlung
recommendation = result.get("recommendation", "UNKNOWN")
recommendation_text = {
"AUTO_PROCESS": "🟢 Automatische Verarbeitung möglich",
"VALIDATION_RECOMMENDED": "🟡 Validierung empfohlen",
"MANUAL_REVIEW_REQUIRED": "🔴 Manuelle Prüfung erforderlich",
"APPROVAL_REQUIRED": "🟠 Genehmigung erforderlich"
}
st.info(f"**Empfehlung:** {recommendation_text.get(recommendation, recommendation)}")
def display_batch_results(results: Dict[str, Any]):
"""Zeigt Ergebnisse der Batch-Verarbeitung an"""
if "error" in results:
st.error(f"Fehler: {results['error']}")
return
# Zusammenfassung
col1, col2, col3 = st.columns(3)
with col1:
st.metric("Gesamt", results.get("batch_size", 0))
with col2:
st.metric("Erfolgreich", results.get("successful", 0))
with col3:
st.metric("Fehlgeschlagen", results.get("failed", 0))
# Detailergebnisse
st.subheader("Detailergebnisse")
batch_results = results.get("results", [])
df_data = []
for result in batch_results:
data = result.get("data", {})
validation = result.get("validation", {})
df_data.append({
"Datei": result.get("filename", "N/A"),
"Status": "✅" if result.get("extraction_successful", False) else "❌",
"Lieferant": data.get("supplier_name", "N/A"),
"Betrag": f"{data.get('total_amount', 0)}€",
"Confidence": f"{data.get('confidence_score', 0):.1%}",
"Validierung": "✅" if validation.get("is_valid", False) else "❌",
"Empfehlung": result.get("recommendation", "N/A")
})
if df_data:
df = pd.DataFrame(df_data)
st.dataframe(df, use_container_width=True)
def generate_demo_analytics():
"""Generiert Demo-Daten für Analytics"""
import numpy as np
# Volume Data
dates = pd.date_range(start='2024-01-01', end='2024-01-31', freq='D')
volume_data = pd.DataFrame({
'date': dates,
'count': np.random.poisson(45, len(dates))
})
# Accuracy Data
accuracy_data = pd.DataFrame({
'confidence_score': np.random.beta(8, 2, 1000)
})
return {
'volume': volume_data,
'accuracy': accuracy_data
}
if __name__ == "__main__":
main()
# Anwendung starten mit: streamlit run dashboard.py
🚀 Deployment & Produktion¶
Docker Setup¶
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
# System Dependencies
RUN apt-get update && apt-get install -y \
gcc \
poppler-utils \
&& rm -rf /var/lib/apt/lists/*
# Python Dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Application Code
COPY src/ ./src/
COPY config/ ./config/
# API starten
CMD ["uvicorn", "src.api.main:app", "--host", "0.0.0.0", "--port", "8000"]
# docker-compose.yml
version: '3.8'
services:
api:
build: .
ports:
- "8000:8000"
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
- DATABASE_URL=${DATABASE_URL}
volumes:
- ./uploads:/app/uploads
depends_on:
- redis
- postgres
dashboard:
build:
context: .
dockerfile: Dockerfile.streamlit
ports:
- "8501:8501"
depends_on:
- api
redis:
image: redis:7-alpine
ports:
- "6379:6379"
postgres:
image: postgres:15
environment:
POSTGRES_DB: invoice_db
POSTGRES_USER: invoice_user
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
volumes:
- postgres_data:/var/lib/postgresql/data
ports:
- "5432:5432"
celery:
build: .
command: celery -A src.tasks worker --loglevel=info
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
- CELERY_BROKER_URL=redis://redis:6379
depends_on:
- redis
- postgres
volumes:
postgres_data:
Monitoring & Logging¶
# src/utils/monitoring.py
import logging
import time
from functools import wraps
from typing import Callable, Any
import psutil
import json
from datetime import datetime
class PerformanceMonitor:
"""Performance und Resource Monitoring"""
def __init__(self):
self.logger = logging.getLogger(__name__)
self.metrics = []
def monitor_performance(self, func: Callable) -> Callable:
"""Decorator für Performance-Monitoring"""
@wraps(func)
def wrapper(*args, **kwargs) -> Any:
start_time = time.time()
start_memory = psutil.Process().memory_info().rss / 1024 / 1024 # MB
try:
result = func(*args, **kwargs)
success = True
error = None
except Exception as e:
result = None
success = False
error = str(e)
raise
finally:
end_time = time.time()
end_memory = psutil.Process().memory_info().rss / 1024 / 1024 # MB
metrics = {
"timestamp": datetime.now().isoformat(),
"function": func.__name__,
"duration_seconds": end_time - start_time,
"memory_usage_mb": end_memory - start_memory,
"success": success,
"error": error
}
self.metrics.append(metrics)
self.logger.info(f"Performance: {json.dumps(metrics)}")
return result
return wrapper
def get_metrics_summary(self) -> dict:
"""Gibt Zusammenfassung der Performance-Metriken zurück"""
if not self.metrics:
return {}
durations = [m["duration_seconds"] for m in self.metrics]
memory_usages = [m["memory_usage_mb"] for m in self.metrics]
success_rate = sum(1 for m in self.metrics if m["success"]) / len(self.metrics)
return {
"total_calls": len(self.metrics),
"avg_duration": sum(durations) / len(durations),
"max_duration": max(durations),
"avg_memory_usage": sum(memory_usages) / len(memory_usages),
"success_rate": success_rate
}
📚 Weiterführende Ressourcen¶
Nützliche Libraries¶
# requirements.txt
langchain==0.1.0
langchain-openai==0.0.5
langgraph==0.0.20
streamlit==1.29.0
fastapi==0.104.1
uvicorn==0.24.0
pydantic==2.5.0
pandas==2.1.4
numpy==1.24.3
scikit-learn==1.3.2
plotly==5.17.0
python-multipart==0.0.6
python-dotenv==1.0.0
celery==5.3.4
redis==5.0.1
psycopg2-binary==2.9.9
PyPDF2==3.0.1
pytesseract==0.3.10
Konfigurationsdateien¶
# config/settings.py
import os
from typing import Optional
from pydantic import BaseSettings
class Settings(BaseSettings):
"""Anwendungseinstellungen"""
# API Keys
openai_api_key: str
# Database
database_url: str = "postgresql://user:pass@localhost/invoice_db"
# Redis
redis_url: str = "redis://localhost:6379"
# File Storage
upload_dir: str = "/tmp/uploads"
max_file_size: int = 50 * 1024 * 1024 # 50MB
# AI Settings
default_llm_temperature: float = 0.0
min_confidence_threshold: float = 0.8
# Validation
max_invoice_amount: float = 100000.0
max_invoice_age_days: int = 730
class Config:
env_file = ".env"
settings = Settings()
✅ Nächste Schritte¶
- Setup: Entwicklungsumgebung einrichten
- Testing: Mit Beispiel-PDFs testen
- Anpassung: Prompts für Ihre spezifischen Rechnungsformate optimieren
- Integration: Mit Ihren bestehenden ERP-Systemen verbinden
- Skalierung: Produktionsdeployment mit Docker
Produktionstipps
- Verwenden Sie Umgebungsvariablen für API-Keys
- Implementieren Sie Rate Limiting für die API
- Setzen Sie Monitoring und Alerting auf
- Führen Sie regelmäßige Backups durch
- Testen Sie mit realen Daten vor dem Produktionsstart