Перейти к содержимому

Рабочие процессы

Просмотреть как Markdown Редактировать на GitHub Документация по 2cc7b2b

Рабочие процессы разбивают длительные задачи на отдельные шаги. Результат каждого шага сохраняется в общей базе данных состояний. Если PHP завершается во время выполнения шага, исполнитель повторяет этот шаг, сохраняя все данные предыдущих шагов.

Что такое рабочие процессы?

Рабочий процесс — это упорядоченная последовательность шагов. Каждый шаг — это класс PHP, который получает накопленное состояние всех предыдущих шагов и возвращает новые данные для объединения с состоянием. Рабочий процесс продвигается по одному шагу за раз, а каждый переход выполняется в рамках атомарной транзакции MySQL.

PHP
use Queuety\Queuety;

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

Накопление состояния

Начальная полезная нагрузка, переданная в dispatch(), становится исходным состоянием. Возвращенные каждым шагом ключи объединяются с состоянием. Ключи, начинающиеся с _, зарезервированы для управляющих значений, таких как _next_step, и не объединяются с общедоступным состоянием:

Код
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'}

Каждый шаг получает полное накопленное состояние. Благодаря этому данные легко передавать между шагами без внешнего хранилища.

Возможность возобновления

Рабочие процессы надежны, поскольку состояние сохраняется в базе данных после каждого шага. Если исполнитель аварийно завершится во время шага 1, рабочий процесс перезапустится с шага 1, а результат шага 0 уже будет в состоянии.

Граница шага — это одна транзакция MySQL, которая атомарно:

  1. Обновляет состояние рабочего процесса результатом шага
  2. Помечает задание текущего шага как завершенное
  3. Ставит в очередь задание следующего шага

Это гарантирует, что состояние никогда не будет потеряно, даже при сбоях, тайм-аутах или фатальных ошибках PHP.

Жизненный цикл рабочего процесса

Код
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)
Статус Описание
running Активно выполняются шаги
completed Все шаги успешно завершены
failed Шаг завершился ошибкой после исчерпания повторных попыток
paused Приостановлен вручную. Текущая выполняемая задача может завершиться, но последующие шаги не ставятся в очередь до возобновления
waiting_for_signal Ожидание внешнего сигнала через Queuety::signal()
waiting_for_workflows Ожидание завершения одного или нескольких других рабочих процессов через wait_for_workflow() или wait_for_workflows()
cancelled Явная отмена через Queuety::cancel_workflow(). Ожидающие задания скрываются, а при наличии выполняется обработчик очистки.

Управление рабочими процессами

Проверка статуса:

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

Приостановка и возобновление:

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

Повторный запуск рабочего процесса с шага, завершившегося ошибкой:

PHP
Queuety::retry_workflow( $workflow_id );

Отмена рабочего процесса и запуск обработчиков очистки:

PHP
Queuety::cancel_workflow( $workflow_id );

Подробности об обработчиках очистки и о том, что происходит при отмене рабочего процесса, см. в разделе Отмена.

Параметры рабочего процесса

Настройка очереди, приоритета и поведения при повторных попытках для всех шагов рабочего процесса:

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

Метод on_cancel() регистрирует обработчик очистки, запускаемый при отмене рабочего процесса. См. раздел Отмена.

Метод prune_state_after() включает автоматическое удаление результатов старых шагов, чтобы ограничить размер состояния рабочего процесса. См. раздел Очистка состояния.

Используйте compensate_with(), чтобы назначить обработчик компенсации для отдельного шага. Используйте compensate_on_failure(), если нужно автоматически запускать компенсацию завершенных шагов после сбоя рабочего процесса, а не только при явной отмене. См. раздел Компенсация.

Используйте ограничения рабочего процесса, чтобы безопаснее управлять длительной оркестрацией: version() помечает определение версией на уровне приложения, Queuety также сохраняет детерминированный хеш определения для каждого запуска, idempotency_key() обеспечивает надежную отправку при повторных попытках или дублирующих запросах, а max_transitions(), max_for_each_items() и max_state_bytes() завершают вышедшие из-под контроля запуски до того, как ситуация усугубится. См. раздел Ограничения.

Типы шагов

Queuety поддерживает одиннадцать типов шагов в рабочем процессе:

Типовые сценарии

Если вы создаете автономные или полуавтономные системы, начните с раздела Оркестрация агентов. В нем показано, как for_each(), start_agents(), wait_for_agents(), wait_for_workflow() и wait_for_decision() взаимодействуют в реальных сценариях планирования и выполнения задач, а также проверки человеком.