Sekventiella kedjor

Visa som Markdown Redigera på GitHub Dokumentation på 2cc7b2b

Det vanligaste arbetsflödesmönstret är en linjär kedja av steg. Varje steg körs när det föregående har slutförts, och det samlade tillståndet förs vidare.

Metoden `->then()`

Använd ->then() för att lägga till sekventiella steg i ett arbetsflöde:

PHP

use Queuety\Queuety;

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

Stegen körs i den ordning de läggs till. Arbetsflödet går vidare först när det aktuella steget har slutförts utan fel.

Gränssnittet Step

Hanterare för arbetsflödessteg implementerar gränssnittet 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,
        ];
    }
}

Viktiga skillnader jämfört med gränssnittet Handler:

	`Handler` (jobb)	`Step` (arbetsflöden)
Metodsignatur	`handle(array $payload): void`	`handle(array $state): array`
Indata	Jobbets nyttolast	Samlat arbetsflödestillstånd
Utdata	Inget (void)	Data som ska slås samman med tillståndet

Tillståndsflöde

Varje steg får hela det samlade tillståndet och returnerar en array med nya nyckel-värde-par. De returnerade uppgifterna slås samman med tillståndet för efterföljande steg:

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

Stegkonfiguration

Metoden config() stöder följande nycklar:

Nyckel	Typ	Standardvärde	Beskrivning
`max_attempts`	`int`	`3`	Maximalt antal omförsök för det här steget
`backoff`	`string`	`exponential`	Väntestrategi mellan omförsök

Exempel på LLM-pipeline

Ett vanligt användningsfall är att kedja ihop anrop till LLM API med databehandling:

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

Starta pipelinen:

PHP

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

Om anropet till OpenAI får tidsgränsöverskridande försöker arbetaren bara det steget igen. Prompten som byggdes i steg 0 är redan säkert sparad i arbetsflödets tillstånd.