コンテンツへスキップ

はじめに

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

Queuetyは、高速なジョブキュー、永続ワークフローエンジン、永続ステートマシンランタイムを提供するWordPressプラグインです。ワーカーはMySQLから直接ジョブを取得し、長時間稼働するWP-CLIプロセス内で処理します。WordPressとWP-CLIを読み込むPHPランタイムには、mysqliまたはpdo_mysqlが必要です。

Queuetyを選ぶ理由

WordPressには実用的なバックグラウンドジョブシステムがありません。wp_cronはページへのアクセス時にのみ実行されます。Action SchedulerのデフォルトランナーはWP-CronとループバックHTTPリクエストを使い、各リクエストで最大30秒間バッチを処理します。60秒かかるLLMAPI呼び出しは、PHPの30秒タイムアウトによって強制終了されます。クラッシュ後も状態を保ち、中断した箇所から再開できる複数ステップの処理を実行する方法はありません。

Queuetyは、次の3つの基本機能でこの問題を解決します:

  1. ジョブ: 非同期のバックグラウンド処理
  2. ワークフロー: 状態を永続化する複数ステップの処理
  3. ステートマシン: 長期間にわたるイベント駆動型のライフサイクル状態

使用例

モダンなAPIを使ってジョブをディスパッチします:

PHP
use Queuety\Contracts\Job;
use Queuety\Dispatchable;

readonly class SendEmailJob implements Job {
    use Dispatchable;

    public function __construct(
        public string $to,
        public string $subject,
        public string $body,
    ) {}

    public function handle(): void {
        wp_mail( $this->to, $this->subject, $this->body );
    }
}

SendEmailJob::dispatch( 'user@example.com', 'Welcome', 'Hello from Queuety!' );

または、従来のハンドラー名を指定する方法を使います:

PHP
use Queuety\Queuety;

Queuety::dispatch( 'send_email', [ 'to' => 'user@example.com' ] );

永続ワークフローを実行します:

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

ステートマシンで長期間にわたるセッションをモデル化します:

PHP
use Queuety\Enums\StateMachineStatus;

Queuety::machine( 'agent_session' )
    ->state( 'awaiting_user' )
    ->on( 'user_message', 'planning' )
    ->state( 'planning' )
    ->action( PlanSessionAction::class )
    ->on( 'planned', 'completed' )
    ->state( 'completed', StateMachineStatus::Completed )
    ->dispatch( [ 'thread_id' => 42 ] );

ワーカーを起動します:

シェル
wp queuety work

エージェント型システムを構築する場合は、エージェントオーケストレーションから始めてください。プランナーと実行役のフロー、人によるレビューのゲート、ワークフローをまたぐ待機を組み合わせる方法を説明しています。

イベント駆動型で長期間稼働するシステムの場合は、ワークフローとあわせてステートマシンから始めてください。

エンドツーエンドの統合例については例を参照してください。プロバイダーの切り替え、ストリーミング、メモリを使ったレビューの再実行、依存関係のあるワークフロー向けに、Neuronを使ったパターンが新たに追加されています。

WordPressのsave_postなどのドメインイベントをきっかけにワークフローを開始する場合は、アクショントリガーを参照してください。

WordPressの外部でデータベース連携ランタイムを使う場合は、スタンドアロンでの使用を参照してください。

