Workflows

Workflows unterteilen lang laufende Aufgaben in einzelne Schritte. Jeder Schritt speichert seine Ausgabe in einem gemeinsamen Statusobjekt in der Datenbank. Wenn PHP mitten in einem Schritt ausfällt, wiederholt der Worker diesen Schritt und behält den gesamten vorherigen Status bei.

Was sind Workflows?

Ein Workflow ist eine geordnete Abfolge von Schritten. Jeder Schritt ist eine PHP-Klasse, die den gesammelten Status aller vorherigen Schritte erhält und neue Daten zurückgibt, die mit dem Status zusammengeführt werden. Der Workflow schreitet Schritt für Schritt voran; jeder Übergang ist in eine atomare MySQL-Transaktion eingebettet.

use Queuety\Queuety;

$workflow_id = Queuety::workflow( 'generate_report' )
    ->then( FetchDataHandler::class )
    ->then( CallLLMHandler::class )
    ->then( FormatOutputHandler::class )
    ->dispatch( [ 'user_id' => 42 ] );

Status sammeln

Die anfänglichen Nutzdaten, die du an dispatch() übergibst, bilden den Ausgangsstatus. Die zurückgegebenen Schlüssel jedes Schritts werden mit dem Status zusammengeführt. Schlüssel, die mit _ beginnen, sind für Steuerwerte wie _next_step reserviert und werden nicht mit dem öffentlichen Status zusammengeführt:

dispatch(['user_id' => 42])

Step 0: FetchDataHandler   -> returns ['user_name' => 'Alice']
  state: {user_id: 42, user_name: 'Alice'}

Step 1: CallLLMHandler     -> returns ['llm_response' => '...']
  state: {user_id: 42, user_name: 'Alice', llm_response: '...'}

Step 2: FormatOutputHandler -> returns ['report_url' => '/reports/42.pdf']
  state: {user_id: 42, user_name: 'Alice', llm_response: '...', report_url: '/reports/42.pdf'}

Jeder Schritt erhält den vollständig gesammelten Status. So kannst du Daten einfach zwischen Schritten weitergeben, ohne externen Speicher zu benötigen.

Fortsetzen

Workflows sind dauerhaft, weil ihr Status nach jedem Schritt in der Datenbank gespeichert wird. Stürzt der Worker während Schritt 1 ab, startet der Workflow wieder bei Schritt 1, während die Ausgabe von Schritt 0 bereits im Status enthalten ist.

Die Schrittgrenze ist eine einzelne MySQL-Transaktion, die atomar Folgendes ausführt:

1. Aktualisiert den Workflow-Status mit der Ausgabe des Schritts
2. Markiert den Auftrag des aktuellen Schritts als abgeschlossen
3. Reiht den Auftrag des nächsten Schritts ein

So geht garantiert kein Status verloren, auch nicht bei Abstürzen, Zeitüberschreitungen oder schwerwiegenden PHP-Fehlern.

Workflow-Lebenszyklus

dispatch() -> running -> step 0 -> step 1 -> ... -> step N -> completed
                  |                   |
                  |                   +-> failed -> retry -> running (from failed step)
                  |
                  +-> paused -> resumed -> running
                  |
                  +-> waiting_for_signal -> signal received -> running
                  |
                  +-> waiting_for_workflows -> dependency settles -> running
                  |
                  +-> cancelled (via cancel_workflow)

Workflows verwalten

Status prüfen:

$state = Queuety::workflow_status( $workflow_id );
echo $state->status->value;   // 'running'
echo $state->current_step;    // 1
echo $state->total_steps;     // 3
print_r( $state->waiting_for ); // ['42'] while waiting on another workflow
print_r( $state->wait_details ); // matched/remaining blockers, result key, correlation info

Pausieren und fortsetzen:

Queuety::pause_workflow( $workflow_id );
Queuety::resume_workflow( $workflow_id );

Einen fehlgeschlagenen Workflow ab dem fehlgeschlagenen Schritt erneut ausführen:

