Sekventielle kæder

Vis som Markdown Rediger på GitHub Dokumentation hos 2cc7b2b

Det mest almindelige mønster for arbejdsgange er en lineær kæde af trin. Hvert trin køres, når det foregående er afsluttet, og den akkumulerede tilstand sendes videre.

Metoden `->then()`

Brug ->then() til at føje sekventielle trin til en arbejdsgang:

PHP

use Queuety\Queuety;

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

Trinene køres i den rækkefølge, de tilføjes. Arbejdsgangen går kun videre, når det aktuelle trin afsluttes uden fejl.

Step-grænsefladen

Handlere for arbejdsgangstrin implementerer grænsefladen 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,
        ];
    }
}

Vigtige forskelle fra grænsefladen Handler:

	`Handler` (jobs)	`Step` (arbejdsgange)
Metodesignatur	`handle(array $payload): void`	`handle(array $state): array`
Input	Jobnyttelast	Akkumuleret arbejdsgangstilstand
Output	Intet (void)	Data, der flettes ind i tilstanden

Tilstandsflow

Hvert trin modtager hele den akkumulerede tilstand og returnerer en række nye nøgle-værdi-par. De returnerede data flettes ind i tilstanden for de efterfølgende trin:

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

Konfiguration af trin

Metoden config() understøtter følgende nøgler:

Nøgle	Type	Standard	Beskrivelse
`max_attempts`	`int`	`3`	Maksimalt antal genforsøg for dette trin
`backoff`	`string`	`exponential`	Ventestrategi for genforsøg

Eksempel på LLM-pipeline

Et almindeligt anvendelsestilfælde er at kæde LLM-API-kald sammen 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 [];
    }
}

Send 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 ] );

Hvis kaldet til OpenAI får timeout, prøver workeren kun dette trin igen. Prompten, der blev opbygget i trin 0, er allerede gemt sikkert i arbejdsgangens tilstand.