Arbetsflöden

Arbetsflöden delar upp långvarigt arbete i avgränsade steg. Varje steg sparar sina utdata i en gemensam tillståndspåse i databasen. Om PHP dör mitt i ett steg försöker arbetaren med steget igen, med allt tidigare tillstånd intakt.

Vad är arbetsflöden?

Ett arbetsflöde är en ordnad sekvens av steg. Varje steg är en klass av typen PHP som tar emot det ackumulerade tillståndet från alla tidigare steg och returnerar nya data som sammanfogas med tillståndet. Arbetsflödet går vidare ett steg i taget, och varje övergång omsluts av en atomisk MySQL-transaktion.

use Queuety\Queuety;

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

Ackumulering av tillstånd

Den ursprungliga nyttolast du skickar till dispatch() blir starttillståndet. Returnerade nycklar från varje steg sammanfogas med tillståndet. Nycklar som börjar med _ är reserverade för kontrollvärden som _next_step och sammanfogas inte med det publika tillståndet:

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

Varje steg ser hela det ackumulerade tillståndet. Det gör det enkelt att skicka data mellan steg utan extern lagring.

Återupptagning

Arbetsflöden är beständiga eftersom tillståndet sparas i databasen efter varje steg. Om arbetaren kraschar under steg 1 startas arbetsflödet om från steg 1, med utdata från steg 0 redan i tillståndet.

Steggränsen är en enda MySQL-transaktion som atomiskt:

1. Uppdaterar arbetsflödets tillstånd med stegets utdata
2. Markerar jobbet för det aktuella steget som slutfört
3. Köar jobbet för nästa steg

Detta garanterar att inget tillstånd går förlorat, även vid krascher, tidsgränser eller fatala PHP-fel.

Arbetsflödets livscykel

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)

Hantera arbetsflöden

Kontrollera status:

$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

Pausa och återuppta:

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

Försök köra ett misslyckat arbetsflöde igen från steget där det misslyckades:

Queuety::retry_workflow( $workflow_id );

Avbryt ett arbetsflöde och kör rensningshanterare:

Queuety::cancel_workflow( $workflow_id );

Läs Avbrytning för information om rensningshanterare och vad som händer när ett arbetsflöde avbryts.

Alternativ för arbetsflöden

Ange kö, prioritet och omförsöksbeteende för alla steg i ett arbetsflöde:

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

Metoden on_cancel() registrerar en rensningshanterare som körs när arbetsflödet avbryts. Se Avbrytning.

Metoden prune_state_after() aktiverar automatisk borttagning av gamla stegresultat för att hålla arbetsflödets tillstånd begränsat. Se Gallring av tillstånd.

Använd compensate_with() för att koppla en kompensationshanterare till ett enskilt steg. Använd compensate_on_failure() när du vill att kompensationer för slutförda steg ska köras automatiskt efter ett arbetsflödesfel, inte bara vid uttrycklig avbrytning. Se Kompensation.

Använd skyddsräcken för arbetsflöden när du behöver säkrare långvarig orkestrering: version() förser en definition med en versionsetikett på applikationsnivå, Queuety lagrar också en deterministisk hash av definitionen för varje körning, idempotency_key() gör dispatch beständig vid omförsök eller duplicerade begäranden, och max_transitions(), max_for_each_items() och max_state_bytes() avbryter skenande körningar innan de drar iväg. Se Skyddsräcken.

Stegtyper

Queuety stöder elva stegtyper i ett arbetsflöde:

• Sekventiella kedjor med ->then() för körning ett steg i taget
• Parallella grupper med ->parallel() för samtidig körning
• Dynamisk for-each med ->for_each() för grenar som upptäcks vid körning och uttryckliga slutförandevillkor
• Asynkrona överlämningar med ->start_workflows() eller ->start_agents() för att skicka fristående underordnade arbetsflöden utifrån objekt som upptäcks vid körning
• Villkorlig förgrening med namngivna steg och _next_step
• Upprepningar med ->repeat_until() och ->repeat_while() för beständiga bakåtlänkar till tidigare namngivna steg
• Körning av arbetsflöden med ->run_workflow() för nästlad sammansättning
• Beständiga fördröjningar med ->delay() för tidsbaserade fördröjningar
• Signaler med ->wait_for_signal(), ->wait_for_signals(), ->wait_for_approval(), ->wait_for_input() och ->wait_for_decision() för externa händelseutlösare
• Arbetsflödesberoenden med ->wait_for_workflow(), ->wait_for_workflows() och ->wait_for_agents() för samordning mellan arbetsflöden
• Strömningssteg med gränssnittet StreamingStep för beständig utdata baserad på datasegment

Vanliga mönster

Om du bygger autonoma eller delvis autonoma system kan du börja med Agentorkestrering. Där visas hur for_each(), start_agents(), wait_for_agents(), wait_for_workflow() och wait_for_decision() samverkar i verkliga flöden med planerare och exekverare samt mänsklig granskning.