Queuety::retry_workflow( $workflow_id );

Einen Workflow abbrechen und Bereinigungs-Handler ausführen:

Queuety::cancel_workflow( $workflow_id );

Weitere Informationen zu Bereinigungs-Handlern und den Folgen eines Workflow-Abbruchs findest du unter Abbruch.

Workflow-Optionen

Lege Warteschlange, Priorität und Wiederholungsverhalten für alle Schritte eines Workflows fest:

use Queuety\Enums\Priority;

Queuety::workflow( 'important_pipeline' )
    ->on_queue( 'pipelines' )
    ->with_priority( Priority::High )
    ->max_attempts( 5 )
    ->on_cancel( PipelineCleanupHandler::class )
    ->prune_state_after( 3 )
    ->then( StepA::class )
    ->then( StepB::class )
    ->dispatch( $payload );

Die Methode on_cancel() registriert einen Bereinigungs-Handler, der beim Abbruch des Workflows ausgeführt wird. Siehe Abbruch.

Die Methode prune_state_after() aktiviert das automatische Entfernen alter Schrittausgaben, damit der Workflow-Status begrenzt bleibt. Siehe Statusbereinigung.

Verwende compensate_with(), um einen Kompensations-Handler pro Schritt zuzuweisen. Verwende compensate_on_failure(), wenn abgeschlossene Schrittkompensationen nach einem Workflow-Fehler automatisch ausgeführt werden sollen, nicht nur bei einem ausdrücklichen Abbruch. Siehe Kompensation.

Verwende Workflow-Schutzmaßnahmen, wenn du eine sicherere Orchestrierung lang laufender Prozesse benötigst: version() versieht eine Definition mit einer Versionsbezeichnung auf Anwendungsebene, Queuety speichert für jede Ausführung zusätzlich einen deterministischen Definitions-Hash, idempotency_key() macht den Start bei Wiederholungen oder doppelten Anfragen dauerhaft zuverlässig und max_transitions(), max_for_each_items() und max_state_bytes() lassen außer Kontrolle geratene Ausführungen fehlschlagen, bevor sie eskalieren. Siehe Schutzmaßnahmen.

Schritttypen

Queuety unterstützt elf Schritttypen innerhalb eines Workflows:

• Sequenzielle Ketten mit ->then() für die Ausführung nacheinander
• Parallele Gruppen mit ->parallel() für die gleichzeitige Ausführung
• Dynamische For-each-Schritte mit ->for_each() für zur Laufzeit ermittelte Zweige und festgelegte Abschlussbedingungen
• Asynchrone Übergaben mit ->start_workflows() oder ->start_agents(), um unabhängige untergeordnete Workflows aus zur Laufzeit ermittelten Elementen zu starten
• Bedingte Verzweigungen mit benannten Schritten und _next_step
• Wiederholungen mit ->repeat_until() und ->repeat_while() für dauerhafte Rücksprünge zu früheren benannten Schritten
• Workflows ausführen mit ->run_workflow() für verschachtelte Komposition
• Dauerhafte Verzögerungen mit ->delay() für zeitbasierte Verzögerungen
• Signale mit ->wait_for_signal(), ->wait_for_signals(), ->wait_for_approval(), ->wait_for_input() und ->wait_for_decision() für externe Ereignisauslöser
• Workflow-Abhängigkeiten mit ->wait_for_workflow(), ->wait_for_workflows() und ->wait_for_agents() zur Koordination zwischen Workflows
• Streaming-Schritte mit der Schnittstelle StreamingStep für dauerhafte, in Blöcken ausgegebene Daten

Häufige Anwendungsfälle

Wenn du autonome oder teilautonome Systeme entwickelst, beginne mit Agenten-Orchestrierung. Dort wird gezeigt, wie for_each(), start_agents(), wait_for_agents(), wait_for_workflow() und wait_for_decision() in echten Planer-Ausführer- und Human-Review-Abläufen zusammenwirken.