Workflow

Workflow rozdělují dlouhotrvající práci na samostatné kroky. Výstup každého kroku se ukládá do společné sady stavových dat v databázi. Pokud PHP během kroku přestane fungovat, worker krok zopakuje se zachováním veškerého předchozího stavu.

Co jsou workflow?

Workflow je uspořádaná posloupnost kroků. Každý krok je třída PHP, která přijímá nashromážděný stav ze všech předchozích kroků a vrací nová data ke sloučení se stavem. Workflow postupuje po jednom kroku a každý přechod probíhá v atomické transakci MySQL.

use Queuety\Queuety;

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

Shromažďování stavu

Počáteční data, která předáte do dispatch(), se stanou výchozím stavem. Vrácené klíče z každého kroku se sloučí se stavem. Klíče začínající na _ jsou vyhrazené pro řídicí hodnoty, například _next_step, a do veřejného stavu se neslučují:

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

Každý krok vidí celý nashromážděný stav. Díky tomu lze snadno předávat data mezi kroky bez externího úložiště.

Možnost obnovení

Workflow jsou odolná, protože se stav po každém kroku ukládá do databáze. Pokud worker během kroku 1 spadne, workflow se restartuje od kroku 1 a výstup kroku 0 už má ve stavu uložený.

Hranici kroku tvoří jediná transakce MySQL, která atomicky:

1. Aktualizuje stav workflow výstupem kroku
2. Označí úlohu aktuálního kroku jako dokončenou
3. Zařadí do fronty úlohu dalšího kroku

Tím je zajištěno, že se žádný stav neztratí ani při pádech, vypršení časového limitu nebo fatálních chybách PHP.

Životní cyklus 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)

Správa workflow

Zkontrolujte stav:

$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

Pozastavení a obnovení:

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

Opakování neúspěšného workflow od kroku, který selhal:

Queuety::retry_workflow( $workflow_id );

Zrušení workflow a spuštění obslužných funkcí pro úklid:

Queuety::cancel_workflow( $workflow_id );

Podrobnosti o obslužných funkcích pro úklid a o tom, co se stane při zrušení workflow, najdete v části Zrušení.

Možnosti workflow

Nastavení fronty, priority a chování při opakování pro všechny kroky 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 );

Metoda on_cancel() registruje obslužnou funkci pro úklid, která se spustí při zrušení workflow. Viz Zrušení.

Metoda prune_state_after() zapne automatické odebírání starých výstupů kroků, aby velikost stavu workflow zůstala omezená. Viz Pročišťování stavu.

Pomocí compensate_with() připojíte obslužnou funkci pro kompenzaci ke konkrétnímu kroku. Použijte compensate_on_failure(), chcete-li kompenzace dokončených kroků spouštět automaticky po selhání workflow, nejen při výslovném zrušení. Viz Kompenzace.

Ochranné prvky workflow použijte, když potřebujete bezpečnější dlouhotrvající orchestraci: version() označí definici verzí na úrovni aplikace, Queuety navíc ukládá pro každý běh deterministický hash definice, idempotency_key() zajišťuje odolnost spouštění při opakovaných pokusech nebo duplicitních požadavcích a max_transitions(), max_for_each_items() a max_state_bytes() ukončí neřízené běhy chybou dřív, než se vymknou kontrole. Viz Ochranné prvky.

Typy kroků

Queuety podporuje v rámci workflow jedenáct typů kroků:

• Sekvenční řetězce používají ->then() pro postupné spouštění
• Paralelní skupiny používají ->parallel() pro souběžné spouštění
• Dynamické iterace používají ->for_each() pro práci ve větvích zjištěnou za běhu a výslovné potvrzení dokončení
• Asynchronní předávání používají ->start_workflows() nebo ->start_agents() ke spuštění nezávislých podřízených workflow z položek zjištěných za běhu
• Podmíněné větvení používá pojmenované kroky a _next_step
• Opakování používá ->repeat_until() a ->repeat_while() pro trvalé návraty k dřívějším pojmenovaným krokům
• Spouštění workflow používá ->run_workflow() pro vnořené skládání
• Trvalé prodlevy používají ->delay() pro časové prodlevy
• Signály používají ->wait_for_signal(), ->wait_for_signals(), ->wait_for_approval(), ->wait_for_input() a ->wait_for_decision() ke spouštění externími událostmi
• Závislosti workflow používají ->wait_for_workflow(), ->wait_for_workflows() a ->wait_for_agents() ke koordinaci mezi workflow
• Kroky s průběžným výstupem používají rozhraní StreamingStep pro trvalý výstup po částech

Běžné vzory

Pokud vytváříte autonomní nebo částečně autonomní systémy, začněte částí Orchestrace agentů. Ukazuje, jak for_each(), start_agents(), wait_for_agents(), wait_for_workflow() a wait_for_decision() spolupracují v reálných postupech planner/executor a při lidské kontrole.