DSPy: Systematische LLM-Programmierung für Unternehmen

# Klassisches Prompt Engineering
prompt = "Du bist ein Experte für... Analysiere folgendes..."
response = llm(prompt + user_input)

# DSPy Approach
class DocumentAnalyzer(dspy.Module):
    def forward(self, document):
        summary = self.summarize(document)
        analysis = self.analyze(summary)
        return self.conclusion(analysis)

# DSPy optimiert automatisch alle Prompts

1. Prompt schreiben
2. Testen mit Beispielen
3. Manuell anpassen
4. Wiederholen bis "gut genug"
5. Bei neuen Daten: Zurück zu Schritt 1

1. Aufgabe definieren (Signatures)
2. Module zusammenbauen
3. Beispieldaten bereitstellen
4. DSPy optimiert automatisch
5. Bei neuen Daten: Automatische Anpassung

# Einfache Signature
class Summarize(dspy.Signature):
    """Summarize the document into key points"""
    document = dspy.InputField()
    summary = dspy.OutputField(desc="3-5 key points")

# Komplexe Signature
class TechnicalAnalysis(dspy.Signature):
    """Analyze technical documentation for compliance"""
    document = dspy.InputField()
    regulations = dspy.InputField(desc="Relevant regulations")
    compliance_score = dspy.OutputField(desc="Score 1-10")
    issues = dspy.OutputField(desc="List of compliance issues")
    recommendations = dspy.OutputField(desc="Specific action items")

class ComplianceChecker(dspy.Module):
    def __init__(self):
        self.extract_requirements = dspy.ChainOfThought(ExtractRequirements)
        self.check_compliance = dspy.ChainOfThought(CheckCompliance)
        self.generate_report = dspy.ChainOfThought(GenerateReport)

    def forward(self, document, regulations):
        requirements = self.extract_requirements(
            document=document,
            regulations=regulations
        )
        compliance = self.check_compliance(
            requirements=requirements.requirements,
            document=document
        )
        report = self.generate_report(
            compliance_result=compliance.result,
            issues=compliance.issues
        )
        return report

# BootstrapFewShot: Findet beste Beispiele
optimizer = dspy.BootstrapFewShot(
    metric=compliance_accuracy,
    max_bootstrapped_demos=8,
    max_labeled_demos=16
)

# COPRO: Optimiert Prompts durch Beschreibungsgeneration
optimizer = dspy.COPRO(
    metric=compliance_accuracy,
    breadth=10,
    depth=3
)

# Kompilierung = Optimierung
optimized_checker = optimizer.compile(
    ComplianceChecker(),
    trainset=training_examples
)

prompt = """
Du bist ein Rechtsexperte. Analysiere den folgenden Vertrag
auf DSGVO-Compliance und gib eine Bewertung von 1-10...
[2000 Zeichen Prompt-Text]
"""

class ContractAnalyzer(dspy.Module):
    def __init__(self):
        self.extract_clauses = dspy.ChainOfThought(ExtractClauses)
        self.assess_gdpr = dspy.ChainOfThought(AssessGDPR)
        self.risk_analysis = dspy.ChainOfThought(RiskAnalysis)
        self.generate_recommendations = dspy.Predict(GenerateRecommendations)

    def forward(self, contract_text, contract_type):
        clauses = self.extract_clauses(
            contract=contract_text,
            type=contract_type
        )

        gdpr_assessment = self.assess_gdpr(
            clauses=clauses.clauses,
            requirements=GDPR_REQUIREMENTS
        )

        risks = self.risk_analysis(
            assessment=gdpr_assessment.assessment,
            clauses=clauses.clauses
        )

        recommendations = self.generate_recommendations(
            risks=risks.risk_list,
            priority=risks.priority_level
        )

        return dspy.Prediction(
            compliance_score=gdpr_assessment.score,
            risk_level=risks.level,
            recommendations=recommendations.action_items
        )

# Training mit historischen Daten
training_data = [
    dspy.Example(
        contract_text="...",
        contract_type="supplier_agreement",
        compliance_score=8,
        risk_level="low"
    ).with_inputs("contract_text", "contract_type"),
    # ... mehr Beispiele
]

# Automatische Optimierung
optimizer = dspy.BootstrapFewShot(metric=contract_accuracy)
optimized_analyzer = optimizer.compile(ContractAnalyzer(), trainset=training_data)

class CustomerServiceBot(dspy.Module):
    def __init__(self):
        self.classify_intent = dspy.Predict(ClassifyIntent)
        self.extract_context = dspy.ChainOfThought(ExtractContext)
        self.generate_response = dspy.ChainOfThought(GenerateResponse)
        self.escalation_check = dspy.Predict(EscalationCheck)

    def forward(self, customer_message, conversation_history):
        intent = self.classify_intent(message=customer_message)

        if intent.category == "technical_support":
            context = self.extract_context(
                message=customer_message,
                history=conversation_history
            )
            response = self.generate_response(
                intent=intent.category,
                context=context.technical_details
            )
        else:
            # ... andere Kategorien

        escalation = self.escalation_check(
            response=response.text,
            confidence=response.confidence
        )

        return dspy.Prediction(
            response=response.text,
            needs_escalation=escalation.escalate
        )

