Sequenzielle Ketten
Das gängigste Workflow-Muster ist eine lineare Kette von Schritten. Jeder Schritt wird ausgeführt, nachdem der vorherige abgeschlossen ist, wobei der angesammelte Zustand weitergegeben wird.
Die Methode ->then()
Verwende ->then(), um einem Workflow sequenzielle Schritte hinzuzufügen:
use Queuety\Queuety;
Queuety::workflow( 'onboarding' )
->then( CreateAccountHandler::class )
->then( SendWelcomeEmailHandler::class )
->then( NotifyAdminHandler::class )
->dispatch( [ 'email' => 'user@example.com' ] );
Die Schritte werden in der Reihenfolge ausgeführt, in der sie hinzugefügt wurden. Der Workflow wird erst fortgesetzt, wenn der aktuelle Schritt erfolgreich abgeschlossen ist.
Die Step-Schnittstelle
Handler für Workflow-Schritte implementieren die Schnittstelle Queuety\Step:
use Queuety\Step;
class CreateAccountHandler implements Step {
public function handle( array $state ): array {
$user_id = wp_create_user( $state['email'], wp_generate_password() );
return [ 'user_id' => $user_id ];
}
public function config(): array {
return [
'max_attempts' => 3,
];
}
}
Wichtige Unterschiede zur Schnittstelle Handler:
Handler (Jobs) |
Step (Workflows) |
|
|---|---|---|
| Methodensignatur | handle(array $payload): void |
handle(array $state): array |
| Eingabe | Job-Nutzlast | Angesammelter Workflow-Zustand |
| Ausgabe | Nichts (void) | Daten, die mit dem Zustand zusammengeführt werden |
Zustandsfluss
Jeder Schritt erhält den gesamten angesammelten Zustand und gibt ein Array neuer Schlüssel-Wert-Paare zurück. Die zurückgegebenen Daten werden mit dem Zustand für nachfolgende Schritte zusammengeführt:
class FetchUserHandler implements Step {
public function handle( array $state ): array {
// $state contains: ['user_id' => 42]
$user = get_user_by( 'ID', $state['user_id'] );
return [ 'user_name' => $user->display_name, 'user_email' => $user->user_email ];
}
public function config(): array {
return [];
}
}
class GenerateReportHandler implements Step {
public function handle( array $state ): array {
// $state contains: ['user_id' => 42, 'user_name' => 'Alice', 'user_email' => 'alice@example.com']
$report = build_report( $state['user_name'] );
return [ 'report_url' => $report->url ];
}
public function config(): array {
return [];
}
}
Schritt konfigurieren
Die Methode config() unterstützt folgende Schlüssel:
| Schlüssel | Typ | Standardwert | Beschreibung |
|---|---|---|---|
max_attempts |
int |
3 |
Maximale Anzahl der Wiederholungsversuche für diesen Schritt |
backoff |
string |
exponential |
Wartezeitstrategie für Wiederholungen |
Beispiel für eine LLM-Pipeline
Ein gängiger Anwendungsfall ist die Verkettung von LLM-Aufrufen API mit Datenverarbeitung:
use Queuety\Step;
class BuildPromptHandler implements Step {
public function handle( array $state ): array {
$user = get_user_by( 'ID', $state['user_id'] );
$posts = get_posts( [ 'author' => $user->ID, 'numberposts' => 10 ] );
$context = implode( "\n", array_map( fn( $p ) => $p->post_content, $posts ) );
return [
'prompt' => "Summarize this author's recent work:\n\n{$context}",
];
}
public function config(): array {
return [];
}
}
class CallOpenAIHandler implements Step {
public function handle( array $state ): array {
$response = wp_remote_post( 'https://api.openai.com/v1/chat/completions', [
'timeout' => 120,
'headers' => [
'Authorization' => 'Bearer ' . OPENAI_API_KEY,
'Content-Type' => 'application/json',
],
'body' => json_encode( [
'model' => 'gpt-4',
'messages' => [ [ 'role' => 'user', 'content' => $state['prompt'] ] ],
] ),
] );
$body = json_decode( wp_remote_retrieve_body( $response ), true );
return [ 'summary' => $body['choices'][0]['message']['content'] ];
}
public function config(): array {
return [ 'max_attempts' => 5 ];
}
}
class SaveSummaryHandler implements Step {
public function handle( array $state ): array {
$post_id = wp_insert_post( [
'post_title' => "Summary for user #{$state['user_id']}",
'post_content' => $state['summary'],
'post_status' => 'publish',
] );
return [ 'summary_post_id' => $post_id ];
}
public function config(): array {
return [];
}
}
Starte die Pipeline:
Queuety::workflow( 'summarize_author' )
->then( BuildPromptHandler::class )
->then( CallOpenAIHandler::class )
->then( SaveSummaryHandler::class )
->on_queue( 'llm' )
->max_attempts( 5 )
->dispatch( [ 'user_id' => 42 ] );
Wenn der OpenAI-Aufruf wegen eines Zeitlimits abbricht, wiederholt der Worker nur diesen Schritt. Der in Schritt 0 erstellte Prompt ist bereits sicher im Workflow-Zustand gespeichert.