コンテンツへスキップ

エージェントオーケストレーション

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

Queuetyは、永続的な状態、再試行、確認、明示的な待機の仕組みを保ちながら、柔軟に作業を進めたいエージェントシステムに適しています。

多くのエージェントフローに、実行時に編集できるDAGエンジンは必要ありません。実際には、少数の永続的なオーケストレーション操作で多くのシステムをモデル化できます。for_each()は1つのワークフロー実行内で作業を展開します。start_agents()とwait_for_agents()は、実行時に見つかった作業を独立したトップレベルワークフローに引き渡します。wait_for_decision()とwait_for_input()は人の判断を待つために一時停止します。wait_for_workflow()、wait_for_workflows()、wait_for_agent_group()を使うと、1つのトップレベル実行で、別の実行や開始済みのクォーラムを待機できます。

考え方のヒント

エージェントオーケストレーションは、層に分けると考えやすくなります。プランナーがどの作業が必要かを判断します。Queuetyがその作業を永続的なジョブやワークフローに変換します。専門処理の実行が結果を生成し、アーティファクトを保存します。監督段階で結果をまとめ、次の処理を決定します。実行の途中でも、人が承認、却下、修正してから続行できます。

重要な違いは、エージェントが実行する作業を決める一方、作業の永続化、再試行、再開、集約の方法はQueuetyが引き続き決めることです。オーケストレーションの対象が期間限定の実行ではなく長期的なセッションなら、State Machinesと組み合わせてください。

パターン1: 独立したエージェント実行を使うプランナー/実行役

プランナーステップが実行時にタスクを見つけ、それぞれを独立したトップレベルワークフローとして実行する場合は、start_agents()を使います。

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

$agent_run = Queuety::workflow( 'agent_run' )
    ->then( FetchSourcesStep::class )
    ->then( DraftFindingStep::class );

Queuety::workflow( 'brief_research' )
    ->version( 'brief-research.v1' )
    ->idempotency_key( 'brief:42:research' )
    ->max_transitions( 20 )
    ->then( PlanResearchTasksStep::class )
    ->start_agents( 'agent_tasks', $agent_run, group_key: 'researchers' )
    ->wait_for_agent_group( 'researchers', WaitMode::Quorum, 2, 'agent_results' )
    ->then( SynthesizeBriefStep::class )
    ->dispatch( [ 'brief_id' => 42 ] );

このパターンでは、PlanResearchTasksStepが実行時に見つかった項目を$state['agent_tasks']に書き込み、start_agents()がタスクごとに永続的なワークフローを1つディスパッチし、wait_for_agent_group()が選択された完了条件を満たすまで親を待機させます。その後、SynthesizeBriefStepが$state['agent_results']の下に完了した子の状態を受け取ります。

この方法は、各エージェントの実行を個別に確認、再試行できるため、プランナー/実行役のシステムを始めるのに最もわかりやすい方法です。

パターン2: 実行を続ける前に人がレビュー

ワークフローを明確な承認または却下の判断で停止する場合はwait_for_decision()を使い、担当者から実行に構造化されたフィードバックを送る必要がある場合はwait_for_input()を使います。

PHP
use Queuety\Queuety;

$workflow_id = Queuety::workflow( 'brief_review' )
    ->then( DraftBriefStep::class )
    ->wait_for_decision( result_key: 'review' )
    ->then( HandleReviewDecisionStep::class )
    ->wait_for_input( result_key: 'revision_notes' )
    ->then( ApplyRevisionNotesStep::class )
    ->dispatch( [ 'brief_id' => 42 ] );

レビュアーが承認した場合:

PHP
Queuety::approve_workflow(
    $workflow_id,
    [
        'reviewer' => 'editor@example.com',
        'notes' => 'Looks good, publish after legal check',
    ],
    'approved'
);

レビュアーが却下した場合:

PHP
Queuety::reject_workflow(
    $workflow_id,
    [
        'reviewer' => 'editor@example.com',
        'reason' => 'Needs citations and a stronger conclusion',
    ],
    'rejected'
);

wait_for_decision()の後、ワークフローには最終結果、実行を再開させたシグナル、およびそれに付随するレビュー担当者のデータを含む構造化ペイロードが渡されます。これにより、実行が待機中でもPHPプロセスを開いたままにせず、真のヒューマン・イン・ザ・ループのチェックポイントを設けられます。

パターン3：トップレベルのワークフローが別のワークフローを待つ

常に1つの巨大なワークフローにまとめるのが最善とは限りません。あるワークフローで調査を完了し、別のワークフローがそれを待ってから次の段階を開始できます。

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

$research_id = Queuety::workflow( 'research_brief' )
    ->then( PlanResearchTasksStep::class )
    ->start_agents( 'agent_tasks', $agent_run )
    ->wait_for_agents( mode: WaitMode::All, result_key: 'agent_results' )
    ->then( SynthesizeBriefStep::class )
    ->dispatch( [ 'brief_id' => 42 ] );

Queuety::workflow( 'publish_brief' )
    ->wait_for_workflow( $research_id, 'research' )
    ->wait_for_decision( result_key: 'editor_review' )
    ->then( PublishBriefStep::class )
    ->dispatch( [ 'brief_id' => 42 ] );

これにより、調査段階と公開段階の間に永続的な境界を設けられます。各実行はそれぞれ独自の運用ライフサイクルを持ちますが、依存先の完了時には2つ目のワークフローが自動的に再開します。

適切なプリミティブの選択

for_each()は、処理が1つのワークフロー実行内で完結し、分岐が同じ共有状態バッグを使うステップ単位の並列実行にすぎない場合に適しています。

start_agents()は、見つかった各タスクを独立した永続的なトップレベル実行にし、それぞれを個別に確認、再試行、またはエクスポートしたい場合に適しています。

wait_for_workflow()とwait_for_workflows()は、あるトップレベルのワークフローが別のトップレベルのワークフローに依存し、オーケストレーション層間で永続的な非同期調整を行いたい場合に適しています。

ワークフローをワーカーの容量を消費せずに一時停止したまま、人が実行を承認、拒否、または誘導する必要がある箇所には、wait_for_decision()とwait_for_input()が適しています。

推奨される状態の規約

状態キーが予測可能だと、エージェントシステムのデバッグが容易になります。実際には、agent_tasksはプランナーの出力、agent_workflow_idsは開始したトップレベル実行、agent_resultsは完了した子ワークフローの状態、reviewまたはdecisionは承認または拒否のペイロードに適しています。下書き、引用、調査マニフェストなどの大きな出力は、通常、メインのワークフロー状態バッグではなくアーティファクトストアに保存します。

エージェントのワークフローは本質的に動的なため、ほかのほぼすべてのユースケース以上にガードレールが役立ちます。

PHP
Queuety::workflow( 'agent_research' )
    ->version( 'agent-research.v2' )
    ->idempotency_key( 'brief:42:agent-research' )
    ->max_transitions( 20 )
    ->max_for_each_items( 12 )
    ->max_state_bytes( 32768 );

こうした制御によって、本番環境で重要となる実務上の疑問に答えられます。実行を生成した定義のバージョン、重複した外部リクエストによる二重ディスパッチの有無、プランナーが分岐を作りすぎる可能性、ワークフロー状態が際限なく増える可能性などです。ほとんどのエージェントシステムでは、ワーカー数やキューの構成を調整する前に、まずこれらのガードレールを設定してください。

関連項目