class ComplianceMonitor(dspy.Module):
    def __init__(self):
        self.scan_document = dspy.ChainOfThought(ScanDocument)
        self.check_regulations = dspy.ChainOfThought(CheckRegulations)
        self.assess_risk = dspy.Predict(AssessRisk)
        self.generate_actions = dspy.ChainOfThought(GenerateActions)

Manuelles Prompt Engineering:
- Senior Developer: 3-6 Monate
- Testing & Iteration: 40-80 Stunden
- Kosten: 45.000-90.000€

DSPy Implementation:
- Senior Developer: 1-2 Monate
- Training Data Prep: 20-40 Stunden
- Kosten: 20.000-40.000€

Manuell:
- Prompt-Anpassungen: 10-20h/Monat
- Quality Regression: Häufig
- Jährliche Kosten: 15.000-30.000€

DSPy:
- Re-Training: 2-5h/Monat
- Automatische Verbesserung: Built-in
- Jährliche Kosten: 3.000-8.000€

# Einfacher Anwendungsfall
class SimpleClassifier(dspy.Module):
    def __init__(self):
        self.classify = dspy.Predict(ClassifyEmail)

    def forward(self, email_text):
        return self.classify(email=email_text)

# 50-100 Beispiele für Training
# Basis-Metriken definieren
# Erste Optimierung

# Komplexere Module
class ProductionWorkflow(dspy.Module):
    def __init__(self):
        self.preprocess = dspy.ChainOfThought(Preprocess)
        self.analyze = dspy.ChainOfThought(Analyze)
        self.postprocess = dspy.Predict(Postprocess)

# Erweiterte Metriken
# A/B Testing Setup
# Monitoring & Alerting

# Multi-Model Support
# Advanced Optimizers
# Integration in bestehende Systeme
# Automatisches Re-Training

# Minimum Setup
CPU: 4 Cores
RAM: 16GB
GPU: Optional (für lokale Models)
Storage: 100GB

# Production Setup
CPU: 8+ Cores
RAM: 32GB+
GPU: RTX 4090 oder Cloud GPU
Storage: 500GB+ SSD

# Core Dependencies
dspy-ai>=2.4.0
transformers>=4.30.0
torch>=2.0.0
openai>=1.0.0  # für GPT-4/3.5
anthropic>=0.25.0  # für Claude

# Optional für lokale Models
vllm>=0.2.0
sentence-transformers>=2.2.0

def contract_accuracy(example, pred, trace=None):
    """Custom metric for contract analysis"""
    score_diff = abs(example.compliance_score - pred.compliance_score)
    risk_match = example.risk_level == pred.risk_level

    accuracy = (score_diff <= 1) * 0.6 + risk_match * 0.4
    return accuracy

def response_quality(example, pred, trace=None):
    """Metric for customer service responses"""
    relevance = semantic_similarity(example.expected, pred.response)
    escalation_correct = example.should_escalate == pred.needs_escalation

    return relevance * 0.7 + escalation_correct * 0.3

# Automatisches Re-Training bei neuen Daten
def retrain_pipeline(new_examples, threshold=0.05):
    current_performance = evaluate_model(current_model, test_set)

    # Neue Daten hinzufügen
    extended_trainset = trainset + new_examples

    # Re-optimierung
    new_model = optimizer.compile(model, trainset=extended_trainset)
    new_performance = evaluate_model(new_model, test_set)

    # Deployment nur bei Verbesserung
    if new_performance > current_performance + threshold:
        deploy_model(new_model)
        log_improvement(current_performance, new_performance)

# Data Validation Pipeline
def validate_training_example(example):
    checks = [
        len(example.input) > 50,  # Mindestlänge
        example.output is not None,  # Vollständige Annotation
        example.quality_score >= 0.8  # Manual Quality Check
    ]
    return all(checks)

clean_trainset = [ex for ex in raw_trainset if validate_training_example(ex)]

# Multi-Model Training
models = ["gpt-4", "claude-3", "local-llama"]
optimized_versions = {}

for model in models:
    dspy.configure(lm=model)
    optimized_versions[model] = optimizer.compile(
        ContractAnalyzer(),
        trainset=training_data
    )

# Prompt Inspection
def analyze_optimized_prompts(compiled_model):
    for module_name, module in compiled_model.named_modules():
        if hasattr(module, 'signature'):
            print(f"Module: {module_name}")
            print(f"Signature: {module.signature}")
            if hasattr(module, 'demos'):
                print(f"Examples: {len(module.demos)}")