Fluxuri de lucru

Fluxurile de lucru împart activitățile de lungă durată în pași distincți. Fiecare pas își salvează rezultatul într-o structură comună de stare din baza de date. Dacă PHP se oprește la mijlocul unui pas, lucrătorul reîncearcă acel pas cu toată starea anterioară intactă.

Ce sunt fluxurile de lucru?

Un flux de lucru este o secvență ordonată de pași. Fiecare pas este o clasă PHP care primește starea acumulată de la toți pașii anteriori și returnează date noi de îmbinat cu starea. Fluxul de lucru avansează câte un pas, iar fiecare tranziție este încadrată într-o tranzacție atomică MySQL.

use Queuety\Queuety;

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

Acumularea stării

Payloadul inițial transmis către dispatch() devine starea de pornire. Cheile returnate de fiecare pas sunt îmbinate cu starea. Cheile care încep cu _ sunt rezervate valorilor de control, precum _next_step, și nu sunt îmbinate în starea publică:

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

Fiecare pas vede întreaga stare acumulată. Astfel, este ușor să transmiteți date între pași fără stocare externă.

Posibilitatea de reluare

Fluxurile de lucru sunt durabile deoarece starea este salvată în baza de date după fiecare pas. Dacă lucrătorul se blochează în timpul pasului 1, fluxul de lucru repornește de la pasul 1, cu rezultatul pasului 0 deja inclus în stare.

Limita dintre pași este o singură tranzacție MySQL care, în mod atomic:

1. Actualizează starea fluxului de lucru cu rezultatul pasului
2. Marchează ca finalizată sarcina pasului curent
3. Adaugă în coadă sarcina pasului următor

Astfel se garantează că starea nu se pierde niciodată, nici în caz de blocări, expirări sau erori fatale PHP.

Ciclul de viață al fluxului de lucru

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)

Gestionarea fluxurilor de lucru

Verificați starea:

$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

Întrerupeți și reluați:

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

Reîncercați un flux de lucru eșuat începând cu pasul la care a eșuat:

Queuety::retry_workflow( $workflow_id );

Anulați un flux de lucru și executați gestionarele de curățare:

Queuety::cancel_workflow( $workflow_id );

Consultați Anularea pentru detalii despre gestionarele de curățare și ce se întâmplă când un flux de lucru este anulat.

Opțiunile fluxului de lucru

Setați coada, prioritatea și comportamentul reîncercărilor pentru toți pașii dintr-un flux de lucru:

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() înregistrează un gestionar de curățare care rulează la anularea fluxului de lucru. Consultați Anularea.

Metoda prune_state_after() activează eliminarea automată a rezultatelor vechi ale pașilor, pentru a menține starea fluxului de lucru în limite. Consultați Eliminarea stării.

Folosiți compensate_with() pentru a atașa un gestionar de compensare pentru fiecare pas. Folosiți compensate_on_failure() când doriți ca compensările pașilor finalizați să ruleze automat după eșecul fluxului de lucru, nu doar la anularea explicită. Consultați Compensarea.

Folosiți măsurile de protecție ale fluxului de lucru pentru o orchestrare de lungă durată mai sigură: version() etichetează o definiție cu o versiune la nivelul aplicației, Queuety stochează și un hash determinist al definiției pentru fiecare execuție, idempotency_key() face expedierea durabilă în caz de reîncercări sau solicitări duplicate, iar max_transitions(), max_for_each_items() și max_state_bytes() opresc execuțiile scăpate de sub control înainte să se amplifice. Consultați Măsuri de protecție.

Tipuri de pași

Queuety acceptă unsprezece tipuri de pași într-un flux de lucru:

• Lanțuri secvențiale folosind ->then() pentru executare pe rând
• Grupuri paralele folosind ->parallel() pentru executare simultană
• For-Each dinamic folosind ->for_each() pentru lucrul pe ramuri descoperite în timpul execuției și reguli explicite de finalizare
• Transferuri asincrone folosind ->start_workflows() sau ->start_agents() pentru a expedia fluxuri de lucru copil independente pornind de la elemente descoperite în timpul execuției
• Ramificare condiționată folosind pași numiți și _next_step
• Repetări folosind ->repeat_until() și ->repeat_while() pentru legături de întoarcere durabile către pași numiți anterior
• Executarea fluxurilor de lucru folosind ->run_workflow() pentru compunere imbricată
• Întârzieri durabile folosind ->delay() pentru întârzieri bazate pe timp
• Semnale folosind ->wait_for_signal(), ->wait_for_signals(), ->wait_for_approval(), ->wait_for_input() și ->wait_for_decision() pentru declanșatoare de evenimente externe
• Dependențe între fluxuri de lucru folosind ->wait_for_workflow(), ->wait_for_workflows() și ->wait_for_agents() pentru coordonare între fluxuri
• Pași de redare în flux folosind interfața StreamingStep pentru rezultate durabile bazate pe fragmente

Tipare uzuale

Dacă dezvoltați sisteme autonome sau semi-autonome, începeți cu Orchestrarea agenților. Prezintă modul în care for_each(), start_agents(), wait_for_agents(), wait_for_workflow() și wait_for_decision() se combină în fluxuri reale de planificare/executare și de verificare umană.