Aller au contenu

Flux de travail

Afficher au format Markdown Modifier sur GitHub Documentation de 2cc7b2b

Les flux de travail décomposent les opérations de longue durée en étapes distinctes. Chaque étape enregistre sa sortie dans un état partagé en base de données. Si PHP s’arrête en cours d’étape, le travailleur relance cette étape en conservant tout l’état antérieur.

Que sont les flux de travail ?

Un flux de travail est une séquence ordonnée d’étapes. Chaque étape est une classe PHP qui reçoit l’état cumulé de toutes les étapes précédentes et renvoie de nouvelles données à fusionner dans l’état. Le flux de travail avance étape par étape, chaque transition étant encapsulée dans une transaction atomique MySQL.

PHP
use Queuety\Queuety;

$workflow_id = Queuety::workflow( 'generate_report' )
    ->then( FetchDataHandler::class )
    ->then( CallLLMHandler::class )
    ->then( FormatOutputHandler::class )
    ->dispatch( [ 'user_id' => 42 ] );

Accumulation de l’état

La charge utile initiale transmise à dispatch() devient l’état de départ. Les clés renvoyées par chaque étape sont fusionnées dans l’état. Les clés commençant par _ sont réservées aux valeurs de contrôle telles que _next_step et ne sont pas fusionnées dans l’état public :

Code
dispatch(['user_id' => 42])

Step 0: FetchDataHandler   -> returns ['user_name' => 'Alice']
  state: {user_id: 42, user_name: 'Alice'}

Step 1: CallLLMHandler     -> returns ['llm_response' => '...']
  state: {user_id: 42, user_name: 'Alice', llm_response: '...'}

Step 2: FormatOutputHandler -> returns ['report_url' => '/reports/42.pdf']
  state: {user_id: 42, user_name: 'Alice', llm_response: '...', report_url: '/reports/42.pdf'}

Chaque étape reçoit l’état cumulé complet. Cela permet de transmettre facilement des données entre les étapes sans stockage externe.

Reprise

Les flux de travail sont durables, car leur état est enregistré dans la base de données après chaque étape. Si le travailleur plante pendant l’étape 1, le flux de travail redémarre à l’étape 1 avec la sortie de l’étape 0 déjà présente dans l’état.

La limite entre les étapes est une transaction unique MySQL qui effectue de façon atomique :

  1. La mise à jour de l’état du flux avec la sortie de l’étape
  2. Le marquage comme terminée de la tâche de l’étape actuelle
  3. La mise en file d’attente de la tâche de l’étape suivante

Cela garantit qu’aucun état n’est perdu, même en cas de plantage, de dépassement de délai ou d’erreurs fatales PHP.

Cycle de vie du flux de travail

Code
dispatch() -> running -> step 0 -> step 1 -> ... -> step N -> completed
                  |                   |
                  |                   +-> failed -> retry -> running (from failed step)
                  |
                  +-> paused -> resumed -> running
                  |
                  +-> waiting_for_signal -> signal received -> running
                  |
                  +-> waiting_for_workflows -> dependency settles -> running
                  |
                  +-> cancelled (via cancel_workflow)
Statut Description
running Étapes en cours de traitement
completed Toutes les étapes se sont terminées avec succès
failed Une étape a échoué après épuisement des tentatives
paused Suspendu manuellement. Le travail en cours peut se terminer, mais les étapes suivantes ne sont pas mises en file d’attente avant la reprise
waiting_for_signal En attente d’un signal externe via Queuety::signal()
waiting_for_workflows En attente de la fin d’un ou plusieurs autres flux de travail via wait_for_workflow() ou wait_for_workflows()
cancelled Annulé explicitement via Queuety::cancel_workflow(). Les tâches en attente sont mises en retrait et un gestionnaire de nettoyage facultatif s’exécute.

Gérer les flux de travail

Vérifier le statut :

PHP
$state = Queuety::workflow_status( $workflow_id );
echo $state->status->value;   // 'running'
echo $state->current_step;    // 1
echo $state->total_steps;     // 3
print_r( $state->waiting_for ); // ['42'] while waiting on another workflow
print_r( $state->wait_details ); // matched/remaining blockers, result key, correlation info

Suspendre et reprendre :

PHP
Queuety::pause_workflow( $workflow_id );
Queuety::resume_workflow( $workflow_id );

Relancer un flux de travail en échec depuis l’étape en échec :

PHP
Queuety::retry_workflow( $workflow_id );

Annuler un flux de travail et exécuter les gestionnaires de nettoyage :

PHP
Queuety::cancel_workflow( $workflow_id );

Consultez Annulation pour en savoir plus sur les gestionnaires de nettoyage et ce qui se passe lorsqu’un flux de travail est annulé.

Options du flux de travail

Définissez la file d’attente, la priorité et le comportement des relances pour toutes les étapes d’un flux de travail :

PHP
use Queuety\Enums\Priority;

Queuety::workflow( 'important_pipeline' )
    ->on_queue( 'pipelines' )
    ->with_priority( Priority::High )
    ->max_attempts( 5 )
    ->on_cancel( PipelineCleanupHandler::class )
    ->prune_state_after( 3 )
    ->then( StepA::class )
    ->then( StepB::class )
    ->dispatch( $payload );

La méthode on_cancel() enregistre un gestionnaire de nettoyage exécuté lors de l’annulation du flux de travail. Consultez Annulation.

La méthode prune_state_after() active la suppression automatique des anciennes sorties d’étape pour limiter la taille de l’état du flux de travail. Consultez Élagage de l’état.

Utilisez compensate_with() pour associer un gestionnaire de compensation à chaque étape. Utilisez compensate_on_failure() pour exécuter automatiquement les compensations des étapes terminées après un échec du flux de travail, et pas uniquement en cas d’annulation explicite. Consultez Compensation.

Utilisez les garde-fous des flux de travail pour sécuriser l’orchestration de longue durée : version() associe une version au niveau de l’application à une définition, Queuety stocke aussi un hachage déterministe de la définition pour chaque exécution, idempotency_key() rend la répartition durable en cas de relances ou de requêtes en double, et max_transitions(), max_for_each_items() et max_state_bytes() interrompent les exécutions incontrôlées avant qu’elles ne dérapent. Consultez Garde-fous.

Types d’étapes

Queuety prend en charge onze types d’étapes dans un flux de travail :

Modèles courants

Si vous créez des systèmes autonomes ou semi-autonomes, commencez par Orchestration d’agents. Cette page montre comment for_each(), start_agents(), wait_for_agents(), wait_for_workflow() et wait_for_decision() s’articulent dans des flux concrets de planification/exécution et de validation humaine.