# 工作流

工作流会将长时间运行的任务拆分成离散步骤。每个步骤都会将输出持久化到数据库中的共享状态包中。如果 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 事务，它会原子性地：
1. 使用步骤输出更新工作流状态
2. 将当前步骤的作业标记为已完成
3. 将下一步骤的作业加入队列

这能确保状态绝不会丢失，即使发生崩溃、超时或 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 );
```

有关清理处理器以及工作流取消后会发生什么的详细信息，请参阅 [取消](https://queuety.inline0.dev/zh/docs/workflows/cancellation)。

## 工作流选项

为工作流中的所有步骤设置队列、优先级和重试行为：

```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()` 方法会注册一个在工作流取消时运行的清理处理器。请参阅 [取消](https://queuety.inline0.dev/zh/docs/workflows/cancellation)。

`prune_state_after()` 方法会启用自动清除旧步骤输出的功能，以限制工作流状态大小。请参阅 [状态修剪](https://queuety.inline0.dev/zh/docs/features/state-pruning)。

使用 `compensate_with()` 为每个步骤附加补偿处理器。如果希望工作流失败后自动运行已完成步骤的补偿操作，而不仅仅在明确取消时运行，请使用 `compensate_on_failure()`。请参阅 [补偿](https://queuety.inline0.dev/zh/docs/workflows/compensation)。

需要更安全地编排长时间运行的任务时，请使用工作流防护措施：`version()` 为定义添加应用层版本标签，Queuety 还会为每次运行存储确定性的定义哈希，`idempotency_key()` 可确保调度在重试或重复请求后仍可靠，而 `max_transitions()`、`max_for_each_items()` 和 `max_state_bytes()` 则能在失控运行演变成严重问题前使其失败。请参阅 [防护措施](https://queuety.inline0.dev/zh/docs/workflows/guardrails)。

## 步骤类型

Queuety 支持在工作流中使用十一个步骤类型：

- [顺序链](https://queuety.inline0.dev/zh/docs/workflows/chains)：使用 `->then()` 逐个执行
- [并行组](https://queuety.inline0.dev/zh/docs/workflows/parallel)：使用 `->parallel()` 并发执行
- [动态 For-Each](https://queuety.inline0.dev/zh/docs/workflows/for-each)：使用 `->for_each()` 处理运行时发现的分支任务，并明确完成语义
- [异步交接](https://queuety.inline0.dev/zh/docs/workflows/async-handoffs)：使用 `->start_workflows()` 或 `->start_agents()`，根据运行时发现的项目调度独立的子工作流
- [条件分支](https://queuety.inline0.dev/zh/docs/workflows/conditional)：使用命名步骤和 `_next_step`
- [重复](https://queuety.inline0.dev/zh/docs/workflows/repeats)：使用 `->repeat_until()` 和 `->repeat_while()` 创建指向较早命名步骤的持久回边
- [运行工作流](https://queuety.inline0.dev/zh/docs/workflows/run-workflows)：使用 `->run_workflow()` 进行嵌套组合
- [持久延迟](https://queuety.inline0.dev/zh/docs/workflows/delays)：使用 `->delay()` 实现基于时间的延迟
- [信号](https://queuety.inline0.dev/zh/docs/workflows/signals)：使用 `->wait_for_signal()`、`->wait_for_signals()`、`->wait_for_approval()`、`->wait_for_input()` 和 `->wait_for_decision()` 触发外部事件
- [工作流依赖](https://queuety.inline0.dev/zh/docs/workflows/dependencies)：使用 `->wait_for_workflow()`、`->wait_for_workflows()` 和 `->wait_for_agents()` 协调跨工作流任务
- [流式步骤](https://queuety.inline0.dev/zh/docs/workflows/streaming)：使用 `StreamingStep` 接口生成持久化的分块输出

## 常见模式

如果你正在构建自主或半自主系统，请从 [代理编排](https://queuety.inline0.dev/zh/docs/workflows/agent-orchestration) 开始。它展示了 `for_each()`、`start_agents()`、`wait_for_agents()`、`wait_for_workflow()` 和 `wait_for_decision()` 如何在实际的规划器/执行器和人工审核流程中协同工作。