# ストリーミングステップ

ストリーミングステップでは、ワークフローが出力を逐次生成できます。yieldされた各チャンクはすぐにデータベースへ書き込まれるため、ステップがストリームの途中で失敗しても、すでに保存されたチャンクは残り、再試行時に利用できます。

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 )` | 文字列チャンクをyieldするジェネレーター。yieldごとに即座にDBへ書き込みます。 |
| `on_complete( $chunks, $state )` | ストリーム終了後に呼び出されます。すべてのチャンク（以前の分と新しい分）を受け取り、ワークフロー状態にマージするデータを返します。 |
| `config()` | 任意の設定。`max_attempts`と`backoff`のキーに対応します。遅延のリストは現在破棄されるため、`backoff`には戦略名（`exponential`、`linear`、`fixed`）を指定します（[#16](https://github.com/inline0/queuety/issues/16)）。 |

## チャンクの保存方法

ジェネレーターが値をyieldするたびに、Queuetyは`ChunkStore::append_chunk()`を呼び出して`queuety_chunks`テーブルに書き込みます。バッチ処理やバッファリングはありません。yield 1回につきINSERT 1回です。

チャンク行には、`job_id`、`workflow_id`、`step_index`、`chunk_index`（0始まり）、文字列`content`が保存されます。

## 再開可能性

ストリーミングステップが失敗後に再試行されると、ワーカーは保存済みのすべてのチャンクを読み込み、`$existing_chunks`パラメーターとして渡します。再開方法は実装によって決まります。

- 既存チャンクの数を確認して、**処理済みの項目をスキップする**
- チャンク数をページオフセットとして使い、ページ分割されたAPI内の位置を**オフセットする**
- `$existing_chunks`を完全に無視して、**最初からやり直す**（前回の試行のチャンクは引き続きDBに残り、先頭に追加されます）

```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/ja/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` | 1つのチャンクを追加 |
| `clear_chunks( int $job_id )` | `void` | ジョブのすべてのチャンクを削除 |
| `chunk_count( int $job_id )` | `int` | ジョブのチャンク数を取得 |
| `get_accumulated( int $job_id )` | `string` | すべてのチャンクを1つの文字列に連結 |

通常、`ChunkStore`を直接呼び出す必要はありません。ワーカーは`StreamingStep`の処理時に、保存とクリーンアップを自動で行います。APIは、進行中のチャンクの確認や手動クリーンアップなど、高度な用途に利用できます。