Darbo eigos

Darbo eigos suskaido ilgai trunkantį darbą į atskirus žingsnius. Kiekvieno žingsnio išvestis išsaugoma duomenų bazėje, bendroje būsenos struktūroje. Jei PHP nutrūksta žingsnio viduryje, darbuotojas pakartoja šį žingsnį, išsaugodamas visą ankstesnę būseną.

Kas yra darbo eigos?

Darbo eiga yra nuosekli žingsnių seka. Kiekvienas žingsnis yra PHP klasė, kuriai perduodama visų ankstesnių žingsnių sukaupta būsena ir kuri grąžina naujus duomenis, sujungiamus su būsena. Darbo eiga juda po vieną žingsnį, o kiekvienas perėjimas vykdomas atominėje MySQL transakcijoje.

use Queuety\Queuety;

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

Būsenos kaupimas

Pradinė naudingoji apkrova, kurią perduodate dispatch(), tampa pradine būsena. Kiekvieno žingsnio grąžinami raktai sujungiami su būsena. Raktais, prasidedančiais _, rezervuojamos valdymo reikšmės, pavyzdžiui, _next_step; jie nesujungiami su viešąja būsena:

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

Kiekvienas žingsnis mato visą sukauptą būseną. Taip lengva perduoti duomenis tarp žingsnių nenaudojant išorinės saugyklos.

Tęsimas

Darbo eigos yra patikimos, nes po kiekvieno žingsnio būsena išsaugoma duomenų bazėje. Jei darbuotojas sugenda vykdydamas 1 žingsnį, darbo eiga paleidžiama iš naujo nuo 1 žingsnio, o 0 žingsnio išvestis jau būna įrašyta būsenoje.

Žingsnio riba yra viena MySQL transakcija, kuri atominiu būdu:

1. Atnaujina darbo eigos būseną žingsnio išvestimi
2. Pažymi dabartinio žingsnio užduotį kaip užbaigtą
3. Įtraukia kito žingsnio užduotį į eilę

Taip užtikrinama, kad būsena niekada nebūtų prarasta net įvykus gedimams, pasibaigus skirtajam laikui ar ištikus PHP kritinėms klaidoms.

Darbo eigos gyvavimo ciklas

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)

Darbo eigų valdymas

Patikrinkite būseną:

$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

Pristabdykite ir tęskite:

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

Pakartokite nepavykusią darbo eigą nuo nepavykusio žingsnio:

Queuety::retry_workflow( $workflow_id );

Atšaukite darbo eigą ir paleiskite valymo apdorojimo funkcijas:

Queuety::cancel_workflow( $workflow_id );

Išsamesnės informacijos apie valymo apdorojimo funkcijas ir tai, kas nutinka atšaukus darbo eigą, žr. Atšaukimas.

Darbo eigos parinktys

Nustatykite visų darbo eigos žingsnių eilę, prioritetą ir pakartotinių bandymų elgseną:

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

Metodas on_cancel() užregistruoja valymo apdorojimo funkciją, kuri paleidžiama atšaukus darbo eigą. Žr. Atšaukimas.

Metodas prune_state_after() įjungia automatinį senų žingsnių išvesčių šalinimą, kad darbo eigos būsena išliktų riboto dydžio. Žr. Būsenos apkarpymas.

Naudokite compensate_with(), kad pridėtumėte kiekvieno žingsnio kompensavimo apdorojimo funkciją. Naudokite compensate_on_failure(), jei norite, kad po darbo eigos nesėkmės automatiškai būtų vykdomos užbaigtų žingsnių kompensavimo funkcijos, o ne tik aiškiai atšaukus darbo eigą. Žr. Kompensavimas.

Naudokite darbo eigos apsaugos priemones, kai reikia saugiau vykdyti ilgai trunkančią orkestraciją: version() pažymi apibrėžtį programos lygmens versija, Queuety taip pat išsaugo kiekvieno vykdymo deterministinę apibrėžties maišą, idempotency_key() užtikrina patikimą išsiuntimą vykdyti pakartotinių bandymų arba pasikartojančių užklausų atveju, o max_transitions(), max_for_each_items() ir max_state_bytes() sustabdo nekontroliuojamus vykdymus dar prieš jiems įsibėgėjant. Žr. Apsaugos priemonės.

Žingsnių tipai

Queuety palaiko vienuolika darbo eigos žingsnių tipų:

• Nuoseklios grandinės naudojant ->then(), kad veiksmai būtų vykdomi po vieną
• Lygiagrečios grupės naudojant ->parallel(), kad veiksmai būtų vykdomi vienu metu
• Dinaminis vykdymas kiekvienam elementui naudojant ->for_each(), skirtą vykdymo metu aptinkamoms šakų užduotims ir aiškiai užbaigimo semantikai
• Asinchroninis perdavimas naudojant ->start_workflows() arba ->start_agents(), kad iš vykdymo metu aptiktų elementų būtų paleistos nepriklausomos antrinės darbo sekos
• Sąlyginis šakojimas naudojant pavadintus veiksmus ir _next_step
• Kartojimai naudojant ->repeat_until() ir ->repeat_while(), kad būtų galima patikimai grįžti prie ankstesnių pavadintų veiksmų
• Darbo sekų paleidimas naudojant ->run_workflow() įdėtinei kompozicijai
• Patikimos delsos naudojant ->delay() laiko delsoms
• Signalai naudojant ->wait_for_signal(), ->wait_for_signals(), ->wait_for_approval(), ->wait_for_input() ir ->wait_for_decision() išorinių įvykių trigeriams
• Darbo sekų priklausomybės naudojant ->wait_for_workflow(), ->wait_for_workflows() ir ->wait_for_agents() skirtingų darbo sekų koordinavimui
• Srautiniai veiksmai naudojant StreamingStep sąsają, skirtą patikimai išvesti duomenis dalimis

Dažni šablonai

Jei kuriate autonomines arba iš dalies autonomines sistemas, pradėkite nuo Agentų orkestravimas. Čia parodoma, kaip for_each(), start_agents(), wait_for_agents(), wait_for_workflow() ir wait_for_decision() dera tarpusavyje realiose planavimo ir vykdymo bei žmogaus peržiūros darbo sekose.