Töövood

Töövood jagavad pikaajalised tööd eraldi sammudeks. Iga samm salvestab oma väljundi andmebaasis ühisesse olekukogumisse. Kui PHP sureb sammu keskel, proovib töötaja seda sammu uuesti, säilitades kogu varasema oleku.

Mis on töövood?

Töövoog on sammude järjestatud jada. Iga samm on PHP klass, mis saab kõigi eelmiste sammude koondoleku ja tagastab olekusse liidetavad uued andmed. Töövoog liigub edasi üks samm korraga ning iga üleminek toimub atomaarse MySQL-tehingu sees.

use Queuety\Queuety;

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

Oleku kogumine

Algne koormus, mille edastate käsule dispatch(), muutub algolekuks. Iga sammu tagastatud võtmed liidetakse olekusse. Tähega _ algavad võtmed on reserveeritud juhtväärtuste jaoks, näiteks _next_step, ning neid avalikku olekusse ei liideta:

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

Iga samm näeb kogu kogunenud olekut. Nii on lihtne andmeid sammude vahel edastada ilma välise salvestuseta.

Jätkatavus

Töövood on püsivad, sest nende olek salvestatakse pärast iga sammu andmebaasi. Kui töötaja jookseb 1. sammu ajal kokku, alustatakse töövoogu uuesti 1. sammust, kusjuures 0. sammu väljund on juba olekus olemas.

Sammupiir on üks MySQL-tehing, mis teeb aatomiliselt järgmist:

1. Uuendab töövoo olekut sammu väljundiga
2. Märgib praeguse sammu töö lõpetatuks
3. Lisab järgmise sammu töö järjekorda

See tagab, et olek ei lähe kunagi kaotsi, isegi kokkujooksmiste, aegumiste või PHP kriitiliste tõrgete korral.

Töövoo elutsükkel

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)

Töövoogude haldamine

Oleku kontrollimine:

$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

Peatamine ja jätkamine:

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

Nurjunud töövoo uuesti proovimine nurjunud sammust:

Queuety::retry_workflow( $workflow_id );

Töövoo tühistamine ja puhastuskäitlejate käivitamine:

Queuety::cancel_workflow( $workflow_id );

Lisateavet puhastuskäitlejate ja töövoo tühistamisel toimuva kohta leiate lehelt Tühistamine.

Töövoo valikud

Kõigile töövoo sammudele järjekorra, prioriteedi ja korduskatsete käitumise määramine:

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

Meetod on_cancel() registreerib puhastuskäitleja, mis käivitub töövoo tühistamisel. Vt Tühistamine.

Meetod prune_state_after() võimaldab vanade sammu väljundite automaatset eemaldamist, et töövoo olek püsiks piiratud suurusega. Vt Olekukärpimine.

Kasutage compensate_with() iga sammu jaoks kompenseerimiskäitleja määramiseks. Kasutage compensate_on_failure(), kui soovite, et lõpetatud sammude kompenseerimised käivituksid pärast töövoo nurjumist automaatselt, mitte ainult selgesõnalise tühistamise korral. Vt Kompenseerimine.

Kasutage töövoo kaitsepiirdeid, kui vajate pikaajalise orkestreerimise jaoks suuremat turvalisust: version() lisab definitsioonile rakenduse tasemel versiooni, Queuety salvestab iga käivituse kohta ka deterministliku definitsiooniräsi, idempotency_key() muudab käivitamise korduskatsete ja duplikaatpäringute korral püsivaks ning max_transitions(), max_for_each_items() ja max_state_bytes() nurjavad kontrolli alt väljuvad käivitused enne, kui need spiraalselt kasvavad. Vt Kaitsepiirded.

Sammude tüübid

Queuety toetab töövoos ühtteist tüüpi samme:

• Järjestikused ahelad, kus ükshaaval täitmiseks kasutatakse ->then()
• Paralleelsed rühmad, kus samaaegseks täitmiseks kasutatakse ->parallel()
• Dünaamiline igaühe jaoks, kus käitusajal tuvastatud harutöö ja selgesõnalise lõpetamisloogika jaoks kasutatakse ->for_each()
• Asünkroonsed üleandmised, kus käitusajal tuvastatud üksuste põhjal sõltumatute alamvoogude käivitamiseks kasutatakse ->start_workflows() või ->start_agents()
• Tingimuslik hargnemine, kus kasutatakse nimega samme ja _next_step
• Kordused, kus varasemate nimega sammude juurde püsivate tagasihüpete jaoks kasutatakse ->repeat_until() ja ->repeat_while()
• Töövoogude käivitamine, kus pesastatud koostamiseks kasutatakse ->run_workflow()
• Püsivad viivitused, kus ajaliste viivituste jaoks kasutatakse ->delay()
• Signaalid, kus väliste sündmuste käivitamiseks kasutatakse ->wait_for_signal(), ->wait_for_signals(), ->wait_for_approval(), ->wait_for_input() ja ->wait_for_decision()
• Töövoosõltuvused, kus töövoogudevaheliseks koordineerimiseks kasutatakse ->wait_for_workflow(), ->wait_for_workflows() ja ->wait_for_agents()
• Voogedastussammud, kus tükipõhise väljundi püsivaks edastamiseks kasutatakse liidest StreamingStep

Levinud kasutusmustrid

Kui loote autonoomseid või poolautonoomseid süsteeme, alustage lehelt Agendi orkestreerimine. Sealt näete, kuidas for_each(), start_agents(), wait_for_agents(), wait_for_workflow() ja wait_for_decision() sobituvad päris planeerija ja täitja ning inimese ülevaatuse töövoogudesse.