Cadenas secuenciales

Ver como Markdown Editar en GitHub Documentación de 2cc7b2b

El patrón de flujo de trabajo más común es una cadena lineal de pasos. Cada paso se ejecuta cuando termina el anterior, que le pasa el estado acumulado.

El método `->then()`

Use ->then() para añadir pasos secuenciales a un flujo de trabajo:

PHP

use Queuety\Queuety;

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

Los pasos se ejecutan en el orden en que se añaden. El flujo de trabajo solo avanza cuando el paso actual se completa correctamente.

La interfaz Step

Los controladores de pasos del flujo de trabajo implementan la interfaz 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,
        ];
    }
}

Diferencias clave con la interfaz Handler:

	`Handler` (tareas)	`Step` (flujos de trabajo)
Firma del método	`handle(array $payload): void`	`handle(array $state): array`
Entrada	Carga útil de la tarea	Estado acumulado del flujo de trabajo
Salida	Ninguna (void)	Datos que se combinarán con el estado

Flujo de estado

Cada paso recibe el estado acumulado completo y devuelve una matriz de nuevos pares clave-valor. Los datos devueltos se combinan con el estado para los pasos siguientes:

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

Configuración de pasos

El método config() admite las siguientes claves:

Clave	Tipo	Valor predeterminado	Descripción
`max_attempts`	`int`	`3`	Número máximo de intentos de reintento para este paso
`backoff`	`string`	`exponential`	Estrategia de espera entre reintentos

Ejemplo de una canalización de LLM

Un caso de uso habitual es encadenar llamadas de API con procesamiento de datos:

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

Inicie la canalización:

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 se agota el tiempo de espera de la llamada OpenAI, el trabajador solo vuelve a intentar ese paso. La instrucción creada en el paso 0 ya está guardada de forma segura en el estado del flujo de trabajo.