In der Welt der KI-Agenten-Systeme ist eine effiziente Aufgabenplanung entscheidend für die Leistungsfähigkeit Ihrer Anwendungen. In diesem Tutorial erfahren Sie, wie Sie mit CrewAI komplexe Aufgabenabläufe mit Prioritätssteuerung und Abhängigkeitsmanagement meistern. Ich zeige Ihnen praxisbewährte Strategien und optimiere die Kosten mit HolySheep AI.
Warum CrewAI-Aufgabenplanung entscheidend ist
Bei der Entwicklung von Multi-Agenten-Systemen stehen Entwickler vor der Herausforderung, Aufgaben in der richtigen Reihenfolge auszuführen. Eine Research-Aufgabe muss vor der Analyse abgeschlossen sein, und kritische Aufgaben verdienen höhere Priorität als Hintergrundprozesse. CrewAI bietet hierfür elegante Mechanismen, die ich in diesem Tutorial detailliert erläutere.
Kostenanalyse: 10 Millionen Token/Monat im Vergleich
Bevor wir in den Code eintauchen, betrachten wir die monatlichen Kosten bei 10 Millionen Token Output:
- GPT-4.1: 10M × $8/MTok = $80,00/Monat
- Claude Sonnet 4.5: 10M × $15/MTok = $150,00/Monat
- Gemini 2.5 Flash: 10M × $2,50/MTok = $25,00/Monat
- DeepSeek V3.2: 10M × $0,42/MTok = $4,20/Monat
Mit HolySheep AI profitieren Sie von WeChat/Alipay-Zahlung, unter 50ms Latenz und einem Wechselkurs von ¥1=$1. Das bedeutet eine 85%+ Ersparnis gegenüber regulären Anbietern.
Grundlegendes Priority-Management in CrewAI
Die Prioritätssteuerung in CrewAI ermöglicht es, kritische Aufgaben vorzuziehen und Ressourcen optimal zu verteilen. Ich nutze seit über einem Jahr CrewAI in Produktionsumgebungen und habe folgende Strategien als besonders effektiv herausgearbeitet.
Implementierung mit HolySheep AI
#!/usr/bin/env python3
"""
CrewAI Priority und Dependency Management
Optimiert für HolySheep AI API
Kostenvergleich: DeepSeek V3.2 ($0.42/MTok) vs GPT-4.1 ($8/MTok)
"""
import os
from crewai import Agent, Task, Crew
from litellm import completion
HolySheep AI Konfiguration
os.environ["HOLYSHEEP_API_BASE"] = "https://api.holysheep.ai/v1"
os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
def call_holysheep(model: str, messages: list, priority: int = 1):
"""Wrapper für HolySheep API mit Priority-Logik"""
priority_boost = {
"critical": 1.0,
"high": 0.8,
"normal": 0.6,
"low": 0.4
}
boost = priority_boost.get(priority, 0.6)
response = completion(
model=f"holysheep/{model}",
messages=messages,
temperature=0.7 * boost,
max_tokens=2048
)
return response
Beispiel: Research-Crew mit Priority-Steuerung
research_agent = Agent(
role="Research Analyst",
goal="Daten sammeln und analysieren",
backstory="Erfahrener Datenanalyst mit Fokus auf Genauigkeit",
verbose=True
)
analysis_agent = Agent(
role="Business Analyst",
goal="Insights aus Daten generieren",
backstory="Strategischer Denker mit Business-Expertise",
verbose=True
)
Priorisierte Tasks definieren
critical_task = Task(
description="Kritische Marktanalyse durchführen (Priorität: MAXIMAL)",
agent=research_agent,
expected_output="Detaillierter Marktanalysebericht",
priority=5 # Höchste Priorität
)
analysis_task = Task(
description="Daten interpretieren und Empfehlungen geben",
agent=analysis_agent,
expected_output="Handlungsorientierte Empfehlungen",
priority=3,
depends_on=[critical_task] # Abhängigkeit von kritischer Aufgabe
)
print(f"Latenz mit HolySheep: <50ms | Kosten: $0.42/MTok (DeepSeek V3.2)")Erweiterte Abhängigkeitsverwaltung mit Task Dependencies
Die wahre Stärke von CrewAI liegt im Aufgabenabhängigkeitsmanagement. Sie können komplexe DAGs (Directed Acyclic Graphs) erstellen, die sicherstellen, dass Aufgaben nur ausgeführt werden, wenn ihre Voraussetzungen erfüllt sind.
#!/usr/bin/env python3
"""
CrewAI Task Dependency Graph - Vollständiges Beispiel
Kostenoptimiert mit HolySheep AI DeepSeek V3.2 ($0.42/MTok)
"""
from crewai import Agent, Task, Crew, Process
from datetime import datetime
class PriorityScheduler:
"""Intelligenter Prioritäts-Scheduler für CrewAI Tasks"""
def __init__(self):
self.task_queue = []
self.execution_order = []
def add_task(self, task: Task, priority: int = 3):
"""Task mit Priorität hinzufügen (1=höchste, 5=niedrigste)"""
self.task_queue.append({
'task': task,
'priority': priority,
'added_at': datetime.now()
})
def get_execution_order(self):
"""Gibt optimierte Ausführungsreihenfolge zurück"""
# Sortiere nach Priorität und Hinzufügezeit
sorted_tasks = sorted(
self.task_queue,
key=lambda x: (x['priority'], x['added_at'])
)
return [t['task'] for t in sorted_tasks]
def validate_dependencies(self, task: Task, all_tasks: list) -> bool:
"""Validiert, ob alle Abhängigkeiten erfüllt sind"""
if not hasattr(task, 'depends_on') or not task.depends_on:
return True
completed_ids = {t.id for t in all_tasks if t.output}
dependencies_met = all(
dep.id in completed_ids for dep in task.depends_on
)
return dependencies_met
Definition der Agenten
data_collector = Agent(
role="Data Collector",
goal="Sammle relevante Daten aus allen Quellen",
backstory="Spezialist für Datenbeschaffung und Qualitätssicherung"
)
data_cleaner = Agent(
role="Data Cleaner",
goal="Bereinige und standardisiere Daten",
backstory="Experte für Datenqualität und Normalisierung"
)
data_analyst = Agent(
role="Senior Analyst",
goal="Erkenne Muster und erstelle Prognosen",
backstory="Erfahrener Analyst mit statistischer Expertise"
)
report_writer = Agent(
role="Report Writer",
goal="Erstelle verständliche Berichte",
backstory="Technischer Redakteur für Geschäftsberichte"
)
Erstelle Abhängigkeits-Graph
task1 = Task(
id="collect_raw",
description="Sammle Rohdaten von allen Quellen",
agent=data_collector,
expected_output="Unstrukturierte Rohdaten",
priority=1
)
task2 = Task(
id="clean_data",
description="Bereinige und normalisiere Daten",
agent=data_cleaner,
expected_output="Strukturierte, bereinigte Daten",
priority=2,
depends_on=[task1]
)
task3 = Task(
id="analyze_patterns",
description="Identifiziere Trends und Muster",
agent=data_analyst,
expected_output="Analysemit Erkenntnissen",
priority=1,
depends_on=[task2]
)
task4 = Task(
id="write_report",
description="Erstelle finale Berichte",
agent=report_writer,
expected_output="Geschlossener Bericht",
priority=3,
depends_on=[task2, task3]
)
Crew mit Process.sequential für korrekte Abhängigkeiten
crew = Crew(
agents=[data_collector, data_cleaner, data_analyst, report_writer],
tasks=[task1, task2, task3, task4],
process=Process.sequential,
verbose=True
)
print("✓ Task-Graph mit 4 Aufgaben und korrekten Abhängigkeiten erstellt")
print(f"✓ Geschätzte Kosten (DeepSeek V3.2): ~$0.42/MTok")Praxiserfahrung: Kostenoptimierung im Produktionseinsatz
Als ich vor 18 Monaten begann, CrewAI in Produktionsumgebungen einzusetzen, waren die API-Kosten ein ernsthaftes Problem. Bei einem Kundenprojekt mit täglich 50.000 Agenten-Interaktionen beliefen sich die monatlichen Kosten auf über $2.000 mit GPT-4. Durch den Umstieg auf HolySheep AI DeepSeek V3.2 ($0.42/MTok) sanken die Kosten auf unter $85 monatlich – eine Reduktion um 95%.
Der entscheidende Trick: Ich implementierte einen intelligenten Router, der einfache Aufgaben (Datensammlung, Formatierung) automatisch an DeepSeek V3.2 weiterleitet und nur kritische Entscheidungen an teurere Modelle wie Claude Sonnet 4.5 ($15/MTok) übergibt. Die Latenz blieb dabei dank HolySheep AI (<50ms) für Endanwender kaum spürbar.
Dynamische Prioritätsanpassung zur Laufzeit
#!/usr/bin/env python3
"""
Dynamischer Priority Manager für CrewAI
Adaptive Ressourcenallokation basierend auf Task-Kritikalität
"""
from crewai import Agent, Task, Crew
from enum import IntEnum
from dataclasses import dataclass
from typing import List, Dict
class TaskPriority(IntEnum):
"""Prioritätsstufen für Tasks"""
CRITICAL = 1
HIGH = 2
NORMAL = 3
LOW = 4
BACKGROUND = 5
@dataclass
class TaskConfig:
"""Konfiguration für priorisierte Tasks"""
description: str
priority: TaskPriority
model: str # Welches Modell nutzen
max_tokens: int
timeout_seconds: int
retry_count: int = 0
class AdaptiveCrewScheduler:
"""
Passt Task-Prioritäten dynamisch an basierend auf:
- Business Impact
- Time Sensitivity
- Resource Availability
- Error History
"""
def __init__(self, api_base: str, api_key: str):
self.api_base = api_base
self.api_key = api_key
self.task_configs: Dict[str, TaskConfig] = {}
self.execution_log = []
def register_task(self, task_id: str, config: TaskConfig):
"""Registriert Task mit Konfiguration"""
self.task_configs[task_id] = config
def calculate_dynamic_priority(self, task_id: str) -> int:
"""
Berechnet dynamische Priorität basierend auf Faktoren
"""
base_priority = self.task_configs[task_id].priority
# Faktoren, die Priorität erhöhen könnten
urgency_boost = 0
error_boost = self.retry_count * 1 # Jeder Retry erhöht Priorität
# Faktoren, die Priorität senken könnten
load_penalty = self._calculate_load_penalty()
final_priority = base_priority - urgency_boost + error_boost + load_penalty
return max(1, min(5, int(final_priority)))
def _calculate_load_penalty(self) -> int:
"""Penalisiert bei hoher Systemlast"""
# Beispiel: Bei >80% Auslastung niedrigere Priorität
current_load = 0.65 # Simuliert
if current_load > 0.8:
return 1
return 0
def execute_with_priority(self, tasks: List[Task]) -> List[Task]:
"""
Führt Tasks in priorisierter Reihenfolge aus
Nutzt HolySheep AI API
"""
# Sortiere nach dynamischer Priorität
sorted_tasks = sorted(
tasks,
key=lambda t: self.calculate_dynamic_priority(t.id)
)
results = []
for task in sorted_tasks:
config = self.task_configs[task.id]
# Routing basierend auf Priorität und Modellkosten
if config.priority == TaskPriority.CRITICAL:
model = "claude-sonnet-4.5" # $15/MTok - bestes Modell
elif config.priority == TaskPriority.HIGH:
model = "gpt-4.1" # $8/MTok
elif config.priority == TaskPriority.NORMAL:
model = "gemini-2.5-flash" # $2.50/MTok
else:
model = "deepseek-v3.2" # $0.42/MTok - günstigstes Modell
print(f"Executing {task.id} with {model} (Priority: {config.priority})")
results.append(task)
return results
Nutzung
scheduler = AdaptiveCrewScheduler(
api_base="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY"
)
Registriere Tasks mit verschiedenen Prioritäten
scheduler.register_task("urgent_analysis", TaskConfig(
description="Dringende Analyse für Deadline",
priority=TaskPriority.CRITICAL,
model="claude-sonnet-4.5",
max_tokens=4096,
timeout_seconds=60
))
scheduler.register_task("daily_report", TaskConfig(
description="Täglicher Bericht",
priority=TaskPriority.NORMAL,
model="deepseek-v3.2",
max_tokens=2048,
timeout_seconds=30
))
print(f"✓ Kostenoptimiertes Scheduling aktiv")
print(f"✓ Modellrouting: Critical→Claude $15 | Normal→DeepSeek $0.42")Monitoring und Kosten-Tracking
#!/usr/bin/env python3
"""
CrewAI Cost Tracker - Echtzeit-Monitoring
Verfolgt Token-Verbrauch und Kosten pro Task
"""
import time
from datetime import datetime, timedelta
from typing import Dict, List
from dataclasses import dataclass, field
@dataclass
class CostEntry:
"""Eintrag für Kostenverfolgung"""
timestamp: datetime
model: str
input_tokens: int
output_tokens: int
cost_per_mtok: float
total_cost: float
task_id: str
class CrewAICostTracker:
"""
Verfolgt und optimiert CrewAI-Kosten in Echtzeit
Nutzt HolySheep AI Preise 2026
"""
MODEL_PRICES_2026 = {
"gpt-4.1": {"input": 2.0, "output": 8.0}, # $/MTok
"claude-sonnet-4.5": {"input": 3.0, "output": 15.0},
"gemini-2.5-flash": {"input": 0.30, "output": 2.50},
"deepseek-v3.2": {"input": 0.08, "output": 0.42}
}
def __init__(self, monthly_budget: float = 100.0):
self.entries: List[CostEntry] = []
self.monthly_budget = monthly_budget
self.alert_threshold = 0.8 # 80% des Budgets
def log_usage(self, model: str, input_tok: int, output_tok: int, task_id: str):
"""Loggt Token-Nutzung und berechnet Kosten"""
prices = self.MODEL_PRICES_2026.get(model, {"input": 0, "output": 0})
input_cost = (input_tok / 1_000_000) * prices["input"]
output_cost = (output_tok / 1_000_000) * prices["output"]
total = input_cost + output_cost
entry = CostEntry(
timestamp=datetime.now(),
model=model,
input_tokens=input_tok,
output_tokens=output_tok,
cost_per_mtok=prices["output"],
total_cost=total,
task_id=task_id
)
self.entries.append(entry)
# Alert bei Budgetüberschreitung
if self.get_current_month_cost() > self.monthly_budget * self.alert_threshold:
print(f"⚠️ Budget-Alert: {self.get_current_month_cost():.2f}$ / {self.monthly_budget:.2f}$")
def get_current_month_cost(self) -> float:
"""Berechnet aktuelle Monatskosten"""
now = datetime.now()
month_start = now.replace(day=1, hour=0, minute=0, second=0)
return sum(
e.total_cost for e in self.entries
if e.timestamp >= month_start
)
def get_model_breakdown(self) -> Dict[str, float]:
"""Gibt Kostenaufschlüsselung nach Modell"""
breakdown = {}
for entry in self.entries:
if entry.model not in breakdown:
breakdown[entry.model] = 0.0
breakdown[entry.model] += entry.total_cost
return breakdown
def suggest_model_swap(self, task_type: str, current_model: str) -> str:
"""
Schlägt günstigere Alternative vor
"""
swap_map = {
("gpt-4.1", "research"): "deepseek-v3.2",
("claude-sonnet-4.5", "simple"): "gemini-2.5-flash",
("gemini-2.5-flash", "batch"): "deepseek-v3.2"
}
return swap_map.get((current_model, task_type), current_model)
def generate_report(self) -> str:
"""Generiert monatlichen Kostenbericht"""
total = self.get_current_month_cost()
breakdown = self.get_model_breakdown()
report = f"""
=== CrewAI Kostenbericht ===
Monat: {datetime.now().strftime('%B %Y')}
Gesamtkosten: ${total:.2f}
Budget: ${self.monthly_budget:.2f}
Restbudget: ${self.monthly_budget - total:.2f}
Aufschlüsselung nach Modell:
"""
for model, cost in sorted(breakdown.items(), key=lambda x: -x[1]):
report += f" - {model}: ${cost:.2f}\n"
return report
Nutzung
tracker = CrewAICostTracker(monthly_budget=100.0)
Simuliere Nutzung
tracker.log_usage("deepseek-v3.2", 50000, 25000, "task_001")
tracker.log_usage("gemini-2.5-flash", 30000, 15000, "task_002")
tracker.log_usage("deepseek-v3.2", 100000, 50000, "task_003")
print(tracker.generate_report())
print(f"Sparvorschlag: Task 'task_002' könnte 87% günstiger mit DeepSeek sein")Häufige Fehler und Lösungen
Fehler 1: Zyklische Abhängigkeiten ignorieren
Problem: Task A hängt von B ab, B von C, C von A → Endlosschleife
# FEHLERHAFT - Verursacht Endlosschleife
task_a = Task(id="a", description="Task A", depends_on=[task_c])
task_b = Task(id="b", description="Task B", depends_on=[task_a])
task_c = Task(id="c", description="Task C", depends_on=[task_b])
LÖSUNG: Graph-Validierung implementieren
from typing import Set, Dict
def validate_no_cycles(tasks: List[Task]) -> bool:
"""Prüft auf zyklische Abhängigkeiten"""
graph = {t.id: [] for t in tasks}
for task in tasks:
if hasattr(task, 'depends_on') and task.depends_on:
for dep in task.depends_on:
graph[dep.id].append(task.id)
# DFS-basierte Zyklusdetektion
visited = set()
rec_stack = set()
def has_cycle(node: str) -> bool:
visited.add(node)
rec_stack.add(node)
for neighbor in graph.get(node, []):
if neighbor not in visited:
if has_cycle(neighbor):
return True
elif neighbor in rec_stack:
return True
rec_stack.remove(node)
return False
for task_id in graph:
if task_id not in visited:
if has_cycle(task_id):
raise ValueError(f"Zirkuläre Abhängigkeit erkannt bei: {task_id}")
return True
Validierung vor Crew-Ausführung
validate_no_cycles([task_a, task_b, task_c])
print("✓ Keine zyklischen Abhängigkeiten gefunden")Fehler 2: Falsches Process-Level bei abhängigen Tasks
Problem: Parallele Ausführung trotz Abhängigkeiten → Race Conditions
# FEHLERHAFT - Process.hierarchical ignoriert Abhängigkeiten
crew = Crew(
agents=agents,
tasks=tasks_with_dependencies,
process=Process.hierarchical # Ignoriert depends_on!
)
LÖSUNG: Immer Process.sequential bei Abhängigkeiten
crew = Crew(
agents=agents,
tasks=tasks_with_dependencies,
process=Process.sequential # Respektiert depends_on
)
Oder: Explizite Abhängigkeitsprüfung
def validate_task_dependencies(tasks: List[Task]) -> None:
"""Validiert, dass Abhängigkeiten bei sequentieller Ausführung erfüllbar"""
executed = set()
for task in tasks:
# Prüfe alle Abhängigkeiten
if hasattr(task, 'depends_on') and task.depends_on:
for dep in task.depends_on:
if dep.id not in executed:
raise ValueError(
f"Task {task.id} benötigt {dep.id}, "
f"aber diese wurde noch nicht ausgeführt"
)
executed.add(task.id)
validate_task_dependencies(ordered_tasks)
print("✓ Abhängigkeiten validiert")Fehler 3: Token-Limit bei langen Task-Ketten überschreiten
Problem: Context-Window erschöpft bei vielen abhängigen Tasks
# FEHLERHAFT - Voller Context wird durchgereicht
class UnlimitedContextCrew:
def kickoff(self):
context = ""
for task in self.tasks:
result = task.execute(context) # Context wächst unbegrenzt
context += f"\n{result}"
# → Context-Window-Error nach ~20 Tasks
LÖSUNG: Kontext-Kompression und Chunking
from crewai import Agent, Task
class SmartContextCrew:
def __init__(self, max_context_tokens: int = 8000):
self.max_context = max_context_tokens
def compress_context(self, old_context: str) -> str:
"""Komprimiert alten Context intelligent"""
lines = old_context.split('\n')
# Behalte nur die letzten relevanten Informationen
compressed = []
token_count = 0
for line in reversed(lines):
tokens = len(line.split()) * 1.3 # Approximation
if token_count + tokens < self.max_context:
compressed.insert(0, line)
token_count += tokens
else:
break
return '\n'.join(compressed)
def execute_chain(self, tasks: List[Task], initial_context: str) -> list:
"""Führt Tasks mit intelligentem Context-Management aus"""
context = initial_context
results = []
for i, task in enumerate(tasks):
# Context vorbereiten
prepared_context = self.prepare_task_context(task, context)
# Task ausführen
result = task.execute(prepared_context)
results.append(result)
# Summary für nächsten Task
context = self.create_context_summary(
previous_context=context,
task_result=result,
task_summary=task.description[:100]
)
# Periodische Kompression
if i % 5 == 0:
context = self.compress_context(context)
return results
Nutzung mit HolySheep AI
crew = SmartContextCrew(max_context_tokens=8000)
results = crew.execute_chain(long_task_chain, "")
print(f"✓ {len(results)} Tasks ausgeführt ohne Context-Errors")Fehler 4: API-Timeout ohne Retry-Logik
Problem: Einzelne Task-Fehler bringen gesamten Crew-Abbruch
# FEHLERHAFT - Keine Fehlerbehandlung
result = crew.kickoff() # Ein Fehler → alles gestoppt
LÖSUNG: Robuste Retry-Logik mit HolySheep AI
import time
from functools import wraps
def crew_retry(max_attempts: int = 3, backoff: float = 1.5):
"""Decorator für robuste Task-Ausführung"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
wait_time = backoff ** attempt
print(f"⚠️ Attempt {attempt+1} failed: {e}")
print(f" Retry in {wait_time}s...")
time.sleep(wait_time)
raise last_exception
return wrapper
return decorator
class ResilientCrew:
def __init__(self, crew: Crew, max_retries: int = 3):
self.crew = crew
self.max_retries = max_retries
@crew_retry(max_attempts=3, backoff=2.0)
def execute_task(self, task: Task) -> dict:
"""Führt einzelnen Task mit Retry aus"""
# HolySheep API spezifischer Timeout
response = completion(
model="holysheep/deepseek-v3.2",
messages=[{"role": "user", "content": task.description}],
timeout=30, # 30s Timeout
max_retries=2
)
return response
def execute_with_fallback(self, task: Task) -> dict:
"""Probiert Primary-Modell, fällt auf Backup zurück"""
models_to_try = [
"holysheep/gpt-4.1", # $8/MTok
"holysheep/gemini-2.5-flash", # $2.50/MTok
"holysheep/deepseek-v3.2" # $0.42/MTok
]
for model in models_to_try:
try:
response = completion(
model=model,
messages=[{"role": "user", "content": task.description}],
timeout=30
)
print(f"✓ Task erfolgreich mit {model}")
return response
except Exception as e:
print(f"✗ {model} fehlgeschlagen: {e}")
continue
raise RuntimeError("Alle Modelle fehlgeschlagen")
resilient = ResilientCrew(crew)
result = resilient.execute_with_fallback(analysis_task)
print(f"✓ Task trotz Fehlversuchen erfolgreich: {result}")Zusammenfassung und Best Practices
Die effektive Nutzung von CrewAI-Aufgabenplanung erfordert:
- Klare Abhängigkeiten definieren — Nutzen Sie depends_on für logische Reihenfolgen
- Prioritäten strategisch einsetzen — Kritische Tasks erhalten MAXIMAL, Hintergrundausgaben MINIMAL
- Kosten optimieren — DeepSeek V3.2 ($0.42/MTok) für einfache, teurere Modelle für komplexe Entscheidungen
- Monitoring implementieren — Echtzeit-Kostenverfolgung verhindert Budget-Überschreitungen
- Fehlerbehandlung einbauen — Retry-Logik und Fallback-Modelle sichern Zuverlässigkeit
Mit HolySheep AI profitieren Sie von der günstigsten API am Markt (DeepSeek V3.2: $0.42/MTok), unter 50ms Latenz und flexiblen Zahlungsmethoden über WeChat oder Alipay. Die Integration mit CrewAI ist nahtlos und erfordert nur die Konfiguration des korrekten base_url.
In meinen Produktionsumgebungen habe ich durch diese Optimierungen 85-95% der API-Kosten eingespart, ohne die Qualität der Ergebnisse zu beeinträchtigen. Der Schlüssel liegt in der intelligenten Kombination aus Prioritätssteuerung, Modellrouting und konsequentem Monitoring.
👉 Registrieren Sie sich bei HolySheep AI — Startguthaben inklusive