# Passaggi di streaming

I passaggi di streaming consentono a un flusso di lavoro di produrre output in modo incrementale. Ogni blocco generato viene scritto subito nel database, così, se il passaggio si interrompe durante lo streaming, i blocchi già salvati restano disponibili al nuovo tentativo.

È utile per le risposte in streaming degli LLM, l'elaborazione di file di grandi dimensioni e ogni scenario in cui non si vuole perdere il lavoro parziale.

## L'interfaccia StreamingStep

Un passaggio di streaming implementa `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;
}
```

| Metodo | Descrizione |
|---|---|
| `stream( $state, $existing_chunks )` | Generatore che produce blocchi di stringhe. Ogni blocco generato attiva una scrittura immediata nel DB. |
| `on_complete( $chunks, $state )` | Chiamato al termine dello stream. Riceve tutti i blocchi (precedenti e nuovi) e restituisce i dati da unire allo stato del workflow. |
| `config()` | Configurazione opzionale. Supporta le chiavi `max_attempts` e `backoff`; `backoff` accetta il nome di una strategia (`exponential`, `linear`, `fixed`), poiché un elenco di ritardi viene attualmente ignorato ([#16](https://github.com/inline0/queuety/issues/16)). |

## Come vengono salvati i blocchi

Ogni volta che il generatore produce un valore, Queuety chiama `ChunkStore::append_chunk()` per scriverlo nella tabella `queuety_chunks`. Non viene effettuato alcun raggruppamento o buffering. Ogni valore generato corrisponde a un INSERT.

La riga del blocco contiene `job_id`, `workflow_id`, `step_index`, `chunk_index` (con indice a partire da zero) e la stringa `content`.

## Ripresa

Quando un passaggio di streaming viene ritentato dopo un errore, il worker carica tutti i blocchi salvati in precedenza e li passa come parametro `$existing_chunks`. Spetta alla vostra implementazione decidere come riprendere:

- **Saltare gli elementi già elaborati** controllando il numero di blocchi esistenti
- **Usare il numero di blocchi come offset della pagina** per scorrere un API paginato
- **Ricominciare da zero** ignorando del tutto `$existing_chunks` (i blocchi del tentativo precedente sono ancora nel DB e verranno anteposti)

```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 durante lo streaming

Gli stream di lunga durata dovrebbero inviare heartbeat per impedire al rilevatore di job obsoleti di riprendere il job. Chiamate `Heartbeat::beat()` all'interno del ciclo del generatore:

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

Per i dettagli, consultate [Heartbeat](https://queuety.inline0.dev/it/docs/features/heartbeats) del API.

## Pulizia dei blocchi

Quando lo stream termina correttamente e `on_complete()` restituisce un risultato, il worker chiama `ChunkStore::clear_chunks()` per eliminare tutti i blocchi del job. Il risultato unito di `on_complete()` costituisce il record durevole. I blocchi sono temporanei per progettazione.

Se il passaggio fallisce e viene ritentato, i blocchi non vengono eliminati. Vengono passati di nuovo tramite `$existing_chunks` al tentativo successivo.

## Esempio completo: 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 ];
    }
}
```

Usatelo in un flusso di lavoro come qualsiasi altro passaggio:

```php
use Queuety\Queuety;

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

## Esempio completo: elaborazione di 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

La classe `ChunkStore` gestisce il salvataggio dei blocchi. Vi si accede tramite `Queuety::chunk_store()`.

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

| Metodo | Restituisce | Descrizione |
|---|---|---|
| `get_chunks( int $job_id )` | `string[]` | Recupera tutti i blocchi di un job, ordinati per indice del blocco |
| `append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )` | `void` | Aggiunge un singolo blocco |
| `clear_chunks( int $job_id )` | `void` | Elimina tutti i blocchi di un job |
| `chunk_count( int $job_id )` | `int` | Conta il numero di blocchi di un job |
| `get_accumulated( int $job_id )` | `string` | Concatena tutti i blocchi in un'unica stringa |

Di norma non è necessario chiamare direttamente `ChunkStore`. Il worker gestisce automaticamente il salvataggio e la pulizia durante l'elaborazione di un `StreamingStep`. API è disponibile per casi d'uso avanzati, ad esempio per esaminare i blocchi in corso o eseguire una pulizia manuale.