コンテンツへスキップ

ワークフローの依存関係

Markdownとして表示 GitHubで編集 ドキュメント: 2cc7b2b

あるトップレベルのワークフローを、他のワークフローが完了した後にのみ続行させる場合は、ワークフロー依存関係の待機を使います。すべてを1つの巨大なワークフロー定義に押し込むことなく、ワークフロー間の連携を実現できます。

こうした待機は非同期ハンドオフと特に相性がよく、1つのワークフローで独立した子ワークフローを開始してから、後でそれらの完了を待てます。

プランナーと実行役の詳しい例については、エージェントオーケストレーションをご覧ください。

Queuetyには、次の3つのビルダーメソッドがあります:

  • 1つのワークフローを対象とするwait_for_workflow()
  • 複数のワークフローを対象とし、all、any、またはquorumのセマンティクスを指定するwait_for_workflows()
  • ワークフローIDが以前に開始したグループから取得される場合に使うwait_for_workflow_group()

1つのワークフローを待つ

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に変わります。待機中にワーカーがそのワークフローに割り当てられたままになることはありません。

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は、必須の依存ワークフローが1つでも失敗またはキャンセルされると失敗します。
  • WaitMode::Anyは、すべての依存ワークフローが終端状態に達し、成功したものが1つもない場合にのみ失敗します。
  • WaitMode::Quorumは、残りの実行中ワークフローでは必要な定足数を満たせなくなった場合に失敗します。

この仕組みにより、1つのワークフローがタスクごとにallでき、別のワークフローがその結果、または最初に完了した成功分岐を待てる、プランナーと実行役のシステムでワークフローの待機が役立ちます。