# Srautiniai veiksmai

Srautiniai veiksmai leidžia darbo sekai kurti išvestį dalimis. Kiekviena pateikta dalis iškart įrašoma į duomenų bazę, todėl, jei veiksmas nepavyksta srautiniam perdavimui dar nepasibaigus, jau išsaugotos dalys išlieka ir jas galima naudoti bandant dar kartą.

Tai naudinga LLM srautinio atsakymo, didelių failų apdorojimo ir bet kokiais kitais atvejais, kai dalinės pažangos nereikėtų prarasti.

## StreamingStep sąsaja

Srautinis veiksmas įgyvendina `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;
}
```

| Metodas | Aprašymas |
|---|---|
| `stream( $state, $existing_chunks )` | Generatorius, pateikiantis eilutės dalis. Kiekviena pateikta dalis iškart įrašoma į DB. |
| `on_complete( $chunks, $state )` | Iškviečiamas srautiniam perdavimui pasibaigus. Gauna visas dalis (ankstesnes ir naujas) ir grąžina duomenis, kurie sujungiami su darbo sekos būsena. |
| `config()` | Pasirenkama konfigūracija. Palaiko raktus `max_attempts` ir `backoff`; `backoff` priima strategijos pavadinimą (`exponential`, `linear`, `fixed`), nes delsų sąrašas šiuo metu atmetamas ([#16](https://github.com/inline0/queuety/issues/16)). |

## Kaip išsaugomos dalys

Kiekvieną kartą, kai generatorius pateikia reikšmę, Queuety iškviečia `ChunkStore::append_chunk()` ir įrašo ją į lentelę `queuety_chunks`. Duomenys negrupuojami ir nekaupiami buferyje. Viena pateikta dalis atitinka vieną INSERT.

Dalos eilutėje saugomos reikšmės `job_id`, `workflow_id`, `step_index`, `chunk_index` (skaičiuojant nuo nulio) ir eilutė `content`.

## Tęsimas

Pakartotinai vykdant srautinį veiksmą po nesėkmės, darbuotojas įkelia visas anksčiau išsaugotas dalis ir perduoda jas kaip parametrą `$existing_chunks`. Jūsų realizacija nusprendžia, kaip tęsti:

- **Praleiskite jau apdorotus elementus** tikrindami esamų dalių skaičių
- **Nurodykite vietą puslapiu suskaidytame API** naudodami dalių skaičių kaip puslapio poslinkį
- **Pradėkite iš naujo** visiškai ignoruodami `$existing_chunks` (ankstesnio bandymo dalys vis tiek lieka DB ir bus pateiktos pradžioje)

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

## Širdies plakimo signalai perduodant srautą

Ilgai veikiantys srautai turėtų siųsti širdies plakimo signalus, kad pasenusių užduočių aptikimo priemonė neatimtų užduoties. Iškvieskite `Heartbeat::beat()` generatoriaus cikle:

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

Daugiau informacijos rasite puslapyje [Širdies plakimo signalai](https://queuety.inline0.dev/lt/docs/features/heartbeats) apie metodą Heartbeat API.

## Dalių išvalymas

Sėkmingai baigus srautinį perdavimą ir grąžinus `on_complete()`, darbuotojas iškviečia `ChunkStore::clear_chunks()`, kad ištrintų visas užduoties dalis. Sujungtas `on_complete()` rezultatas yra patikimas įrašas. Dalys pagal paskirtį yra laikinos.

Jei veiksmas nepavyksta ir paleidžiamas iš naujo, dalys neištrinamos. Kito bandymo metu jos perduodamos per `$existing_chunks`.

## Visas pavyzdys: LLM srautinis perdavimas

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

Naudokite tai darbo sekoje kaip ir bet kurį kitą veiksmą:

```php
use Queuety\Queuety;

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

## Visas pavyzdys: failų apdorojimas

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

Klasė `ChunkStore` tvarko dalių išsaugojimą. Ją pasieksite per `Queuety::chunk_store()`.

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

| Metodas | Grąžina | Aprašymas |
|---|---|---|
| `get_chunks( int $job_id )` | `string[]` | Grąžina visas užduoties dalis, surikiuotas pagal dalies indeksą |
| `append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )` | `void` | Prideda vieną dalį |
| `clear_chunks( int $job_id )` | `void` | Ištrina visas užduoties dalis |
| `chunk_count( int $job_id )` | `int` | Suskaičiuoja užduoties dalis |
| `get_accumulated( int $job_id )` | `string` | Sujungia visas dalis į vieną eilutę |

Paprastai `ChunkStore` tiesiogiai kviesti nereikia. Darbinis procesas automatiškai pasirūpina duomenų išsaugojimu ir išvalymu, kai apdoroja `StreamingStep`. API galima naudoti pažangesniais atvejais, pavyzdžiui, norint peržiūrėti dar apdorojamas dalis arba išvalyti jas rankiniu būdu.