# Étapes de diffusion en continu

Les étapes de diffusion en continu permettent à un workflow de produire progressivement des résultats. Chaque bloc généré est immédiatement écrit dans la base de données. Ainsi, si l’étape échoue en cours de diffusion, les blocs déjà enregistrés sont conservés et disponibles lors d’une nouvelle tentative.

C’est utile pour les réponses en continu des LLM, le traitement de fichiers volumineux et tout scénario où il ne faut pas perdre les progrès partiels.

## L’interface StreamingStep

Une étape de diffusion en continu implémente `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;
}
```

| Méthode | Description |
|---|---|
| `stream( $state, $existing_chunks )` | Générateur qui produit des blocs de chaîne de caractères. Chaque génération déclenche immédiatement une écriture en base de données. |
| `on_complete( $chunks, $state )` | Appelée à la fin de la diffusion. Reçoit tous les blocs (anciens et nouveaux) et renvoie les données à fusionner dans l’état du workflow. |
| `config()` | Configuration facultative. Accepte les clés `max_attempts` et `backoff` ; `backoff` prend un nom de stratégie (`exponential`, `linear`, `fixed`), car une liste de délais est actuellement supprimée ([#16](https://github.com/inline0/queuety/issues/16)). |

## Persistance des blocs

Chaque fois que votre générateur produit une valeur, Queuety appelle `ChunkStore::append_chunk()` pour l’écrire dans la table `queuety_chunks`. Il n’y a ni traitement par lots ni mise en mémoire tampon. Une génération équivaut à une insertion.

La ligne du bloc stocke le `job_id`, le `workflow_id`, le `step_index`, le `chunk_index` (à partir de zéro) et la chaîne `content`.

## Reprise

Lorsqu’une étape de diffusion en continu est relancée après un échec, le worker charge tous les blocs précédemment enregistrés et les transmet comme paramètre `$existing_chunks`. Votre implémentation détermine comment reprendre :

- **Ignorer les éléments déjà traités** en vérifiant le nombre de blocs existants
- **Se positionner dans un API paginé** en utilisant le nombre de blocs comme décalage de page
- **Recommencer depuis le début** en ignorant complètement `$existing_chunks` (les blocs de la tentative précédente sont toujours dans la base de données et seront ajoutés au début)

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

## Signaux de vie pendant la diffusion en continu

Les diffusions de longue durée doivent envoyer des signaux de vie pour éviter que le détecteur de tâches obsolètes ne récupère la tâche. Appelez `Heartbeat::beat()` dans la boucle de votre générateur :

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

Consultez [Signaux de vie](https://queuety.inline0.dev/fr/docs/features/heartbeats) pour en savoir plus sur le API Heartbeat.

## Nettoyage des blocs

Lorsque la diffusion se termine avec succès et que `on_complete()` renvoie une valeur, le worker appelle `ChunkStore::clear_chunks()` pour supprimer tous les blocs associés à la tâche. Le résultat fusionné de `on_complete()` constitue l’enregistrement durable. Les blocs sont temporaires par conception.

Si l’étape échoue et fait l’objet d’une nouvelle tentative, les blocs ne sont pas supprimés. Ils sont transmis à nouveau via `$existing_chunks` lors de la tentative suivante.

## Exemple complet : diffusion en continu avec un 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 ];
    }
}
```

Utilisez-le dans un workflow comme n’importe quelle autre étape :

```php
use Queuety\Queuety;

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

## Exemple complet : traitement de fichiers

```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` gère la persistance des blocs. Accédez-y via `Queuety::chunk_store()`.

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

| Méthode | Renvoie | Description |
|---|---|---|
| `get_chunks( int $job_id )` | `string[]` | Récupère tous les blocs d’une tâche, classés par index de bloc |
| `append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )` | `void` | Ajoute un bloc |
| `clear_chunks( int $job_id )` | `void` | Supprime tous les blocs d’une tâche |
| `chunk_count( int $job_id )` | `int` | Compte le nombre de blocs d’une tâche |
| `get_accumulated( int $job_id )` | `string` | Concatène tous les blocs en une seule chaîne |

En général, vous n’avez pas besoin d’appeler `ChunkStore` directement. Le worker gère automatiquement la persistance et le nettoyage lors du traitement d’un `StreamingStep`. Le API est disponible pour les cas avancés, par exemple pour examiner des blocs en cours ou effectuer un nettoyage manuel.