# Pași de transmitere în flux

Pașii de transmitere în flux permit unui flux de lucru să producă rezultate treptat. Fiecare fragment generat este scris imediat în baza de date, astfel încât, dacă pasul eșuează în timpul transmiterii, fragmentele deja persistate se păstrează și sunt disponibile la reîncercare.

Acest lucru este util pentru răspunsurile transmise în flux de LLM, procesarea fișierelor mari și orice situație în care progresul parțial nu trebuie pierdut.

## Interfața StreamingStep

Un pas de transmitere în flux implementează `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;
}
```

| Metodă | Descriere |
|---|---|
| `stream( $state, $existing_chunks )` | Generator care produce fragmente de text. Fiecare valoare generată declanșează imediat o scriere în DB. |
| `on_complete( $chunks, $state )` | Se apelează după încheierea fluxului. Primește toate fragmentele (anterioare + noi) și returnează date de îmbinat cu starea fluxului de lucru. |
| `config()` | Configurare opțională. Acceptă cheile `max_attempts` și `backoff`; `backoff` primește un nume de strategie (`exponential`, `linear`, `fixed`), deoarece o listă de întârzieri este eliminată în prezent ([#16](https://github.com/inline0/queuety/issues/16)). |

## Cum se persistă fragmentele

De fiecare dată când generatorul dumneavoastră produce o valoare, Queuety apelează `ChunkStore::append_chunk()` pentru a o scrie în tabelul `queuety_chunks`. Nu există grupare sau tamponare. O valoare generată înseamnă o instrucțiune INSERT.

Rândul fragmentului stochează `job_id`, `workflow_id`, `step_index`, `chunk_index` (indexare de la zero) și șirul `content`.

## Posibilitatea de reluare

Când un pas de transmitere în flux este reîncercat după o eroare, lucrătorul încarcă toate fragmentele persistate anterior și le transmite ca parametru `$existing_chunks`. Implementarea dumneavoastră decide cum să reia:

- **Omiteți elementele deja procesate** verificând numărul fragmentelor existente
- **Avansați într-un API paginat** folosind numărul fragmentelor drept decalajul paginii
- **Reporniți de la zero** ignorând complet `$existing_chunks` (fragmentele din încercarea anterioară sunt încă în DB și vor fi adăugate la început)

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

## Semnale de puls în timpul transmiterii în flux

Fluxurile de lungă durată ar trebui să trimită semnale de puls pentru a împiedica detectorul joburilor învechite să recupereze jobul. Apelați `Heartbeat::beat()` în bucla generatorului:

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

Consultați [Semnale de puls](https://queuety.inline0.dev/ro/docs/features/heartbeats) pentru detalii despre API de puls.

## Curățarea fragmentelor

Când fluxul se încheie cu succes și `on_complete()` returnează un rezultat, lucrătorul apelează `ChunkStore::clear_chunks()` pentru a șterge toate fragmentele jobului. Rezultatul îmbinat din `on_complete()` este înregistrarea durabilă. Fragmentele sunt temporare prin proiectare.

Dacă pasul eșuează și este reîncercat, fragmentele nu sunt șterse. Ele sunt transmise din nou prin `$existing_chunks` la următoarea încercare.

## Exemplu complet: transmitere în flux pentru 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 ];
    }
}
```

Folosiți-l într-un flux de lucru ca pe orice alt pas:

```php
use Queuety\Queuety;

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

## Exemplu complet: procesarea fișierelor

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

## Metodele ChunkStore API

Clasa `ChunkStore` gestionează persistența fragmentelor. Accesați-o prin `Queuety::chunk_store()`.

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

| Metodă | Returnează | Descriere |
|---|---|---|
| `get_chunks( int $job_id )` | `string[]` | Preia toate fragmentele unui job, ordonate după indexul fragmentului |
| `append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )` | `void` | Adaugă un singur fragment |
| `clear_chunks( int $job_id )` | `void` | Șterge toate fragmentele unui job |
| `chunk_count( int $job_id )` | `int` | Numără fragmentele unui job |
| `get_accumulated( int $job_id )` | `string` | Concatenează toate fragmentele într-un singur șir |

În mod normal, nu este nevoie să apelați direct `ChunkStore`. Lucrătorul gestionează automat persistența și curățarea când procesează un `StreamingStep`. API este disponibil pentru cazuri avansate, precum inspectarea fragmentelor în curs sau curățarea manuală.