Preskočiť na obsah

Workflowy

Zobraziť ako Markdown Upraviť na GitHub Dokumentácia na 2cc7b2b

Workflowy rozdeľujú dlhotrvajúcu prácu na jednotlivé kroky. Výstup každého kroku sa ukladá do zdieľaného úložiska stavu v databáze. Ak PHP zlyhá uprostred kroku, pracovník tento krok zopakuje so zachovaným celým predchádzajúcim stavom.

Čo sú workflowy?

Workflow je usporiadaný sled krokov. Každý krok je trieda PHP, ktorá prijíma nahromadený stav zo všetkých predchádzajúcich krokov a vracia nové údaje na zlúčenie do stavu. Workflow postupuje po jednom kroku, pričom každý prechod je zabalený do atómovej transakcie MySQL.

PHP
use Queuety\Queuety;

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

Zhromažďovanie stavu

Počiatočné údaje, ktoré odovzdáte do dispatch(), sa stanú počiatočným stavom. Kľúče vrátené jednotlivými krokmi sa zlúčia do stavu. Kľúče začínajúce na _ sú vyhradené pre riadiace hodnoty, ako je _next_step, a do verejného stavu sa nezlučujú:

Kód
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'}

Každý krok vidí celý nahromadený stav. Vďaka tomu možno jednoducho odovzdávať údaje medzi krokmi bez externého úložiska.

Možnosť obnovenia

Workflowy sú odolné, pretože stav sa po každom kroku ukladá do databázy. Ak pracovník počas kroku 1 zlyhá, workflow sa reštartuje od kroku 1 a výstup kroku 0 už bude v stave.

Hranica kroku tvorí jedinú transakciu MySQL, ktorá atómovo:

  1. Aktualizuje stav workflowu výstupom kroku
  2. Označí úlohu aktuálneho kroku ako dokončenú
  3. Zaradí úlohu nasledujúceho kroku do frontu

To zaručuje, že sa nikdy nestratí žiadny stav, ani v prípade zlyhania, vypršania časového limitu alebo závažných chýb PHP.

Životný cyklus workflowu

Kód
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)
Stav Popis
running Aktívne spracúva kroky
completed Všetky kroky sa úspešne dokončili
failed Krok zlyhal po vyčerpaní opakovaných pokusov
paused Manuálne pozastavený. Aktuálna rozpracovaná práca sa môže dokončiť, no ďalšie kroky sa nezaradia do frontu, kým workflow neobnovíte
waiting_for_signal Čaká na externý signál cez Queuety::signal()
waiting_for_workflows Čaká na dokončenie jedného alebo viacerých iných workflowov cez wait_for_workflow() alebo wait_for_workflows()
cancelled Výslovne zrušený cez Queuety::cancel_workflow(). Čakajúce úlohy sa presunú do archívu a spustí sa voliteľná obslužná rutina na vyčistenie.

Správa workflowov

Kontrola stavu:

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

Pozastavenie a obnovenie:

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

Opätovné spustenie neúspešného workflowu od kroku, ktorý zlyhal:

PHP
Queuety::retry_workflow( $workflow_id );

Zrušenie workflowu a spustenie obslužných rutín na vyčistenie:

PHP
Queuety::cancel_workflow( $workflow_id );

Podrobnosti o obslužných rutinách na vyčistenie a o tom, čo sa stane pri zrušení workflowu, nájdete v časti Zrušenie.

Možnosti workflowu

Nastavte front, prioritu a správanie pri opakovaní pre všetky kroky workflowu:

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

Metóda on_cancel() zaregistruje obslužnú rutinu na vyčistenie, ktorá sa spustí pri zrušení workflowu. Pozrite si časť Zrušenie.

Metóda prune_state_after() zapne automatické odstraňovanie starých výstupov krokov, aby sa obmedzila veľkosť stavu workflowu. Pozrite si časť Orezávanie stavu.

Pomocou compensate_with() priraďte obslužnú rutinu kompenzácie ku každému kroku. Použite compensate_on_failure(), ak chcete, aby sa kompenzácie dokončených krokov automaticky vykonali po zlyhaní workflowu, nielen pri jeho výslovnom zrušení. Pozrite si časť Kompenzácia.

Ak potrebujete bezpečnejšiu orchestráciu s dlhým trvaním, použite ochranné mechanizmy workflowov: version() označí definíciu verziou na úrovni aplikácie, Queuety tiež uloží deterministický hash definície pre každý beh, idempotency_key() zabezpečí trvalé spustenie pri opakovaných pokusoch alebo duplicitných požiadavkách a max_transitions(), max_for_each_items() a max_state_bytes() ukončia nekontrolované behy skôr, než sa vymknú spod kontroly. Pozrite si Ochranné mechanizmy.

Typy krokov

Queuety podporuje v rámci workflowu jedenásť typov krokov:

Bežné vzory

Ak vytvárate autonómne alebo poloautonómne systémy, začnite témou Orchestrácia agentov. Ukazuje, ako do seba zapadajú for_each(), start_agents(), wait_for_agents(), wait_for_workflow() a wait_for_decision() v reálnych procesoch plánovania a vykonávania aj pri kontrole človekom.