Flussi di lavoro

I flussi di lavoro suddividono le attività di lunga durata in passaggi distinti. L'output di ogni passaggio viene salvato in un contenitore di stato condiviso nel database. Se PHP si arresta durante un passaggio, il worker riprova quel passaggio mantenendo intatto tutto lo stato precedente.

Cosa sono i flussi di lavoro?

Un flusso di lavoro è una sequenza ordinata di passaggi. Ogni passaggio è una classe PHP che riceve lo stato accumulato da tutti i passaggi precedenti e restituisce nuovi dati da unire allo stato. Il flusso di lavoro avanza un passaggio alla volta, racchiudendo ogni transizione in una transazione atomica MySQL.

use Queuety\Queuety;

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

Accumulo dello stato

Il payload iniziale passato a dispatch() diventa lo stato di partenza. Le chiavi restituite da ogni passaggio vengono unite allo stato. Le chiavi che iniziano con _ sono riservate a valori di controllo come _next_step e non vengono unite allo stato pubblico:

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'}

Ogni passaggio riceve l'intero stato accumulato. In questo modo è facile trasferire dati tra i passaggi senza ricorrere a sistemi di archiviazione esterni.

Ripresa

I flussi di lavoro sono durevoli perché lo stato viene salvato nel database dopo ogni passaggio. Se il worker si arresta durante il passaggio 1, il flusso di lavoro riparte dal passaggio 1, con l'output del passaggio 0 già presente nello stato.

Il confine del passaggio è una singola transazione MySQL che esegue atomicamente:

1. Aggiorna lo stato del flusso di lavoro con l'output del passaggio
2. Segna come completato il job del passaggio corrente
3. Accoda il job del passaggio successivo

Questo garantisce che lo stato non vada mai perso, anche in caso di arresti anomali, timeout o errori fatali di PHP.

Ciclo di vita del flusso di lavoro

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)

Gestione dei flussi di lavoro

Controllare lo stato:

$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

Sospendere e riprendere:

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

Riprovare un flusso di lavoro non riuscito a partire dal passaggio in errore:

Queuety::retry_workflow( $workflow_id );

Annullare un flusso di lavoro ed eseguire i gestori di pulizia:

Queuety::cancel_workflow( $workflow_id );

Per informazioni sui gestori di pulizia e su cosa accade quando un flusso di lavoro viene annullato, consultate Annullamento.

Opzioni del flusso di lavoro

Impostare la coda, la priorità e il comportamento dei tentativi per tutti i passaggi di un flusso di lavoro:

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

Il metodo on_cancel() registra un gestore di pulizia che viene eseguito quando il flusso di lavoro viene annullato. Consultate Annullamento.

Il metodo prune_state_after() abilita la rimozione automatica dei vecchi output dei passaggi per mantenere limitate le dimensioni dello stato del flusso di lavoro. Consultate Pulizia dello stato.

Usate compensate_with() per associare un gestore di compensazione a ogni passaggio. Usate compensate_on_failure() quando volete che le compensazioni dei passaggi completati vengano eseguite automaticamente dopo un errore del flusso di lavoro, e non solo in caso di annullamento esplicito. Consultate Compensazione.

Usate le protezioni dei flussi di lavoro per orchestrare processi di lunga durata in modo più sicuro: version() assegna una versione a livello applicativo alla definizione, Queuety memorizza anche un hash deterministico della definizione per ogni esecuzione, idempotency_key() rende l'invio durevole in caso di tentativi o richieste duplicate, mentre max_transitions(), max_for_each_items() e max_state_bytes() bloccano le esecuzioni fuori controllo prima che degenerino. Consultate Protezioni.

Tipi di passaggio

Queuety supporta undici tipi di passaggio all'interno di un flusso di lavoro:

• Catene sequenziali tramite ->then() per l'esecuzione un passaggio alla volta
• Gruppi paralleli tramite ->parallel() per l'esecuzione simultanea
• Cicli for-each dinamici tramite ->for_each() per le diramazioni di lavoro individuate durante l'esecuzione e una semantica di completamento esplicita
• Passaggi asincroni tramite ->start_workflows() o ->start_agents() per avviare flussi di lavoro figlio indipendenti a partire da elementi individuati durante l'esecuzione
• Diramazioni condizionali tramite passaggi con nome e _next_step
• Ripetizioni tramite ->repeat_until() e ->repeat_while() per tornare in modo durevole a passaggi precedenti con nome
• Esecuzione di flussi di lavoro tramite ->run_workflow() per la composizione annidata
• Ritardi durevoli tramite ->delay() per i ritardi basati sul tempo
• Segnali tramite ->wait_for_signal(), ->wait_for_signals(), ->wait_for_approval(), ->wait_for_input() e ->wait_for_decision() per attivare eventi esterni
• Dipendenze tra flussi di lavoro tramite ->wait_for_workflow(), ->wait_for_workflows() e ->wait_for_agents() per il coordinamento tra flussi di lavoro
• Passaggi con output in streaming tramite l'interfaccia StreamingStep per produrre output durevoli suddivisi in blocchi

Schemi comuni

Se state creando sistemi autonomi o semi-autonomi, iniziate da Orchestrazione di agenti. Spiega come for_each(), start_agents(), wait_for_agents(), wait_for_workflow() e wait_for_decision() si combinano nei flussi reali di pianificazione/esecuzione e revisione umana.