# Suoratoistovaiheet

Suoratoistovaiheet antavat työnkulun tuottaa tulostetta vähitellen. Jokainen tuotettu osa kirjoitetaan heti tietokantaan, joten jos vaihe epäonnistuu suoratoiston aikana, jo tallennetut osat säilyvät ja ovat käytettävissä uudelleenyrityksessä.

Tästä on hyötyä LLM-suoratoistovastauksissa, suurten tiedostojen käsittelyssä ja kaikissa tilanteissa, joissa osittaista edistymistä ei pidä menettää.

## StreamingStep-rajapinta

Suoratoistovaihe toteuttaa kohteen `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;
}
```

| Metodi | Kuvaus |
|---|---|
| `stream( $state, $existing_chunks )` | Generaattori, joka tuottaa merkkijono-osia. Jokainen tuotto käynnistää välittömän tietokantakirjoituksen. |
| `on_complete( $chunks, $state )` | Kutsutaan suoratoiston päätyttyä. Saa kaikki osat (aiemmat ja uudet) ja palauttaa tiedot yhdistettäviksi työnkulun tilaan. |
| `config()` | Valinnaiset asetukset. Tukee avaimia `max_attempts` ja `backoff`; `backoff` saa strategiannimen (`exponential`, `linear`, `fixed`), sillä viiveiden luettelo hylätään tällä hetkellä ([#16](https://github.com/inline0/queuety/issues/16)). |

## Osien tallennus

Aina kun generaattorisi tuottaa arvon, Queuety kutsuu kohdetta `ChunkStore::append_chunk()` kirjoittaakseen sen taulukkoon `queuety_chunks`. Eräkirjoitusta tai puskurointia ei ole. Yksi tuotto vastaa yhtä INSERT-lauseketta.

Osariville tallennetaan arvot `job_id`, `workflow_id`, `step_index`, `chunk_index` (nollasta alkava) ja merkkijono `content`.

## Jatkaminen

Kun suoratoistovaihe yritetään uudelleen epäonnistumisen jälkeen, työntekijä lataa kaikki aiemmin tallennetut osat ja välittää ne parametrina `$existing_chunks`. Toteutuksesi päättää, miten suoritus jatkuu:

- **Ohita jo käsitellyt kohteet** tarkistamalla olemassa olevien osien määrä
- **Siirry sivutetussa kohteessa API** käyttämällä osien määrää sivun siirtymänä
- **Aloita alusta** ohittamalla arvo `$existing_chunks` kokonaan (aiemman yrityksen osat ovat edelleen tietokannassa ja lisätään alkuun)

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

## Sydämenlyönnit suoratoiston aikana

Pitkien suoratoistojen aikana tulee lähettää sydämenlyöntejä, jotta vanhentuneiden työtehtävien tunnistin ei ota tehtävää haltuunsa uudelleen. Kutsu kohdetta `Heartbeat::beat()` generaattorisilmukan sisällä:

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

Katso lisätietoja kohdasta [Sydämenlyönnit](https://queuety.inline0.dev/fi/docs/features/heartbeats) ja Heartbeat-rajapinnasta API.

## Osien siivous

Kun suoratoisto valmistuu onnistuneesti ja `on_complete()` palauttaa arvon, työntekijä kutsuu kohdetta `ChunkStore::clear_chunks()` poistaakseen kaikki työn tehtävän osat. Kohteen `on_complete()` yhdistetty tulos on pysyvä tietue. Osat ovat tarkoituksella väliaikaisia.

Jos vaihe epäonnistuu ja sitä yritetään uudelleen, osia ei poisteta. Ne välitetään seuraavalle yritykselle kohteen `$existing_chunks` kautta.

## Täydellinen esimerkki: LLM-suoratoisto

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

Käytä sitä työnkulussa kuten mitä tahansa muuta vaihetta:

```php
use Queuety\Queuety;

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

## Täydellinen esimerkki: tiedoston käsittely

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

## ChunkStore API

Luokka `ChunkStore` hallinnoi osien tallennusta. Käytä sitä kohteen `Queuety::chunk_store()` kautta.

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

| Metodi | Palauttaa | Kuvaus |
|---|---|---|
| `get_chunks( int $job_id )` | `string[]` | Hakee kaikki työn tehtävän osat osan indeksin mukaisessa järjestyksessä |
| `append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )` | `void` | Lisää yhden osan |
| `clear_chunks( int $job_id )` | `void` | Poistaa kaikki työn tehtävän osat |
| `chunk_count( int $job_id )` | `int` | Laskee työn tehtävän osien määrän |
| `get_accumulated( int $job_id )` | `string` | Yhdistää kaikki osat yhdeksi merkkijonoksi |

Yleensä kohdetta `ChunkStore` ei tarvitse kutsua suoraan. Työntekijä hoitaa tallennuksen ja siivouksen automaattisesti käsitellessään kohdetta `StreamingStep`. API on käytettävissä edistyneempiin käyttötapauksiin, kuten keskeneräisten osien tarkasteluun tai manuaaliseen siivoukseen.