非同期ハンドオフ

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

すべてを1つの実行内にとどめるのではなく、ワークフローが見つけた作業を独立したトップレベルのワークフローに引き渡す必要がある場合は、start_workflows()を使います。

エージェント実行を明示的にモデル化する場合、start_agents()とwait_for_agents()は、エージェント向けのデフォルト値を持つ意味上のエイリアスです。

エンドツーエンドの詳細な例は、エージェントオーケストレーションを参照してください。

これは、エージェント形式のオーケストレーションに便利です。

プランナーワークフローが実行時にタスクを見つける
各タスクが独自の永続的なワークフローになる
親がそれらのワークフローIDを状態に保存する
親は後からall、any、quorum、または名前付きの開始済みグループを待てる

`->start_workflows()`メソッド

PHP

use Queuety\Enums\WaitMode;
use Queuety\Queuety;

$agent_task = Queuety::workflow( 'agent_task' )
    ->then( ResearchTopicStep::class )
    ->then( SummarizeTopicStep::class );

Queuety::workflow( 'brief_research' )
    ->then( PlanTopicsStep::class )
    ->start_workflows(
        items_key: 'topics',
        workflow_builder: $agent_task,
        result_key: 'child_workflow_ids',
        group_key: 'researchers',
    )
    ->wait_for_workflow_group( 'researchers', WaitMode::All, result_key: 'child_results' )
    ->then( SynthesizeBriefStep::class )
    ->dispatch( [ 'brief_id' => 42 ] );

この例では、次のようになっています。

PlanTopicsStepがトピックのペイロード配列を$state['topics']に書き込む
start_workflows()がトピックごとにagent_taskワークフローを1つずつディスパッチする
開始したワークフローのIDが$state['child_workflow_ids']に書き込まれる
wait_for_workflow_group()は、選択した完了条件を開始済みグループが満たすまで一時停止する
SynthesizeBriefStepは、完了した子の状態を$state['child_results']に受け取る

子のペイロード

items_keyの各項目が、開始する1つのワークフローの初期ペイロードになります。

項目が配列の場合、その公開キーが子のペイロードにマージされます。

PHP

[
    [ 'topic' => 'pricing', 'source' => 'competitors' ],
    [ 'topic' => 'reviews', 'source' => 'customers' ],
]

項目がスカラーの場合、Queuetyはそれをpayload_keyの下に保存します。

PHP

->start_workflows(
    items_key: 'topic_names',
    workflow_builder: $agent_task,
    payload_key: 'topic',
)

親の状態を引き継ぐ

デフォルトでは、start_workflows()は親ワークフローの公開状態をすべての子のペイロードにマージします。

つまり、brief_id、account_id、campaignなどの共有コンテキストは、開始した各ワークフローで自動的に利用できます。

ソースのitems_keyとresult_keyは引き継ぎ対象から除外されるため、各子にタスクリスト全体や最終的なワークフローID配列が渡されることはありません。

各子に自分の項目のペイロードだけを渡す場合は、inherit_state: falseを設定します。

`run_workflow()`との違い

run_workflow()は親子関係を作り、子が完了するまで親はそのステップで待機します。

start_workflows()は、より疎結合です。

開始したワークフローは独立したトップレベルのワークフロー
開始した各ワークフローを個別に確認、再試行、またはエクスポートできる
親は後から、待つかどうか、またどのように待つかを決定する

そのため、start_workflows()は非同期のエージェントタスク引き渡しに適しており、run_workflow()は1つのオーケストレーションツリー内でのネスト構成に適しています。

後から待つ

start_workflows()は、ワークフロー依存関係と組み合わせて使うよう設計されています。

PHP

use Queuety\Enums\WaitMode;

Queuety::workflow( 'review_board' )
    ->start_workflows( 'review_tasks', $review_task_workflow, 'review_workflow_ids' )
    ->wait_for_workflows( 'review_workflow_ids', WaitMode::Any, 'winner' )
    ->then( FinalizeDecisionStep::class )
    ->dispatch( $payload );

WaitMode::Allは、完了した子の状態をすべて収集する
WaitMode::Anyは、最初に正常完了した子ワークフローで再開する

これにより、実行時に編集可能な完全なDAGエンジンを用意しなくても、プランナーとエグゼキューターによるオーケストレーションが可能です。

エージェント向けエイリアス

start_agents()は、プランナーとエグゼキューターのフローに合わせたデフォルト値を持つstart_workflows()のエイリアスです。

result_keyのデフォルトはagent_workflow_ids
payload_keyのデフォルトはagent_task
nameのデフォルトはstart_agents

wait_for_agents()は、エージェントの結果収集に合わせたデフォルト値を持つwait_for_workflows()のエイリアスです。

workflowsのデフォルトはagent_workflow_ids
result_keyのデフォルトはagent_results
nameのデフォルトはwait_for_agents

wait_for_agent_group()は、エージェントの一括処理に合わせたデフォルト値を持つwait_for_workflow_group()のエイリアスです。

PHP

use Queuety\Enums\WaitMode;

$agent_run = Queuety::workflow( 'agent_run' )
    ->then( ResearchTopicStep::class )
    ->then( SummarizeTopicStep::class );

Queuety::workflow( 'brief_research' )
    ->then( PlanTopicsStep::class )
    ->start_agents( 'agent_tasks', $agent_run, group_key: 'researchers' )
    ->wait_for_agent_group( 'researchers', WaitMode::Quorum, 2 )
    ->then( SynthesizeBriefStep::class )
    ->dispatch( [ 'brief_id' => 42 ] );