Workflows

Workflows splitsen langlopende taken op in afzonderlijke stappen. Elke stap slaat de uitvoer op in een gedeelde statusverzameling in de database. Als PHP halverwege een stap uitvalt, probeert de worker die stap opnieuw met behoud van alle eerdere status.

Wat zijn workflows?

Een workflow is een geordende reeks stappen. Elke stap is een klasse van PHP die de verzamelde status van alle voorgaande stappen ontvangt en nieuwe gegevens retourneert die met de status worden samengevoegd. De workflow doorloopt stap voor stap, waarbij elke overgang is verpakt in een atomaire MySQL-transactie.

use Queuety\Queuety;

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

Status verzamelen

De initiële payload die u aan dispatch() doorgeeft, wordt de beginstatus. De geretourneerde sleutels van elke stap worden met de status samengevoegd. Sleutels die beginnen met _ zijn gereserveerd voor besturingswaarden zoals _next_step en worden niet samengevoegd met de openbare status:

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

Elke stap ziet de volledige verzamelde status. Zo kunt u eenvoudig gegevens tussen stappen doorgeven zonder externe opslag.

Hervatbaarheid

Workflows zijn duurzaam omdat de status na elke stap in de database wordt opgeslagen. Als de worker tijdens stap 1 crasht, wordt de workflow hervat vanaf stap 1, met de uitvoer van stap 0 al in de status.

De stapgrens is één MySQL-transactie die atomair:

1. de workflowstatus bijwerkt met de uitvoer van de stap
2. de taak van de huidige stap markeert als voltooid
3. de taak van de volgende stap in de wachtrij plaatst

Zo raakt er nooit status verloren, zelfs niet bij crashes, time-outs of fatale fouten van PHP.

Levenscyclus van een workflow

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)

Workflows beheren

Status controleren:

$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

Pauzeren en hervatten:

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

Een mislukte workflow opnieuw proberen vanaf de mislukte stap:

Queuety::retry_workflow( $workflow_id );

Een workflow annuleren en opruimhandlers uitvoeren:

Queuety::cancel_workflow( $workflow_id );

Zie Annulering voor meer informatie over opruimhandlers en wat er gebeurt wanneer een workflow wordt geannuleerd.

Workflowopties

Stel de wachtrij, prioriteit en het gedrag bij nieuwe pogingen in voor alle stappen in een workflow:

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

De methode on_cancel() registreert een opruimhandler die wordt uitgevoerd wanneer de workflow wordt geannuleerd. Zie Annulering.

De methode prune_state_after() schakelt automatische verwijdering van oude stapuitvoer in, zodat de workflowstatus begrensd blijft. Zie Status opschonen.

Gebruik compensate_with() om een compensatiehandler aan een afzonderlijke stap te koppelen. Gebruik compensate_on_failure() als u wilt dat compensaties van voltooide stappen automatisch worden uitgevoerd na een workflowfout, en niet alleen bij expliciete annulering. Zie Compensatie.

Gebruik workflowbeveiligingen voor veiligere langlopende orkestratie: version() voegt een applicatieversie toe aan een definitie, Queuety slaat ook een deterministische definitiehash op voor elke uitvoering, idempotency_key() zorgt dat dispatches bestand zijn tegen nieuwe pogingen of dubbele aanvragen, en max_transitions(), max_for_each_items() en max_state_bytes() beëindigen ontspoorde uitvoeringen voordat ze uit de hand lopen. Zie Beveiligingen.

Staptypen

Queuety ondersteunt elf typen stappen binnen een workflow:

• Sequentiële ketens met ->then() voor uitvoering één voor één
• Parallelle groepen met ->parallel() voor gelijktijdige uitvoering
• Dynamische for-each met ->for_each() voor vertakkingswerk dat tijdens uitvoering wordt ontdekt en expliciete voltooiingssemantiek
• Asynchrone overdrachten met ->start_workflows() of ->start_agents() om vanuit tijdens uitvoering ontdekte items onafhankelijke onderliggende workflows te starten
• Voorwaardelijke vertakkingen met benoemde stappen en _next_step
• Herhalingen met ->repeat_until() en ->repeat_while() voor duurzame terugkoppelingen naar eerder benoemde stappen
• Workflows uitvoeren met ->run_workflow() voor geneste samenstelling
• Duurzame vertragingen met ->delay() voor vertragingen op basis van tijd
• Signalen met ->wait_for_signal(), ->wait_for_signals(), ->wait_for_approval(), ->wait_for_input() en ->wait_for_decision() voor externe gebeurtenistriggers
• Workflowafhankelijkheden met ->wait_for_workflow(), ->wait_for_workflows() en ->wait_for_agents() voor coördinatie tussen workflows
• Streamingstappen met de interface StreamingStep voor duurzame uitvoer in chunks

Veelgebruikte patronen

Als u autonome of semi-autonome systemen bouwt, begin dan met Agentorkestratie. Hier ziet u hoe for_each(), start_agents(), wait_for_agents(), wait_for_workflow() en wait_for_decision() samenwerken in echte planner-/uitvoerder- en menselijk-beoordelingsprocessen.