Chaînes séquentielles

Afficher au format Markdown Modifier sur GitHub Documentation de 2cc7b2b

Le modèle de workflow le plus courant est une chaîne linéaire d’étapes. Chaque étape s’exécute après l’achèvement de la précédente, qui lui transmet l’état accumulé.

La méthode `->then()`

Utilisez ->then() pour ajouter des étapes séquentielles à un workflow :

PHP

use Queuety\Queuety;

Queuety::workflow( 'onboarding' )
    ->then( CreateAccountHandler::class )
    ->then( SendWelcomeEmailHandler::class )
    ->then( NotifyAdminHandler::class )
    ->dispatch( [ 'email' => 'user@example.com' ] );

Les étapes s’exécutent dans l’ordre où elles sont ajoutées. Le workflow ne progresse que lorsque l’étape en cours s’achève avec succès.

L’interface Step

Les gestionnaires d’étapes de workflow implémentent l’interface Queuety\Step :

PHP

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,
        ];
    }
}

Principales différences avec l’interface Handler :

	`Handler` (tâches)	`Step` (workflows)
Signature de méthode	`handle(array $payload): void`	`handle(array $state): array`
Entrée	Charge utile de tâche	État accumulé du workflow
Sortie	Aucune (void)	Données à fusionner dans l’état

Flux d’état

Chaque étape reçoit l’état accumulé complet et renvoie un tableau de nouvelles paires clé-valeur. Les données renvoyées sont fusionnées dans l’état pour les étapes suivantes :

PHP

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 [];
    }
}

Configuration des étapes

La méthode config() accepte les clés suivantes :

Clé	Type	Valeur par défaut	Description
`max_attempts`	`int`	`3`	Nombre maximal de tentatives pour cette étape
`backoff`	`string`	`exponential`	Stratégie de délai avant nouvelle tentative

Exemple de pipeline LLM

Un cas d’usage courant consiste à enchaîner des appels API avec du traitement de données :

PHP

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 [];
    }
}

Lancez le pipeline :

PHP

Queuety::workflow( 'summarize_author' )
    ->then( BuildPromptHandler::class )
    ->then( CallOpenAIHandler::class )
    ->then( SaveSummaryHandler::class )
    ->on_queue( 'llm' )
    ->max_attempts( 5 )
    ->dispatch( [ 'user_id' => 42 ] );

Si l’appel OpenAI expire, le travailleur ne relance que cette étape. L’invite créée à l’étape 0 est déjà enregistrée en toute sécurité dans l’état du workflow.