ワークフロー

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

ワークフローは長時間かかる処理を複数のステップに分割します。各ステップの出力は、データベース内の共有状態バッグに保存されます。PHPがステップの途中で停止すると、ワーカーはそれまでの状態をすべて保持したまま、そのステップを再試行します。

ワークフローとは？

ワークフローは、順序付けられたステップの連続です。各ステップはPHPクラスで、前のすべてのステップで蓄積された状態を受け取り、新しいデータを返して状態にマージします。ワークフローは一度に1ステップずつ進み、各遷移はアトミックなMySQLトランザクションで実行されます。

PHP

use Queuety\Queuety;

$workflow_id = Queuety::workflow( 'generate_report' )
    ->then( FetchDataHandler::class )
    ->then( CallLLMHandler::class )
    ->then( FormatOutputHandler::class )
    ->dispatch( [ 'user_id' => 42 ] );

状態の蓄積

dispatch()に渡す初期ペイロードが開始状態になります。各ステップが返すキーは状態にマージされます。_で始まるキーは_next_stepなどの制御値用に予約されており、公開状態にはマージされません。

コード

dispatch(['user_id' => 42])

Step 0: FetchDataHandler   -> returns ['user_name' => 'Alice']
  state: {user_id: 42, user_name: 'Alice'}

Step 1: CallLLMHandler     -> returns ['llm_response' => '...']
  state: {user_id: 42, user_name: 'Alice', llm_response: '...'}

Step 2: FormatOutputHandler -> returns ['report_url' => '/reports/42.pdf']
  state: {user_id: 42, user_name: 'Alice', llm_response: '...', report_url: '/reports/42.pdf'}

各ステップは、蓄積された状態全体を参照できます。外部ストレージを使わずにステップ間でデータを渡せます。

再開可能性

ワークフローは、各ステップの後に状態がデータベースへ保存されるため、確実に実行できます。ワーカーがステップ1の実行中にクラッシュしても、ワークフローはステップ0の出力が状態に残ったまま、ステップ1から再開します。

ステップの境界は、次の処理をアトミックに実行する単一のMySQLトランザクションです。

ステップの出力でワークフロー状態を更新する
現在のステップのジョブを完了済みとしてマークする
次のステップのジョブをキューに追加する

これにより、クラッシュ、タイムアウト、PHPの致命的エラーが発生しても、状態が失われないことが保証されます。

ワークフローのライフサイクル

コード

dispatch() -> running -> step 0 -> step 1 -> ... -> step N -> completed
                  |                   |
                  |                   +-> failed -> retry -> running (from failed step)
                  |
                  +-> paused -> resumed -> running
                  |
                  +-> waiting_for_signal -> signal received -> running
                  |
                  +-> waiting_for_workflows -> dependency settles -> running
                  |
                  +-> cancelled (via cancel_workflow)

ステータス	説明
`running`	ステップを処理中
`completed`	すべてのステップが正常に完了
`failed`	再試行を使い切った後にステップが失敗
`paused`	手動で一時停止中。進行中の作業は完了する場合がありますが、再開するまで後続のステップはキューに追加されません
`waiting_for_signal`	`Queuety::signal()`を介した外部シグナルを待機中
`waiting_for_workflows`	`wait_for_workflow()`または`wait_for_workflows()`を介して、他の1つ以上のワークフローの完了を待機中
`cancelled`	`Queuety::cancel_workflow()`で明示的にキャンセル済み。保留中のジョブは埋められ、任意のクリーンアップハンドラーが実行されます。

ワークフローの管理

ステータスを確認する：

PHP

$state = Queuety::workflow_status( $workflow_id );
echo $state->status->value;   // 'running'
echo $state->current_step;    // 1
echo $state->total_steps;     // 3
print_r( $state->waiting_for ); // ['42'] while waiting on another workflow
print_r( $state->wait_details ); // matched/remaining blockers, result key, correlation info

一時停止と再開：

PHP

Queuety::pause_workflow( $workflow_id );
Queuety::resume_workflow( $workflow_id );

失敗したワークフローを、失敗したステップから再試行する：

PHP

Queuety::retry_workflow( $workflow_id );

ワークフローをキャンセルし、クリーンアップハンドラーを実行する：

PHP

Queuety::cancel_workflow( $workflow_id );

クリーンアップハンドラーやワークフローのキャンセル時の動作については、キャンセルを参照してください。

ワークフローのオプション

ワークフロー内のすべてのステップに、キュー、優先度、再試行の動作を設定します。

PHP

use Queuety\Enums\Priority;

Queuety::workflow( 'important_pipeline' )
    ->on_queue( 'pipelines' )
    ->with_priority( Priority::High )
    ->max_attempts( 5 )
    ->on_cancel( PipelineCleanupHandler::class )
    ->prune_state_after( 3 )
    ->then( StepA::class )
    ->then( StepB::class )
    ->dispatch( $payload );

on_cancel()メソッドは、ワークフローがキャンセルされたときに実行するクリーンアップハンドラーを登録します。キャンセルを参照してください。

prune_state_after()メソッドは、ワークフロー状態のサイズを抑えるために、古いステップ出力を自動的に削除します。状態の整理を参照してください。

compensate_with()を使うと、ステップごとの補償ハンドラーを設定できます。ワークフローが失敗したときに、明示的なキャンセル時だけでなく、完了済みステップの補償処理も自動実行するにはcompensate_on_failure()を使います。補償処理を参照してください。

長時間のオーケストレーションをより安全に行うには、ワークフローのガードレールを使います。version()は定義にアプリケーションレベルのバージョンを付け、Queuetyは各実行の決定論的な定義ハッシュも保存し、idempotency_key()は再試行や重複リクエストに対するディスパッチの耐久性を高めます。max_transitions()、max_for_each_items()、max_state_bytes()は、実行が暴走する前に失敗させます。ガードレールを参照してください。

ステップの種類

Queuetyは、ワークフロー内で11種類のステップをサポートします。

逐次チェーンでは->then()を使って1つずつ実行
並列グループでは->parallel()を使って並行実行
動的for-eachでは->for_each()を使って、実行時に検出されたブランチ作業と明示的な完了条件を処理
非同期ハンドオフでは->start_workflows()または->start_agents()を使って、実行時に検出された項目から独立した子ワークフローをディスパッチ
条件分岐では名前付きステップと_next_stepを使用
繰り返しでは->repeat_until()と->repeat_while()を使って、以前の名前付きステップへ永続的に戻る
ワークフローの実行では->run_workflow()を使って入れ子に構成
永続的な遅延では->delay()を使って時間ベースの遅延を設定
シグナルでは->wait_for_signal()、->wait_for_signals()、->wait_for_approval()、->wait_for_input()、->wait_for_decision()を使って外部イベントをトリガー
ワークフローの依存関係では->wait_for_workflow()、->wait_for_workflows()、->wait_for_agents()を使ってワークフロー間を調整
ストリーミングステップではStreamingStepインターフェースを使って、チャンク単位の出力を永続化

よくあるパターン

自律型または半自律型システムを構築する場合は、エージェントオーケストレーションから始めてください。for_each()、start_agents()、wait_for_agents()、wait_for_workflow()、wait_for_decision()が、実際のプランナー／エグゼキューターや人間によるレビューのフローでどのように連携するかを説明しています。