# 流式步骤

流式步骤可让工作流逐步生成输出。每个生成的块都会立即写入数据库，因此如果步骤在流式处理中途失败，已持久化的块仍会保留，并可在重试时使用。

这适用于 LLM 流式响应、大型文件处理，以及任何不应丢失部分进度的场景。

## StreamingStep 接口

流式步骤需要实现 `Queuety\Contracts\StreamingStep`：

```php
namespace Queuety\Contracts;

interface StreamingStep {
    public function stream( array $state, array $existing_chunks = [] ): \Generator;
    public function on_complete( array $chunks, array $state ): array;
    public function config(): array;
}
```

| 方法 | 描述 |
|---|---|
| `stream( $state, $existing_chunks )` | 生成字符串块的生成器。每次生成都会立即写入数据库。 |
| `on_complete( $chunks, $state )` | 在流结束后调用。接收所有块（之前的和新的），并返回要合并到工作流状态的数据。 |
| `config()` | 可选配置。支持 `max_attempts` 和 `backoff` 键；`backoff` 接收策略名称（`exponential`、`linear`、`fixed`），因为延迟列表目前会被丢弃（[#16](https://github.com/inline0/queuety/issues/16)）。 |

## 块的持久化方式

每当生成器生成一个值，Queuety 就会调用 `ChunkStore::append_chunk()` 将其写入 `queuety_chunks` 表。不会进行批处理或缓冲。每次生成都会执行一次 INSERT。

块记录会存储 `job_id`、`workflow_id`、`step_index`、`chunk_index`（从零开始），以及字符串 `content`。

## 可恢复性

流式步骤失败后重试时，工作进程会加载所有之前持久化的块，并将它们作为 `$existing_chunks` 参数传入。由你的实现决定如何恢复：

- **跳过已处理的项目**：检查现有块的数量
- **在分页的 API 中定位**：使用块数量作为页偏移量
- **从头开始**：完全忽略 `$existing_chunks`（上次尝试的块仍保留在数据库中，并会置于前面）

```php
public function stream( array $state, array $existing_chunks = [] ): \Generator {
    $offset = count( $existing_chunks );

    $pages = $this->client->paginate( $state['query'], offset: $offset );

    foreach ( $pages as $page ) {
        yield json_encode( $page );
    }
}
```

## 流式处理期间的心跳

长时间运行的流应发送心跳，以防止过期任务检测器回收任务。在生成器循环中调用 `Heartbeat::beat()`：

```php
use Queuety\Heartbeat;

public function stream( array $state, array $existing_chunks = [] ): \Generator {
    foreach ( $this->client->stream( $state['prompt'] ) as $i => $chunk ) {
        Heartbeat::beat( [ 'chunks' => $i + 1 ] );
        yield $chunk;
    }
}
```

详情请参阅 [心跳](https://queuety.inline0.dev/zh/docs/features/heartbeats)，了解 Heartbeat API。

## 清理块

流成功完成且 `on_complete()` 返回后，工作进程会调用 `ChunkStore::clear_chunks()` 删除该任务的所有块。来自 `on_complete()` 的合并结果是持久记录。块在设计上是临时数据。

如果步骤失败并重试，块不会被删除。下一次尝试时，它们会通过 `$existing_chunks` 传回。

## 完整示例：LLM 流式处理

```php
use Queuety\Contracts\StreamingStep;
use Queuety\Heartbeat;

class StreamLLMResponse implements StreamingStep {

    public function stream( array $state, array $existing_chunks = [] ): \Generator {
        // If we have existing chunks from a previous attempt, we can skip
        // re-requesting. The on_complete method will concatenate everything.
        if ( ! empty( $existing_chunks ) ) {
            return;
        }

        $response = OpenAI::chat()->createStreamed( [
            'model'    => 'gpt-4',
            'messages' => $state['messages'],
        ] );

        foreach ( $response as $i => $chunk ) {
            $text = $chunk->choices[0]->delta->content ?? '';
            if ( '' !== $text ) {
                Heartbeat::beat( [ 'tokens' => $i ] );
                yield $text;
            }
        }
    }

    public function on_complete( array $chunks, array $state ): array {
        return [
            'llm_response' => implode( '', $chunks ),
        ];
    }

    public function config(): array {
        return [ 'max_attempts' => 5 ];
    }
}
```

像使用其他步骤一样，在工作流中使用它：

```php
use Queuety\Queuety;

Queuety::workflow( 'ai_chat' )
    ->then( BuildPromptHandler::class )
    ->then( StreamLLMResponse::class )
    ->then( SaveResponseHandler::class )
    ->dispatch( [ 'user_input' => $input ] );
```

## 完整示例：文件处理

```php
use Queuety\Contracts\StreamingStep;
use Queuety\Heartbeat;

class ProcessCSVRows implements StreamingStep {

    public function stream( array $state, array $existing_chunks = [] ): \Generator {
        $file   = fopen( $state['csv_path'], 'r' );
        $offset = count( $existing_chunks );
        $row    = 0;

        // Skip the header.
        fgetcsv( $file );

        while ( $line = fgetcsv( $file ) ) {
            $row++;

            // Skip rows already processed in a previous attempt.
            if ( $row <= $offset ) {
                continue;
            }

            $result = $this->process_row( $line );
            Heartbeat::beat( [ 'row' => $row ] );
            yield json_encode( $result );
        }

        fclose( $file );
    }

    public function on_complete( array $chunks, array $state ): array {
        $results = array_map( 'json_decode', $chunks );
        return [
            'processed_count' => count( $results ),
            'results'         => $results,
        ];
    }

    public function config(): array {
        return [];
    }

    private function process_row( array $line ): array {
        // Your row processing logic here.
        return [ 'id' => $line[0], 'status' => 'ok' ];
    }
}
```

## 分块存储 API

`ChunkStore` 类负责管理块的持久化。通过 `Queuety::chunk_store()` 访问它。

```php
$store = Queuety::chunk_store();
```

| 方法 | 返回值 | 描述 |
|---|---|---|
| `get_chunks( int $job_id )` | `string[]` | 获取某个任务的所有块，并按块索引排序 |
| `append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )` | `void` | 追加单个块 |
| `clear_chunks( int $job_id )` | `void` | 删除某个任务的所有块 |
| `chunk_count( int $job_id )` | `int` | 统计某个任务的块数量 |
| `get_accumulated( int $job_id )` | `string` | 将所有块拼接为一个字符串 |

通常无需直接调用 `ChunkStore`。处理 `StreamingStep` 时，工作进程会自动处理持久化和清理。API 可用于检查处理中块或手动清理等高级场景。