Sequentiële ketens

Bekijken als Markdown Bewerken op GitHub Documentatie op 2cc7b2b

Het meest gebruikelijke workflowpatroon is een lineaire keten van stappen. Elke stap wordt uitgevoerd nadat de vorige is voltooid, waarbij de opgebouwde status wordt doorgegeven.

De methode `->then()`

Gebruik ->then() om sequentiële stappen aan een workflow toe te voegen:

PHP

use Queuety\Queuety;

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

Stappen worden uitgevoerd in de volgorde waarin ze zijn toegevoegd. De workflow gaat pas verder als de huidige stap succesvol is voltooid.

De Step-interface

Handlers voor workflowstappen implementeren de 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,
        ];
    }
}

Belangrijkste verschillen met de interface Handler:

	`Handler` (taken)	`Step` (workflows)
Methodesignatuur	`handle(array $payload): void`	`handle(array $state): array`
Invoer	Taakpayload	Opgebouwde workflowstatus
Uitvoer	Niets (void)	Gegevens die met de status worden samengevoegd

Statusoverdracht

Elke stap ontvangt de volledige opgebouwde status en retourneert een array met nieuwe sleutel-waardeparen. De geretourneerde gegevens worden samengevoegd met de status voor volgende stappen:

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

Stapconfiguratie

De methode config() ondersteunt de volgende sleutels:

Sleutel	Type	Standaardwaarde	Beschrijving
`max_attempts`	`int`	`3`	Maximumaantal pogingen voor deze stap
`backoff`	`string`	`exponential`	Wachtstrategie voor nieuwe pogingen

Voorbeeld van een LLM-pijplijn

Een veelgebruikte toepassing is LLM-aanroepen API koppelen aan gegevensverwerking:

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

Verzend de pijplijn:

PHP

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

Als de aanroep OpenAI een time-out krijgt, probeert de worker alleen die stap opnieuw. De prompt die in stap 0 is samengesteld, is al veilig opgeslagen in de workflowstatus.