主な特長

  • 高速な実行。 ワーカーはMySQLキューに直接アクセスし、WordPressでのバックグラウンド処理を遅くする、リクエストごとにcronを実行するパターンを避けます。
  • 永続ワークフロー。 PHPのタイムアウト、クラッシュ、再試行を経ても維持される状態を使い、複数ステップの処理を実行できます。各ステップの境界は、単一のMySQLトランザクションです。
  • 明示的なステートマシン。 チャットセッション、承認、その他の長期間にわたるライフサイクルを、名前付き状態、ガード付き遷移、キューに登録された状態遷移時のアクション、永続イベント履歴を使ってモデル化できます。
  • モダンなディスパッチAPI。 Dispatchableトレイト、ペイロードとしての型付きコンストラクタープロパティ、クラス上で宣言する$tries/$timeout/$backoffを備えた、単独で完結するジョブクラスです。readonlyクラスではこれらのプロパティにデフォルト値を設定できないため、readonlyの昇格コンストラクタープロパティを持つ通常のクラスで宣言してください。
  • ミドルウェアパイプライン。 レート制限、例外のスロットリング、ジョブの一意性、カスタムロジックに対応するオニオン型ミドルウェアです。ジョブごとにmiddleware()メソッドで宣言します。
  • ストリーミングステップ。 StreamingStepインターフェースから生成されたチャンクは、すぐにデータベースへ保存されます。再試行時には、保存済みのチャンクが渡されるため、中断箇所からストリームを再開できます。LLMの応答や大容量のダウンロードに適しています。
  • プラグ可能なキャッシュレイヤー。 CacheFactoryはAPCuを自動検出し、利用できない場合はインメモリキャッシュに切り替えます。カスタムバックエンドはContracts\Cacheを実装します。内部ではレート制限の参照とキュー状態に使用されます。
  • バッチ処理とチェーン。 then/catch/finallyコールバックと進捗追跡を使ってジョブのグループをディスパッチできます。または、前のジョブの完了を条件に、ジョブを順番に連結できます。
  • 永続的な遅延と待機。 delay()は、プロセスの再起動後も維持される遅延を追加します。wait_for_signal()、wait_for_signals()、wait_for_approval()、wait_for_input()、wait_for_workflow()を使うと、PHPを稼働させたままにせず、外部イベントや別のワークフローを待つようワークフローを一時停止できます。
  • 動的なfor-each。 for_each()は、実行時に検出した分岐処理をワークフロー状態から展開し、全成功、最初の成功、クォーラムによる完了モードに対応します。
  • 非同期のワークフロー引き継ぎ。 start_workflows()は、実行時に検出した項目を、後から待機できる独立したトップレベルのワークフローに変換します。
  • WordPressアクショントリガー。 on_action()は、コアまたはプラグインのアクションフックを、ゲート制御と冪等性を備えた永続ワークフローのディスパッチに対応付けます。
  • 永続成果物。 名前付きのワークフロー出力をメインの状態バッグとは別に保存し、ステータス、CLIによる検査、エクスポート、再生で引き継げます。
  • ワークフローのガードレール。 version()、決定論的な定義ハッシュ、冪等なディスパッチキー、ワークフローの予算により、エージェント型の実行を確認しやすくし、重複や気づかないままの暴走を防ぎます。
  • リソース管理。 同時実行グループ、コスト単位、ワークフロー予算、ワーカーの受け入れ制御により、コストの高いエージェント型処理を定義済みの範囲内に保ちます。
  • ステップの補償処理。 compensate_with()とcompensate_on_failure()は、ワークフローがキャンセルまたは失敗した際に、完了済みのステップをロールバックするフックを提供します。
  • ハートビート。 長時間実行されるステップはハートビートを送り、古いジョブの復旧システムが実行中の処理を回収しないようにします。
  • ワークフローのトレース。 各ステップ遷移について、入力、出力、遷移前後の状態、コンテキスト、成果物、チャンク、エラーを記録し、workflow_state_at()によるタイムトラベルデバッグにも対応します。
  • 並列ステップ。 ステップを同時に実行し、すべての完了を待ってから先へ進みます。
  • 条件分岐。 _next_stepを使い、前の状態に応じて名前付きステップへ分岐します。
  • ワークフロー実行。 結果を親ワークフローに返す子ワークフローを開始します。
  • 優先度付きキュー。 型安全なPHP列挙型で、低、通常、高、緊急の4段階を指定できます。
  • 複数キュー対応ワーカー。 1つのワーカーで、厳密な優先順位に従って複数のキューを処理します。
  • レート制限。 スライディングウィンドウ方式で、ハンドラーごとの実行回数を制限します。
  • 繰り返しジョブ。 間隔ベースおよびcronベースのスケジューリングに対応し、重複実行のポリシー（許可、スキップ、バッファ）を設定できます。
  • ワークフローのキャンセル。 実行中のワークフローをキャンセルし、登録済みのクリーンアップハンドラーを起動します。
  • 状態の整理。 古いステップ出力を自動的に削除し、大規模なワークフロー状態を管理しやすくします。
  • ワーカーの並行実行。 --workers=Nは複数のプロセスをフォークし、クラッシュ時には自動的に再起動します。
  • 包括的な可観測性。 データベースログ、ハンドラーごとのメトリクス、イベントWebhookに対応します。
  • テストユーティリティ。 QueueFakeはディスパッチされたジョブとバッチを記録し、単体テストでアサーションできます。
  • PHP属性。 #[QueuetyHandler('name')]を使い、設定不要でハンドラーを登録できます。

仕組み

ワーカーは長時間稼働するWP-CLIプロセス内で動き、MySQLから直接ジョブを取得します。キュー行、ワークフロー状態、ログ、バッチ、シグナル、ストリーミングチャンクはすべてMySQLに保存されるため、ワーカーを再起動してもオーケストレーションの状態は失われません。

ワークフローは、長時間かかる処理をステップに分割します。各ステップは出力をデータベース内の共有状態バッグに保存します。PHPがステップの途中で停止しても、ワーカーはそれまでの状態をすべて保ったまま、そのステップを再試行します。ステップの境界では単一のMySQLトランザクションが使われ、状態の更新、ジョブの完了、次のステップのキュー登録がすべてアトミックに行われます。

コード
Workflow: generate_report (3 steps)

Step 0: fetch_data     -> state: {user_data: {...}}
Step 1: call_llm       -> PHP dies -> retry -> state: {user_data: {...}, llm_response: "..."}
Step 2: format_output  -> state: {user_data: {...}, llm_response: "...", report_url: "/reports/42.pdf"}

Workflow: completed

必要条件

  • PHP 8.2以降
  • WordPress 6.4以降
  • MySQL 8.0+またはMariaDB 10.6+（ワーカーはSELECT ... FOR UPDATE SKIP LOCKEDを使ってジョブを取得するため）
  • WordPressとWP-CLIを読み込むPHPランタイムで、mysqliまたはpdo_mysql
  • 専用ワーカーをwp queuety workで実行するためのWP-CLI
  • ワーカープール（--workers、--min-workers、--max-workers）用のpcntlおよびposix PHP拡張機能

インストール方法ははじめにを参照してください。