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

Зависимости workflow

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

Используйте ожидание зависимостей workflow, если один основной workflow должен продолжить работу только после завершения других workflow. Это позволяет координировать разные workflow, не объединяя все в одно огромное определение.

Такое ожидание особенно хорошо сочетается с Асинхронной передачей управления, когда один workflow сначала запускает независимые дочерние workflow, а затем ожидает их.

Примеры планировщика и исполнителя см. в разделе Оркестрация агентов.

Queuety предоставляет три метода конструктора:

  • wait_for_workflow() для одного workflow
  • wait_for_workflows() для нескольких workflow с семантикой all, any или quorum
  • wait_for_workflow_group(), если ID workflow взяты из ранее запущенной группы

Ожидание одного workflow

PHP
$research_id = Queuety::workflow( 'research_run' )
    ->then( PlanResearchStep::class )
    ->for_each( 'tasks', ExecuteResearchTask::class, 'results' )
    ->dispatch( [ 'brief_id' => 42 ] );

$editorial_id = Queuety::workflow( 'editorial_run' )
    ->wait_for_workflow( $research_id, 'research' )
    ->then( DraftBriefStep::class )
    ->dispatch();

Когда завершается research_run, выполнение editorial_run возобновляется. Если задано result_key, состояние завершенной зависимости сохраняется под этим ключом:

PHP
$state['research']; // public state from the completed research workflow

Если result_key не задан, публичное состояние зависимого workflow объединяется с состоянием ожидающего workflow верхнего уровня.

Ожидание нескольких workflow

PHP
use Queuety\Enums\WaitMode;

Queuety::workflow( 'synthesis_run' )
    ->wait_for_workflows(
        [ $research_id, $editorial_id, $legal_id ],
        WaitMode::All,
        'dependencies'
    )
    ->then( PublishPacketStep::class )
    ->dispatch();
  • WaitMode::All возобновляет выполнение только после завершения каждого workflow из списка.
  • WaitMode::Any возобновляет выполнение после завершения первого workflow из списка.
  • WaitMode::Quorum возобновляет выполнение, когда необходимое число workflow успешно завершится.
PHP
Queuety::workflow( 'synthesis_run' )
    ->wait_for_workflows(
        [ $research_id, $editorial_id, $legal_id ],
        WaitMode::Quorum,
        'dependencies',
        quorum: 2,
    )
    ->then( PublishPacketStep::class )
    ->dispatch();

При использовании result_key ожидание нескольких workflow сохраняет состояние завершенных зависимостей с привязкой к их ID:

PHP
[
    '42' => [ 'summary' => '...' ],
    '43' => [ 'summary' => '...' ],
]

Если result_key не задан, состояние завершенных зависимостей объединяется с состоянием ожидающего workflow верхнего уровня в порядке зависимостей.

Получение ID зависимостей из состояния workflow

Вы можете отложить создание списка зависимостей до выполнения, получив ID workflow из публичного состояния:

PHP
Queuety::workflow( 'agent_follow_up' )
    ->wait_for_workflow( 'parent_workflow_id', 'parent' )
    ->then( FinalizeAgentRunStep::class )
    ->dispatch( [ 'parent_workflow_id' => $workflow_id ] );

Для нескольких зависимостей:

PHP
Queuety::workflow( 'aggregator' )
    ->wait_for_workflows( 'dependency_ids', WaitMode::Any, 'winner' )
    ->dispatch( [ 'dependency_ids' => [ 12, 18, 22 ] ] );

Ожидание запущенной группы workflow

Если предыдущий шаг start_workflows() или start_agents() сохранил постоянный ключ группы, можно ожидать завершения этой группы напрямую, не передавая ID workflow через публичное состояние вручную.

PHP
Queuety::workflow( 'brief_research' )
    ->then( PlanTopicsStep::class )
    ->start_agents( 'agent_tasks', $agent_run, group_key: 'researchers' )
    ->wait_for_agent_group( 'researchers', WaitMode::Quorum, 2, 'agent_results' )
    ->then( SynthesizeBriefStep::class )
    ->dispatch();

Что происходит во время ожидания

Когда workflow доходит до одного из этих шагов, его статус меняется на waiting_for_workflows. Пока он ожидает, ни один worker не остается прикрепленным к нему.

PHP
$state = Queuety::workflow_status( $workflow_id );
echo $state->status->value; // 'waiting_for_workflows'
echo $state->wait_type;     // 'workflow'
print_r( $state->waiting_for ); // ['42', '43']

Если зависимость завершилась до того, как workflow дошел до шага ожидания, workflow продолжит выполнение сразу.

Обработка сбоев

Ожидание зависимостей workflow рассчитано на успешное завершение:

  • wait_for_workflow() и WaitMode::All завершаются сбоем, если какая-либо обязательная зависимость завершается сбоем или отменяется.
  • WaitMode::Any завершается сбоем, только если все зависимости достигли конечного состояния и ни одна из них не завершилась успешно.
  • WaitMode::Quorum завершается сбоем, когда оставшиеся активные workflow уже не могут обеспечить необходимый кворум.

Поэтому ожидание workflow полезно в системах с планировщиком и исполнителем: один workflow может распределять задания, а другой — ожидать результаты all или первого завершившегося успешного варианта.