工作流

以 Markdown 格式查看在 GitHub 上编辑文档位置 2cc7b2b

工作流会将长时间运行的任务拆分成离散步骤。每个步骤都会将输出持久化到数据库中的共享状态包中。如果 PHP 在步骤中途退出，工作进程会在保留所有先前状态的情况下重试该步骤。

什么是工作流？

工作流是按顺序排列的步骤序列。每个步骤都是一个 PHP 类，它接收此前所有步骤累积的状态，并返回要合并到状态中的新数据。工作流每次推进一个步骤，每次状态转换都由原子性的 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 期间崩溃，工作流会从步骤 1 重新开始，同时步骤 0 的输出已保存在状态中。

步骤边界是一个单独的 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()` 等待一个或多个其他工作流完成
`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 支持在工作流中使用十一个步骤类型：

顺序链：使用 ->then() 逐个执行
并行组：使用 ->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() 如何在实际的规划器/执行器和人工审核流程中协同工作。