# Langkah Streaming

Langkah streaming memungkinkan alur kerja menghasilkan output secara bertahap. Setiap potongan yang dihasilkan langsung ditulis ke database, sehingga jika langkah gagal di tengah streaming, potongan yang sudah tersimpan tetap ada dan tersedia saat dicoba lagi.

Ini berguna untuk respons streaming LLM, pemrosesan file besar, dan skenario apa pun ketika progres parsial tidak boleh hilang.

## Antarmuka StreamingStep

Langkah streaming mengimplementasikan `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;
}
```

| Metode | Deskripsi |
|---|---|
| `stream( $state, $existing_chunks )` | Generator yang menghasilkan potongan string. Setiap hasil langsung memicu penulisan ke DB. |
| `on_complete( $chunks, $state )` | Dipanggil setelah streaming selesai. Menerima semua potongan (sebelumnya + baru) dan mengembalikan data untuk digabungkan ke status alur kerja. |
| `config()` | Konfigurasi opsional. Mendukung kunci `max_attempts` dan `backoff`; `backoff` menerima nama strategi (`exponential`, `linear`, `fixed`), karena daftar jeda saat ini diabaikan ([#16](https://github.com/inline0/queuety/issues/16)). |

## Cara potongan disimpan

Setiap kali generator Anda menghasilkan nilai, Queuety memanggil `ChunkStore::append_chunk()` untuk menulisnya ke tabel `queuety_chunks`. Tidak ada pengelompokan atau penyanggaan. Satu hasil berarti satu INSERT.

Baris potongan menyimpan `job_id`, `workflow_id`, `step_index`, `chunk_index` (dimulai dari nol), dan string `content`.

## Kemampuan untuk dilanjutkan

Saat langkah streaming dicoba lagi setelah gagal, worker memuat semua potongan yang sebelumnya tersimpan dan meneruskannya sebagai parameter `$existing_chunks`. Implementasi Anda menentukan cara melanjutkan:

- **Lewati item yang sudah diproses** dengan memeriksa jumlah potongan yang ada
- **Lanjutkan dari offset pada API berpaginasi** dengan menggunakan jumlah potongan sebagai offset halaman
- **Mulai ulang dari awal** dengan mengabaikan `$existing_chunks` sepenuhnya (potongan dari percobaan sebelumnya tetap ada di DB dan akan ditambahkan di awal)

```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 selama streaming

Streaming yang berjalan lama harus mengirim heartbeat agar pendeteksi job usang tidak mengambil alih job. Panggil `Heartbeat::beat()` di dalam loop generator Anda:

```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;
    }
}
```

Lihat [Heartbeat](https://queuety.inline0.dev/id/docs/features/heartbeats) untuk detail tentang API Heartbeat.

## Pembersihan potongan

Saat streaming berhasil diselesaikan dan `on_complete()` mengembalikan hasil, worker memanggil `ChunkStore::clear_chunks()` untuk menghapus semua potongan untuk job tersebut. Hasil gabungan dari `on_complete()` menjadi catatan tahan lama. Potongan memang dirancang untuk bersifat sementara.

Jika langkah gagal dan dicoba lagi, potongan tidak dihapus. Potongan tersebut diteruskan kembali melalui `$existing_chunks` pada percobaan berikutnya.

## Contoh lengkap: streaming 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 ];
    }
}
```

Gunakan dalam alur kerja seperti langkah lainnya:

```php
use Queuety\Queuety;

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

## Contoh lengkap: pemrosesan file

```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

Kelas `ChunkStore` mengelola penyimpanan potongan. Akses melalui `Queuety::chunk_store()`.

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

| Metode | Mengembalikan | Deskripsi |
|---|---|---|
| `get_chunks( int $job_id )` | `string[]` | Mengambil semua potongan untuk sebuah job, diurutkan berdasarkan indeks potongan |
| `append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )` | `void` | Menambahkan satu potongan |
| `clear_chunks( int $job_id )` | `void` | Menghapus semua potongan untuk sebuah job |
| `chunk_count( int $job_id )` | `int` | Menghitung jumlah potongan untuk sebuah job |
| `get_accumulated( int $job_id )` | `string` | Menggabungkan semua potongan menjadi satu string |

Biasanya Anda tidak perlu memanggil `ChunkStore` secara langsung. Worker menangani penyimpanan dan pembersihan secara otomatis saat memproses `StreamingStep`. API tersedia untuk kasus penggunaan tingkat lanjut, seperti memeriksa potongan yang sedang diproses atau melakukan pembersihan manual.