Tasks & Workflows

Tasks sind das Herzstück jeder CrewAI-Anwendung. Eine gut definierte Task sagt dem Agent genau, was zu tun ist, welches Format erwartet wird und welcher Kontext zur Verfügung steht.

Task-Definition im Detail

from crewai import Task

analysis_task = Task(
    description=(
        "Analysiere die bereitgestellten Verkaufsdaten für Q4 2025. "
        "Identifiziere die Top-3-Produkte nach Umsatz, "
        "berechne Wachstumsraten gegenüber Q3, "
        "und identifiziere Anomalien oder auffällige Trends."
    ),
    expected_output=(
        "Ein strukturierter Analysebericht mit: "
        "1) Top-3-Produkte mit Umsatzzahlen, "
        "2) Quartal-über-Quartal-Wachstumsraten, "
        "3) Identifizierte Anomalien mit Erklärung, "
        "4) Drei konkrete Handlungsempfehlungen"
    ),
    agent=data_analyst,
    output_file="reports/q4_analysis.md"
)

Wichtige Task-Parameter

Context Passing

Tasks können die Ergebnisse vorheriger Tasks als Kontext nutzen:

research_task = Task(
    description="Recherchiere aktuelle Markttrends für {topic}",
    expected_output="Zusammenfassung der Top-5-Trends",
    agent=researcher
)

analysis_task = Task(
    description="Analysiere die recherchierten Trends auf Relevanz für unser Unternehmen",
    expected_output="Priorisierte Trend-Analyse mit Impact-Bewertung",
    agent=analyst,
    context=[research_task]  # Erhält Output von research_task
)

writing_task = Task(
    description="Erstelle einen Management-Bericht basierend auf der Analyse",
    expected_output="Professioneller 2-Seiten-Bericht",
    agent=writer,
    context=[research_task, analysis_task]  # Erhält beide Outputs
)

Task Dependencies

Im Sequential Process bestimmt die Reihenfolge der Tasks in der Liste die Ausführungsreihenfolge. Mit Context-Referenzen wird die Abhängigkeit explizit:

research_task (keine Abhängigkeit)
    ↓ Output als Context
analysis_task (abhängig von research_task)
    ↓ Output als Context
writing_task (abhängig von research + analysis)

Callback Functions

Führen Sie nach jeder Task eine Funktion aus — für Logging, Notifications oder Datenverarbeitung:

def on_task_complete(output):
    # Ergebnis in Datenbank speichern
    db.save_result(output.raw)
    # Notification senden
    slack.send(f"Task abgeschlossen: {output.description}")
    # Metriken tracken
    metrics.track("task_completed", {"tokens": output.token_usage})

analysis_task = Task(
    description="Analysiere die Daten...",
    expected_output="Strukturierter Bericht",
    agent=analyst,
    callback=on_task_complete
)

Human Input

Für kritische Tasks können Sie menschliche Überprüfung erzwingen:

approval_task = Task(
    description="Erstelle den finalen Kundenbericht",
    expected_output="Freigegebener Bericht im PDF-Format",
    agent=writer,
    human_input=True  # Pausiert und wartet auf menschliche Eingabe
)

Workflow-Beispiel: Content Pipeline

crew = Crew(
    agents=[researcher, seo_expert, writer, editor],
    tasks=[
        Task(description="Keyword-Research...", agent=seo_expert),
        Task(description="Themen-Research...", agent=researcher, context=[task_0]),
        Task(description="Artikel schreiben...", agent=writer, context=[task_0, task_1]),
        Task(description="Artikel redigieren...", agent=editor, context=[task_2], human_input=True),
    ],
    process=Process.sequential
)

Praxis-Tipp: Definieren Sie expected_output so präzise wie möglich. "Ein Bericht" ist zu vage. "Ein strukturierter Bericht mit Executive Summary, 3 Kernpunkten und konkreten Handlungsempfehlungen" gibt dem Agent klare Leitplanken.