Workflows

Workflows opdeler langvarigt arbejde i afgrænsede trin. Hvert trin gemmer sit output i en fælles tilstandsbeholder i databasen. Hvis PHP dør midt i et trin, prøver workerprocessen trinnet igen med al tidligere tilstand intakt.

Hvad er workflows?

Et workflow er en ordnet sekvens af trin. Hvert trin er en klasse af typen PHP, der modtager den samlede tilstand fra alle foregående trin og returnerer nye data, som flettes ind i tilstanden. Workflowet går frem ét trin ad gangen, og hver overgang er omsluttet af en atomisk transaktion af typen MySQL.

use Queuety\Queuety;

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

Akkumulering af tilstand

Den oprindelige nyttelast, du sender til dispatch(), bliver starttilstanden. Nøglerne, som hvert trin returnerer, flettes ind i tilstanden. Nøgler, der begynder med _, er reserveret til kontrolværdier som _next_step og flettes ikke ind i den offentlige tilstand:

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

Hvert trin ser hele den akkumulerede tilstand. Det gør det nemt at overføre data mellem trin uden ekstern lagring.

Mulighed for genoptagelse

Workflows er holdbare, fordi tilstanden gemmes i databasen efter hvert trin. Hvis workerprocessen går ned under trin 1, starter workflowet igen fra trin 1, mens outputtet fra trin 0 allerede er i tilstanden.

Grænsen mellem trinene er en enkelt transaktion af typen MySQL, der atomisk:

1. Opdaterer workflowtilstanden med trinnets output
2. Markerer jobbet for det aktuelle trin som fuldført
3. Sætter jobbet for næste trin i kø

Det sikrer, at ingen tilstand går tabt, heller ikke ved nedbrud, timeout eller fatale fejl af typen PHP.

Workflowets livscyklus

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)

Administration af workflows

Kontrollér 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

Sæt på pause, og genoptag:

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

Prøv et fejlet workflow igen fra det fejlede trin:

Queuety::retry_workflow( $workflow_id );

Annullér et workflow, og kør oprydningsbehandlere:

Queuety::cancel_workflow( $workflow_id );

Se Annullering for detaljer om oprydningsbehandlere, og hvad der sker, når et workflow annulleres.

Workflowindstillinger

Angiv kø, prioritet og politik for nye forsøg for alle trin i et 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 );

Metoden on_cancel() registrerer en oprydningsbehandler, der køres, når workflowet annulleres. Se Annullering.

Metoden prune_state_after() aktiverer automatisk fjernelse af gamle trinoutputs for at holde workflowtilstanden begrænset. Se Tilstandsbeskæring.

Brug compensate_with() til at knytte en kompenseringsbehandler til et enkelt trin. Brug compensate_on_failure(), når kompenseringer for fuldførte trin automatisk skal køres efter en workflowfejl, ikke kun ved eksplicit annullering. Se Kompensering.

Brug workflowværn, når du har brug for sikrere orkestrering over længere tid: version() knytter en applikationsversion til en definition, Queuety gemmer også en deterministisk definitionens hash for hver kørsel, idempotency_key() gør afsendelse holdbar på tværs af nye forsøg eller dublerede anmodninger, og max_transitions(), max_for_each_items() og max_state_bytes() får løbske kørsler til at fejle, før de løber løbsk. Se Værn.

Trintyper

Queuety understøtter elleve typer trin i et workflow:

• Sekventielle kæder med ->then() til udførelse ét ad gangen
• Parallelle grupper med ->parallel() til samtidig udførelse
• Dynamisk for-each med ->for_each() til forgreningsarbejde, der opdages under kørslen, og eksplicitte regler for fuldførelse
• Asynkrone overleveringer med ->start_workflows() eller ->start_agents() til at sende uafhængige underordnede workflows afsted for elementer, der opdages under kørslen
• Betinget forgrening med navngivne trin og _next_step
• Gentagelser med ->repeat_until() og ->repeat_while() til holdbare tilbagekanter til tidligere navngivne trin
• Kør workflows med ->run_workflow() til indlejret sammensætning
• Holdbare forsinkelser med ->delay() til tidsbaserede forsinkelser
• Signaler med ->wait_for_signal(), ->wait_for_signals(), ->wait_for_approval(), ->wait_for_input() og ->wait_for_decision() til eksterne hændelsesudløsere
• Workflowafhængigheder med ->wait_for_workflow(), ->wait_for_workflows() og ->wait_for_agents() til koordinering på tværs af workflows
• Streamende trin med grænsefladen StreamingStep til holdbart chunkbaseret output

Almindelige mønstre

Hvis du bygger autonome eller delvist autonome systemer, så start med Agentorkestrering. Her vises, hvordan for_each(), start_agents(), wait_for_agents(), wait_for_workflow() og wait_for_decision() fungerer sammen i faktiske planner/executor- og menneskelige evalueringsforløb.