Delovni tokovi

Delovni tokovi razdelijo dolgotrajno delo na posamezne korake. Izhod vsakega koraka se shrani v skupno zbirko stanja v podatkovni zbirki. Če PHP umre sredi koraka, delavec korak ponovi z ohranjenim celotnim prejšnjim stanjem.

Kaj so delovni tokovi?

Delovni tok je urejeno zaporedje korakov. Vsak korak je razred PHP, ki prejme zbrano stanje vseh prejšnjih korakov in vrne nove podatke za združitev s stanjem. Delovni tok napreduje po en korak naenkrat, vsak prehod pa je ovit v atomsko transakcijo MySQL.

use Queuety\Queuety;

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

Kopičenje stanja

Začetni podatki, ki jih posredujete dispatch(), postanejo začetno stanje. Vrnjeni ključi vsakega koraka se združijo s stanjem. Ključi, ki se začnejo z _, so rezervirani za nadzorne vrednosti, kot je _next_step, in se ne združijo z javnim stanjem:

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

Vsak korak vidi celotno zbrano stanje. Tako lahko preprosto posredujete podatke med koraki brez zunanje shrambe.

Možnost nadaljevanja

Delovni tokovi so zanesljivi, ker se stanje po vsakem koraku shrani v podatkovno zbirko. Če se delavec sesuje med 1. korakom, se delovni tok znova začne pri 1. koraku, pri čemer je izhod 0. koraka že v stanju.

Meja koraka je ena transakcija MySQL, ki atomsko:

1. Posodobi stanje delovnega toka z izhodom koraka
2. Označi opravilo trenutnega koraka kot dokončano
3. Uvrsti opravilo naslednjega koraka v čakalno vrsto

To zagotavlja, da se stanje nikoli ne izgubi, tudi ob sesutjih, časovnih omejitvah ali usodnih napakah PHP.

Življenjski cikel delovnega toka

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)

Upravljanje delovnih tokov

Preverjanje stanja:

$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

Začasna zaustavitev in nadaljevanje:

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

Ponovni poskus neuspešnega delovnega toka od koraka, pri katerem je prišlo do napake:

Queuety::retry_workflow( $workflow_id );

Preklic delovnega toka in zagon obdelovalnikov za čiščenje:

Queuety::cancel_workflow( $workflow_id );

Podrobnosti o obdelovalnikih za čiščenje in dogajanju ob preklicu delovnega toka najdete v razdelku Preklic.

Možnosti delovnega toka

Nastavite čakalno vrsto, prednost in vedenje pri ponovnih poskusih za vse korake delovnega toka:

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

Metoda on_cancel() registrira obdelovalnik za čiščenje, ki se izvede ob preklicu delovnega toka. Glejte razdelek Preklic.

Metoda prune_state_after() omogoči samodejno odstranjevanje starih izhodov korakov, da omeji velikost stanja delovnega toka. Glejte razdelek Obrezovanje stanja.

Uporabite compensate_with() za pripenjanje obdelovalnika za kompenzacijo posameznemu koraku. Uporabite compensate_on_failure(), kadar želite, da se kompenzacije dokončanih korakov po napaki delovnega toka izvedejo samodejno in ne le ob izrecnem preklicu. Glejte razdelek Kompenzacija.

Varovalke delovnega toka uporabite, kadar potrebujete varnejšo dolgotrajno orkestracijo: version() definiciji doda različico na ravni aplikacije, Queuety za vsako izvajanje shrani tudi deterministično zgoščeno vrednost definicije, idempotency_key() zagotavlja trajnost odpreme ob ponovnih poskusih ali podvojenih zahtevah, max_transitions(), max_for_each_items() in max_state_bytes() pa ustavijo nenadzorovano izvajanje, preden uide izpod nadzora. Glejte razdelek Varovalke.

Vrste korakov

Queuety podpira enajst vrst korakov v delovnem toku:

• Zaporedne verige z uporabo ->then() za izvajanje po enega koraka naenkrat
• Vzporedne skupine z uporabo ->parallel() za sočasno izvajanje
• Dinamični for-each z uporabo ->for_each() za delo v vejah, odkritih med izvajanjem, in izrecno določeno dokončanje
• Asinhrone predaje z uporabo ->start_workflows() ali ->start_agents() za odpošiljanje neodvisnih podrejenih delovnih tokov iz elementov, odkritih med izvajanjem
• Pogojno razvejanje z uporabo poimenovanih korakov in _next_step
• Ponovitve z uporabo ->repeat_until() in ->repeat_while() za trajne povratne povezave na prejšnje poimenovane korake
• Izvajanje delovnih tokov z uporabo ->run_workflow() za gnezdeno sestavljanje
• Trajni zamiki z uporabo ->delay() za časovne zamike
• Signali z uporabo ->wait_for_signal(), ->wait_for_signals(), ->wait_for_approval(), ->wait_for_input() in ->wait_for_decision() za zunanje sprožilce dogodkov
• Odvisnosti delovnih tokov z uporabo ->wait_for_workflow(), ->wait_for_workflows() in ->wait_for_agents() za usklajevanje med delovnimi tokovi
• Koraki pretakanja z uporabo vmesnika StreamingStep za trajni izhod po kosih

Pogosti vzorci

Če gradite avtonomne ali delno avtonomne sisteme, začnite z razdelkom Orkestracija agentov. Pokaže, kako se for_each(), start_agents(), wait_for_agents(), wait_for_workflow() in wait_for_decision() povezujejo v dejanskih potekih načrtovalnika in izvajalnika ter pregledov, ki jih opravi človek.