跳转到正文

简介

以 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 秒的 LLM API 调用会被 PHP 的 30 秒超时机制终止。它无法运行在崩溃后仍能存续并从中断处恢复的多步骤流程。

Queuety 通过三个基本功能解决了这一问题：

  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 队列，避免了 cron 模式；这种模式会拖慢 WordPress 中的后台工作。
  • 持久工作流。 具有持久状态的多步骤流程，即使遇到 PHP 超时、崩溃和重试也能继续。每个步骤边界都是一个原子的 MySQL 事务。
  • 显式状态机。 使用命名状态、受保护的转换、排队的状态进入操作和持久事件时间线，对聊天会话、审批及其他长期运行的生命周期进行建模。
  • 现代派发 API。 自包含任务类，使用 Dispatchable trait、将类型化构造函数属性用作负载，并在类上声明 $tries/$timeout/$backoff。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() 会根据工作流状态展开运行时发现的分支任务，并支持全部成功、首个成功和法定人数完成模式。
  • 异步工作流交接。 start_workflows() 会将运行时发现的项目转换为独立的顶层工作流，之后可等待这些工作流完成。
  • WordPress 操作触发器。 on_action() 会将核心或插件操作钩子映射到持久工作流派发，并提供门控和幂等性。
  • 持久制品。 将命名的工作流输出存储在主状态包之外，并在状态、CLI 检查、导出和重放过程中保留这些输出。
  • 工作流防护机制。 version()、确定性定义哈希、幂等派发键和工作流预算，让智能体风格的运行更易检查，也更不容易重复执行或在无人察觉时失控运行。
  • 资源管理。 并发组、成本单位、工作流预算和工作进程准入机制有助于将高成本的智能体风格任务限制在明确范围内。
  • 步骤补偿。 compensate_with() 和 compensate_on_failure() 为工作流取消或失败时已完成的步骤提供回滚钩子。
  • 心跳。 长时间运行的步骤会发送心跳，防止过期任务恢复系统回收仍在执行的工作。
  • 工作流追踪。 每次步骤转换都会记录输入、输出、转换前后的状态、上下文、制品、数据块、错误和时间穿梭调试信息（通过 workflow_state_at()）。
  • 并行步骤。 并发运行多个步骤，并在全部完成后再继续。
  • 条件分支。 使用 _next_step 根据先前状态跳转到指定步骤。
  • 运行工作流。 启动子工作流，并将结果返回给父工作流。
  • 优先级队列。 通过类型安全的 PHP 枚举划分四个级别（低、普通、高、紧急）。
  • 多队列工作进程。 在单个工作进程中按严格优先级顺序处理多个队列。
  • 速率限制。 限制每个处理器的执行次数，采用滑动窗口。
  • 定期任务。 基于间隔和 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
  • 用于工作进程池的 pcntl 和 posix PHP 扩展（--workers、--min-workers、--max-workers）

安装说明请参阅入门。