工作流依赖

以 Markdown 格式查看在 GitHub 上编辑文档位置 2cc7b2b

当一个顶层工作流应仅在其他工作流完成后继续执行时，可以使用工作流依赖等待。这样就能协调多个工作流，而无需将所有内容塞进一个庞大的工作流定义中。

这种等待方式与异步交接尤其适合搭配使用：一个工作流先启动多个独立的子工作流，然后再等待它们完成。

如需更完整的规划器/执行器示例，请参阅代理编排。

Queuety 提供三种构建器方法：

wait_for_workflow() 用于等待一个工作流
wait_for_workflows() 用于等待多个工作流，支持 all、any 或 quorum 语义
wait_for_workflow_group() 用于等待工作流 ID 来自先前启动的组的情况

等待一个工作流

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，依赖工作流的公开状态会合并到等待中的工作流的顶层状态中。

等待多个工作流

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 仅在列表中的所有工作流都完成后恢复执行。
WaitMode::Any 在列表中的第一个工作流完成时恢复执行。
WaitMode::Quorum 在所需数量的工作流成功完成后恢复执行。

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 时，多工作流等待会按工作流 ID 为已完成依赖项的状态建立索引并存储：

PHP

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

如果没有 result_key，已完成依赖项的状态会按依赖顺序合并到等待中的工作流的顶层状态中。

从工作流状态解析依赖 ID

你可以在运行时通过读取公开状态中的工作流 ID，延后确定依赖项列表：

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

等待已启动的工作流组

如果先前的 start_workflows() 或 start_agents() 步骤存储了一个持久化组键，你可以直接等待该组，而不必自己通过公开状态传递工作流 ID。

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

等待期间会发生什么

工作流到达这些步骤之一时，会进入 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']

如果依赖项在到达等待步骤之前就已完成，工作流会立即继续执行。

失败语义

工作流依赖等待以成功为导向：

wait_for_workflow() 和 WaitMode::All 在任何必需依赖项失败或取消时都会失败。
WaitMode::Any 仅当所有依赖项都已进入终止状态且没有任何一个成功完成时才会失败。
WaitMode::Quorum 在剩余活动工作流已无法满足所需法定数量时失败。

因此，工作流等待适用于规划器/执行器系统：一个工作流可以为任务派发工作，另一个工作流则等待 all 结果，或等待第一个完成的成功分支。