Étapes de diffusion en continu

Afficher au format Markdown Modifier sur GitHub Documentation de 2cc7b2b

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).

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