Darbplūsmas

Darbplūsmas sadala ilgstošu darbu atsevišķās darbībās. Katras darbības izvade tiek saglabāta kopīgā stāvokļa objektā datubāzē. Ja PHP pārstāj darboties darbības vidū, darbinieks mēģina šo darbību izpildīt vēlreiz, saglabājot visu iepriekšējo stāvokli.

Kas ir darbplūsmas?

Darbplūsma ir sakārtota darbību secība. Katra darbība ir PHP klase, kas saņem visu iepriekšējo darbību uzkrāto stāvokli un atgriež jaunus datus, ko sapludināt ar šo stāvokli. Darbplūsma virzās uz priekšu pa vienai darbībai, un katra pāreja ir ietverta atomārā MySQL transakcijā.

use Queuety\Queuety;

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

Stāvokļa uzkrāšana

Sākotnējā lietderīgā slodze, ko nododat dispatch(), kļūst par sākuma stāvokli. Katras darbības atgrieztās atslēgas tiek sapludinātas ar šo stāvokli. Atslēgas, kas sākas ar _, ir rezervētas vadības vērtībām, piemēram, _next_step, un netiek sapludinātas ar publisko stāvokli:

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

Katra darbība redz visu uzkrāto stāvokli. Tādējādi datus starp darbībām var viegli nodot bez ārējas krātuves.

Izpildes atsākšana

Darbplūsmas ir noturīgas, jo pēc katras darbības stāvoklis tiek saglabāts datubāzē. Ja darbinieks avarē 1. darbības laikā, darbplūsma tiek atsākta no 1. darbības, un 0. darbības izvade jau ir iekļauta stāvoklī.

Darbības robežu veido viena MySQL transakcija, kas atomāri:

1. Atjaunina darbplūsmas stāvokli ar darbības izvadi
2. Atzīmē pašreizējās darbības darbu kā pabeigtu
3. Ievieto rindā nākamās darbības darbu

Tas nodrošina, ka stāvoklis nekad netiek zaudēts, pat avāriju, noildzes vai PHP fatālu kļūdu gadījumā.

Darbplūsmas dzīves cikls

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)

Darbplūsmu pārvaldība

Pārbaudīt statusu:

$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

Apturēšana un atsākšana:

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

Atkārtoti mēģināt izpildīt neizdevušos darbplūsmu, sākot ar darbību, kurā radās kļūme:

Queuety::retry_workflow( $workflow_id );

Atcelt darbplūsmu un izpildīt tīrīšanas apstrādātājus:

Queuety::cancel_workflow( $workflow_id );

Informāciju par tīrīšanas apstrādātājiem un darbplūsmas atcelšanas sekām skatiet sadaļā Atcelšana.

Darbplūsmas opcijas

Iestatiet rindu, prioritāti un atkārtotas mēģināšanas darbību visām darbplūsmas darbībām:

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

Metode on_cancel() reģistrē tīrīšanas apstrādātāju, kas tiek palaists, kad darbplūsma tiek atcelta. Skatiet sadaļu Atcelšana.

Metode prune_state_after() ieslēdz veco darbību izvadu automātisku dzēšanu, lai ierobežotu darbplūsmas stāvokļa apjomu. Skatiet sadaļu Stāvokļa attīrīšana.

Izmantojiet compensate_with(), lai pievienotu katrai darbībai kompensācijas apstrādātāju. Izmantojiet compensate_on_failure(), ja vēlaties, lai pabeigto darbību kompensācijas tiktu izpildītas automātiski pēc darbplūsmas kļūmes, nevis tikai skaidras atcelšanas gadījumā. Skatiet sadaļu Kompensācija.

Izmantojiet darbplūsmas aizsargmehānismus, lai drošāk veiktu ilgstošu orkestrāciju: version() pievieno definīcijai lietotnes līmeņa versiju, Queuety katrai izpildei saglabā arī deterministisku definīcijas jaucējvērtību, idempotency_key() nodrošina izpildes nosūtīšanas noturību pēc atkārtotiem mēģinājumiem vai dublētiem pieprasījumiem, bet max_transitions(), max_for_each_items() un max_state_bytes() aptur nekontrolētu izpildi, pirms tā kļūst pārmērīga. Skatiet sadaļu Aizsargmehānismi.

Darbību veidi

Queuety darbplūsmās atbalsta vienpadsmit darbību veidus:

• Secīgas ķēdes, izmantojot ->then() izpildei pa vienai darbībai
• Paralēlās grupas, izmantojot ->parallel() vienlaicīgai izpildei
• Dinamiska izpilde katram vienumam, izmantojot ->for_each() izpildlaikā noteiktu atzaru darbu un skaidru pabeigšanas semantiku
• Asinhrona nodošana, izmantojot ->start_workflows() vai ->start_agents(), lai no izpildlaikā noteiktiem vienumiem nosūtītu neatkarīgas pakārtotās darbplūsmas
• Nosacījuma atzarojums, izmantojot nosauktas darbības un _next_step
• Atkārtojumi, izmantojot ->repeat_until() un ->repeat_while(), lai noturīgi atgrieztos pie iepriekšējām nosauktajām darbībām
• Darbplūsmu izpilde, izmantojot ->run_workflow() ligzdotai kompozīcijai
• Noturīgas aizkaves, izmantojot ->delay() laika aizkavēm
• Signāli, izmantojot ->wait_for_signal(), ->wait_for_signals(), ->wait_for_approval(), ->wait_for_input() un ->wait_for_decision() ārēju notikumu aktivizēšanai
• Darbplūsmu atkarības, izmantojot ->wait_for_workflow(), ->wait_for_workflows() un ->wait_for_agents() darbplūsmu savstarpējai koordinācijai
• Straumēšanas darbības, izmantojot saskarni StreamingStep, lai noturīgi izvadītu datus pa daļām

Biežākie lietošanas veidi

Ja veidojat autonomas vai daļēji autonomas sistēmas, sāciet ar sadaļu Aģentu orkestrācija. Tajā parādīts, kā for_each(), start_agents(), wait_for_agents(), wait_for_workflow() un wait_for_decision() darbojas kopā reālās plānotāja un izpildītāja, kā arī cilvēka pārskatīšanas darbplūsmās.