# Darbplūsmas straumēšanas darbības

Straumēšanas darbības ļauj darbplūsmai pakāpeniski ģenerēt izvadi. Katrs nodotais fragments tiek uzreiz ierakstīts datubāzē, tāpēc, ja darbība neizdodas straumēšanas laikā, jau saglabātie fragmenti tiek saglabāti un ir pieejami atkārtotā mēģinājumā.

Tas noder LLM straumēšanas atbildēm, lielu failu apstrādei un jebkurā situācijā, kad nedrīkst zaudēt daļēju progresu.

## StreamingStep saskarne

Straumēšanas darbība implementē `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 | Apraksts |
|---|---|
| `stream( $state, $existing_chunks )` | Ģenerators, kas nodod virknes fragmentus. Katra nodošana izraisa tūlītēju DB ierakstu. |
| `on_complete( $chunks, $state )` | Tiek izsaukta pēc straumes pabeigšanas. Saņem visus fragmentus (iepriekšējos un jaunos) un atgriež datus, ko apvienot ar darbplūsmas stāvokli. |
| `config()` | Neobligāta konfigurācija. Atbalsta atslēgas `max_attempts` un `backoff`; `backoff` pieņem stratēģijas nosaukumu (`exponential`, `linear`, `fixed`), jo aizkavju saraksts pašlaik tiek atmests ([#16](https://github.com/inline0/queuety/issues/16)). |

## Kā tiek saglabāti fragmenti

Katru reizi, kad ģenerators nodod vērtību, Queuety izsauc `ChunkStore::append_chunk()`, lai ierakstītu to tabulā `queuety_chunks`. Netiek veikta pakešu apstrāde vai buferēšana. Viena nodošana atbilst vienam INSERT.

Fragmenta rindā tiek saglabātas vērtības `job_id`, `workflow_id`, `step_index`, `chunk_index` (sākot no nulles) un virkne `content`.

## Izpildes atsākšana

Ja straumēšanas darbību mēģina atkārtoti pēc kļūmes, izpildītājs ielādē visus iepriekš saglabātos fragmentus un nodod tos kā parametru `$existing_chunks`. Jūs izvēlaties, kā atsākt darbu:

- **Izlaist jau apstrādātos vienumus**, pārbaudot esošo fragmentu skaitu
- **Pāriet uz lappusēm sadalītā API nobīdi**, izmantojot fragmentu skaitu kā lappuses nobīdi
- **Sākt no jauna**, pilnībā ignorējot `$existing_chunks` (iepriekšējā mēģinājuma fragmenti joprojām ir DB un tiks pievienoti sākumā)

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

## Sirdspuksti straumēšanas laikā

Ilgstošām straumēm jāsūta sirdspuksti, lai novērstu novecojušu darbu noteicēja mēģinājumus pārņemt darbu. Izsauciet `Heartbeat::beat()` ģeneratora ciklā:

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

Plašāku informāciju par Heartbeat API skatiet sadaļā [Sirdspuksti](https://queuety.inline0.dev/lv/docs/features/heartbeats).

## Fragmentu tīrīšana

Kad straume tiek veiksmīgi pabeigta un `on_complete()` atgriežas, izpildītājs izsauc `ChunkStore::clear_chunks()`, lai dzēstu visus darba fragmentus. Apvienotais rezultāts no `on_complete()` ir pastāvīgais ieraksts. Fragmenti pēc konstrukcijas ir īslaicīgi.

Ja darbība neizdodas un tiek mēģināta atkārtoti, fragmenti netiek dzēsti. Nākamajā mēģinājumā tie tiek nodoti, izmantojot `$existing_chunks`.

## Pilns piemērs: LLM straumēšana

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

Izmantojiet to darbplūsmā tāpat kā jebkuru citu darbību:

```php
use Queuety\Queuety;

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

## Pilns piemērs: failu apstrāde

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

Klase `ChunkStore` pārvalda fragmentu saglabāšanu. Tai var piekļūt, izmantojot `Queuety::chunk_store()`.

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

| Metode | Atgriež | Apraksts |
|---|---|---|
| `get_chunks( int $job_id )` | `string[]` | Iegūst visus darba fragmentus, sakārtotus pēc fragmenta indeksa |
| `append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )` | `void` | Pievieno vienu fragmentu |
| `clear_chunks( int $job_id )` | `void` | Dzēš visus darba fragmentus |
| `chunk_count( int $job_id )` | `int` | Saskaita darba fragmentus |
| `get_accumulated( int $job_id )` | `string` | Savieno visus fragmentus vienā virknē |

Parasti `ChunkStore` nav jāizsauc tieši. Apstrādājot `StreamingStep`, izpildītājs automātiski parūpējas par saglabāšanu un tīrīšanu. API ir pieejams īpašiem lietojumiem, piemēram, apstrādes gaitā esošo fragmentu pārbaudei vai manuālai tīrīšanai.