# Kroky so streamovaním

Kroky so streamovaním umožňujú pracovným postupom vytvárať výstup postupne. Každá odovzdaná časť sa okamžite zapíše do databázy, takže ak krok zlyhá uprostred streamovania, už uložené časti zostanú zachované a budú dostupné pri opakovanom pokuse.

Hodí sa to pri streamovaných odpovediach LLM, spracovaní veľkých súborov a v každom prípade, keď sa nesmie stratiť čiastočný pokrok.

## Rozhranie StreamingStep

Krok so streamovaním implementuje `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;
}
```

| Metóda | Opis |
|---|---|
| `stream( $state, $existing_chunks )` | Generátor, ktorý odovzdáva textové časti. Každé odovzdanie sa okamžite zapíše do DB. |
| `on_complete( $chunks, $state )` | Zavolá sa po skončení streamu. Dostane všetky časti (predchádzajúce aj nové) a vráti údaje, ktoré sa zlúčia so stavom pracovného postupu. |
| `config()` | Voliteľná konfigurácia. Podporuje kľúče `max_attempts` a `backoff`; `backoff` prijíma názov stratégie (`exponential`, `linear`, `fixed`), pretože zoznam oneskorení sa momentálne zahadzuje ([#16](https://github.com/inline0/queuety/issues/16)). |

## Ako sa ukladajú časti

Vždy, keď váš generátor odovzdá hodnotu, Queuety zavolá `ChunkStore::append_chunk()` a zapíše ju do tabuľky `queuety_chunks`. Nedochádza k dávkovaniu ani vyrovnávacej pamäti. Jedno odovzdanie znamená jeden príkaz INSERT.

Záznam časti obsahuje `job_id`, `workflow_id`, `step_index`, `chunk_index` (od nuly) a textový reťazec `content`.

## Možnosť obnovenia

Pri opakovanom spustení kroku so streamovaním po zlyhaní pracovník načíta všetky predtým uložené časti a odovzdá ich ako parameter `$existing_chunks`. Vaša implementácia určí, ako pokračovať:

- **Preskočiť už spracované položky** kontrolou počtu existujúcich častí
- **Preskočiť na príslušnú pozíciu v stránkovanom API** pomocou počtu častí ako posunu stránky
- **Začať od začiatku** úplným ignorovaním `$existing_chunks` (časti z predchádzajúceho pokusu zostávajú v DB a pripoja sa na začiatok)

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

## Signály aktivity počas streamovania

Dlho trvajúce streamy by mali odosielať signály aktivity, aby detektor neaktívnych úloh úlohu neprevzal. Zavolajte `Heartbeat::beat()` vo vnútri cyklu generátora:

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

Podrobnosti o rozhraní Heartbeat nájdete v téme [Signály aktivity](https://queuety.inline0.dev/sk/docs/features/heartbeats); pozrite si rozhranie API.

## Čistenie častí

Keď sa stream úspešne dokončí a `on_complete()` vráti výsledok, pracovník zavolá `ChunkStore::clear_chunks()` a odstráni všetky časti úlohy. Zlúčený výsledok z `on_complete()` je trvalým záznamom. Časti sú zámerne dočasné.

Ak krok zlyhá a spustí sa znova, časti sa neodstránia. Pri ďalšom pokuse sa odovzdajú cez `$existing_chunks`.

## Úplný príklad: streamovanie 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 ];
    }
}
```

Použite ho v pracovnom postupe ako ktorýkoľvek iný krok:

```php
use Queuety\Queuety;

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

## Úplný príklad: spracovanie súboru

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

Trieda `ChunkStore` spravuje ukladanie častí. Pristupujte k nej cez `Queuety::chunk_store()`.

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

| Metóda | Vráti | Opis |
|---|---|---|
| `get_chunks( int $job_id )` | `string[]` | Načíta všetky časti úlohy zoradené podľa indexu časti |
| `append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )` | `void` | Pripojí jednu časť |
| `clear_chunks( int $job_id )` | `void` | Odstráni všetky časti úlohy |
| `chunk_count( int $job_id )` | `int` | Spočíta počet častí úlohy |
| `get_accumulated( int $job_id )` | `string` | Spojí všetky časti do jedného reťazca |

`ChunkStore` zvyčajne nemusíte volať priamo. Worker sa pri spracovaní `StreamingStep` automaticky postará o uchovávanie údajov a čistenie. API je k dispozícii na pokročilé použitie, napríklad na kontrolu práve spracúvaných častí alebo manuálne čistenie.