Työnkulut

Työnkulut jakavat pitkäkestoisen työn erillisiin vaiheisiin. Kunkin vaiheen tulos tallennetaan tietokantaan yhteiseen tilasäilöön. Jos PHP pysähtyy kesken vaiheen, työntekijä yrittää vaihetta uudelleen kaikki aiempi tila säilyttäen.

Mitä työnkulut ovat?

Työnkulku on järjestetty vaiheiden sarja. Kukin vaihe on PHP-luokka, joka vastaanottaa kaikkien edellisten vaiheiden kertyneen tilan ja palauttaa uusia tietoja yhdistettäväksi tilaan. Työnkulku etenee vaihe kerrallaan, ja jokainen siirtymä suoritetaan atomisessa MySQL-transaktiossa.

use Queuety\Queuety;

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

Tilan kerryttäminen

Toiminnolle dispatch() välittämäsi alkukuorma muodostaa alkutilan. Kunkin vaiheen palauttamat avaimet yhdistetään tilaan. Merkillä _ alkavat avaimet on varattu ohjausarvoille, kuten _next_step, eikä niitä yhdistetä julkiseen tilaan:

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

Jokainen vaihe näkee koko kertyneen tilan. Näin tietojen välittäminen vaiheiden välillä onnistuu helposti ilman ulkoista tallennustilaa.

Jatkaminen

Työnkulut ovat pysyviä, koska tila tallennetaan tietokantaan jokaisen vaiheen jälkeen. Jos työntekijä kaatuu vaiheen 1 aikana, työnkulku käynnistyy uudelleen vaiheesta 1, ja vaiheen 0 tulos on jo tilassa.

Vaiheraja on yksittäinen MySQL-transaktio, joka tekee atomisesti seuraavat asiat:

1. Päivittää työnkulun tilan vaiheen tuloksella
2. Merkitsee nykyisen vaiheen työn valmiiksi
3. Lisää seuraavan vaiheen työn jonoon

Tämä varmistaa, ettei tilaa koskaan menetetä edes kaatumisten, aikakatkaisujen tai PHP-vakavien virheiden yhteydessä.

Työnkulun elinkaari

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)

Työnkulkujen hallinta

Tarkista tila:

$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

Keskeytä ja jatka:

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

Yritä epäonnistunutta työnkulkua uudelleen sen epäonnistuneesta vaiheesta:

Queuety::retry_workflow( $workflow_id );

Peruuta työnkulku ja suorita siivouskäsittelijät:

Queuety::cancel_workflow( $workflow_id );

Katso kohdasta Peruutus lisätietoja siivouskäsittelijöistä ja työnkulun peruuttamisen seurauksista.

Työnkulun asetukset

Aseta kaikkien työnkulun vaiheiden jono, prioriteetti ja uudelleenyritysten toiminta:

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

on_cancel()-menetelmä rekisteröi siivouskäsittelijän, joka suoritetaan työnkulun peruuttamisen yhteydessä. Katso Peruutus.

prune_state_after()-menetelmä ottaa käyttöön vanhojen vaihetulosten automaattisen poiston, jotta työnkulun tila pysyy rajattuna. Katso Tilan karsiminen.

Käytä toimintoa compensate_with() liittääksesi vaiheeseen korvauskäsittelijän. Käytä toimintoa compensate_on_failure(), kun haluat suorittaa valmistuneiden vaiheiden korvaukset automaattisesti työnkulun epäonnistumisen jälkeen, pelkän nimenomaisen peruutuksen sijaan. Katso Korvaustoimet.

Käytä työnkulun suojauksia, kun tarvitset turvallisempaa pitkäkestoista orkestrointia: version() merkitsee määrityksen sovellustason versiolla, Queuety tallentaa lisäksi deterministisen määritystiivisteen jokaiselle suoritukselle, idempotency_key() varmistaa lähetyksen kestävyyden uudelleenyrityksissä ja kaksoispyynnöissä, ja max_transitions(), max_for_each_items() sekä max_state_bytes() pysäyttävät hallitsemattomasti kasvavat suoritukset ennen kuin ne karkaavat käsistä. Katso Suojaukset.

Vaihetyypit

Queuety tukee työnkulussa yhdentoista tyyppisiä vaiheita:

• Peräkkäiset ketjut, joissa käytetään toimintoa ->then() vaiheiden suorittamiseen yksi kerrallaan
• Rinnakkaiset ryhmät, joissa käytetään toimintoa ->parallel() samanaikaiseen suorittamiseen
• Dynaaminen for-each, jossa käytetään toimintoa ->for_each() ajonaikaisesti löydettyyn haaratyöhön ja nimenomaisiin valmistumisehtoihin
• Asynkroniset siirrot, joissa käytetään toimintoa ->start_workflows() tai ->start_agents() itsenäisten alityönkulkujen käynnistämiseen ajonaikaisesti löydetyistä kohteista
• Ehdollinen haarautuminen, jossa käytetään nimettyjä vaiheita ja toimintoa _next_step
• Toistot, joissa käytetään toimintoja ->repeat_until() ja ->repeat_while() pysyviin paluusiirtymiin aiempiin nimettyihin vaiheisiin
• Työnkulkujen suorittaminen, jossa käytetään toimintoa ->run_workflow() sisäkkäiseen koostamiseen
• Pysyvät viiveet, joissa käytetään toimintoa ->delay() aikaperusteisiin viiveisiin
• Signaalit, joissa käytetään toimintoja ->wait_for_signal(), ->wait_for_signals(), ->wait_for_approval(), ->wait_for_input() ja ->wait_for_decision() ulkoisten tapahtumien käynnistämiseen
• Työnkulkujen riippuvuudet, joissa käytetään toimintoja ->wait_for_workflow(), ->wait_for_workflows() ja ->wait_for_agents() työnkulkujen väliseen koordinointiin
• Suoratoistovaiheet, joissa käytetään StreamingStep-rajapintaa pysyvään lohkopohjaiseen tulosteeseen

Yleisiä käyttötapoja

Jos rakennat autonomisia tai osittain autonomisia järjestelmiä, aloita kohdasta Agenttien orkestrointi. Siinä näytetään, miten for_each(), start_agents(), wait_for_agents(), wait_for_workflow() ja wait_for_decision() toimivat yhdessä suunnittelija/suorittaja- ja ihmisen tarkastuksen työnkuluissa.