Przepływy pracy

Wyświetl jako Markdown Edytuj w GitHub Dokumentacja: 2cc7b2b

Przepływy pracy dzielą długotrwałe zadania na odrębne kroki. Wynik każdego kroku jest zapisywany we wspólnym zbiorze stanu w bazie danych. Jeśli PHP przestanie działać w trakcie kroku, pracownik ponowi ten krok, zachowując cały wcześniejszy stan.

Czym są przepływy pracy?

Przepływ pracy to uporządkowana sekwencja kroków. Każdy krok jest klasą PHP, która otrzymuje stan zgromadzony we wszystkich poprzednich krokach i zwraca nowe dane do scalenia ze stanem. Przepływ przechodzi przez kolejne kroki, a każde przejście jest objęte atomową transakcją MySQL.

PHP

use Queuety\Queuety;

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

Gromadzenie stanu

Początkowy ładunek przekazany do dispatch() staje się stanem początkowym. Klucze zwrócone przez każdy krok są scalane ze stanem. Klucze zaczynające się od _ są zarezerwowane na wartości sterujące, takie jak _next_step, i nie są scalane ze stanem publicznym:

Kod

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żdy krok otrzymuje pełny zgromadzony stan. Ułatwia to przekazywanie danych między krokami bez korzystania z zewnętrznego magazynu.

Wznawianie

Przepływy pracy są trwałe, ponieważ stan jest zapisywany w bazie danych po każdym kroku. Jeśli pracownik ulegnie awarii podczas kroku 1, przepływ uruchamia się ponownie od kroku 1, a wynik kroku 0 jest już dostępny w stanie.

Granica kroku to pojedyncza transakcja MySQL, która atomowo:

Aktualizuje stan przepływu pracy, zapisując wynik kroku
Oznacza zadanie bieżącego kroku jako ukończone
Dodaje do kolejki zadanie następnego kroku

Gwarantuje to, że stan nigdy nie zostanie utracony, nawet w przypadku awarii, przekroczenia limitu czasu lub krytycznych błędów PHP.

Cykl życia przepływu pracy

Kod

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)

Status	Opis
`running`	Aktywnie przetwarza kroki
`completed`	Wszystkie kroki zakończone pomyślnie
`failed`	Krok zakończył się niepowodzeniem po wyczerpaniu ponownych prób
`paused`	Ręcznie wstrzymany. Bieżąca praca w toku może się zakończyć, ale kolejne kroki nie są dodawane do kolejki do czasu wznowienia
`waiting_for_signal`	Oczekuje na sygnał zewnętrzny przez `Queuety::signal()`
`waiting_for_workflows`	Oczekuje na ukończenie co najmniej jednego innego przepływu pracy przez `wait_for_workflow()` lub `wait_for_workflows()`
`cancelled`	Jawnie anulowany przez `Queuety::cancel_workflow()`. Oczekujące zadania są ukrywane, a opcjonalna procedura czyszczenia jest uruchamiana.

Zarządzanie przepływami pracy

Sprawdzanie statusu:

PHP

$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

Wstrzymywanie i wznawianie:

PHP

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

Ponawianie nieudanego przepływu od kroku, który zakończył się niepowodzeniem:

PHP

Queuety::retry_workflow( $workflow_id );

Anulowanie przepływu pracy i uruchamianie procedur czyszczenia:

PHP

Queuety::cancel_workflow( $workflow_id );

Szczegółowe informacje o procedurach czyszczenia i skutkach anulowania przepływu pracy znajdziesz w sekcji Anulowanie.

Opcje przepływu pracy

Ustaw kolejkę, priorytet i reguły ponawiania prób dla wszystkich kroków w przepływie pracy:

PHP

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() rejestruje procedurę czyszczenia uruchamianą po anulowaniu przepływu pracy. Zobacz sekcję Anulowanie.

Metoda prune_state_after() włącza automatyczne usuwanie starych wyników kroków, aby ograniczyć rozmiar stanu przepływu pracy. Zobacz sekcję Przycinanie stanu.

Użyj compensate_with(), aby przypisać procedurę kompensacyjną do każdego kroku. Użyj compensate_on_failure(), jeśli chcesz, aby procedury kompensacyjne ukończonych kroków uruchamiały się automatycznie po niepowodzeniu przepływu pracy, a nie tylko po jego jawnym anulowaniu. Zobacz sekcję Kompensacja.

Używaj zabezpieczeń przepływu pracy, aby bezpieczniej zarządzać długotrwałą orkiestracją: version() oznacza definicję wersją na poziomie aplikacji, Queuety zapisuje też deterministyczny skrót definicji dla każdego uruchomienia, idempotency_key() zapewnia trwałość przekazywania po ponowieniu lub powtórnym żądaniu, a max_transitions(), max_for_each_items() i max_state_bytes() przerywają niekontrolowane uruchomienia, zanim wymkną się spod kontroli. Zobacz sekcję Zabezpieczenia.

Typy kroków

Queuety obsługuje jedenaście typów kroków w ramach przepływu pracy:

Łańcuchy sekwencyjne z użyciem ->then() do wykonywania krok po kroku
Grupy równoległe z użyciem ->parallel() do wykonywania równoczesnego
Dynamiczne For-Each z użyciem ->for_each() do obsługi gałęzi wykrywanych w czasie działania i jawnego określania warunków ukończenia
Przekazania asynchroniczne z użyciem ->start_workflows() lub ->start_agents() do uruchamiania niezależnych przepływów podrzędnych dla elementów wykrytych w czasie działania
Rozgałęzianie warunkowe z użyciem nazwanych kroków i _next_step
Powtórzenia z użyciem ->repeat_until() i ->repeat_while() do trwałego powracania do wcześniejszych nazwanych kroków
Uruchamianie przepływów pracy z użyciem ->run_workflow() do zagnieżdżonego komponowania
Trwałe opóźnienia z użyciem ->delay() do opóźnień czasowych
Sygnały z użyciem ->wait_for_signal(), ->wait_for_signals(), ->wait_for_approval(), ->wait_for_input() i ->wait_for_decision() do wyzwalania przez zdarzenia zewnętrzne
Zależności między przepływami pracy z użyciem ->wait_for_workflow(), ->wait_for_workflows() i ->wait_for_agents() do koordynacji między przepływami
Kroki strumieniowe z użyciem interfejsu StreamingStep do trwałego wyjścia w postaci fragmentów

Typowe zastosowania

Jeśli tworzysz systemy autonomiczne lub częściowo autonomiczne, zacznij od sekcji Orkiestracja agentów. Pokazuje ona, jak for_each(), start_agents(), wait_for_agents(), wait_for_workflow() i wait_for_decision() współdziałają w rzeczywistych przepływach planista/wykonawca i z udziałem weryfikacji przez człowieka.