Pasos de transmisión

Ver como Markdown Editar en GitHub Documentación de 2cc7b2b

Los pasos de transmisión permiten que un flujo de trabajo genere resultados de forma incremental. Cada fragmento generado se escribe de inmediato en la base de datos, de modo que, si el paso falla durante la transmisión, los fragmentos ya persistidos se conservan y están disponibles al reintentar.

Esto resulta útil para respuestas transmitidas por LLM, el procesamiento de archivos grandes y cualquier situación en la que no deba perderse el progreso parcial.

La interfaz StreamingStep

Un paso de transmisión implementa 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étodo	Descripción
`stream( $state, $existing_chunks )`	Generador que produce fragmentos de texto. Cada generación provoca una escritura inmediata en la base de datos.
`on_complete( $chunks, $state )`	Se llama cuando termina la transmisión. Recibe todos los fragmentos (anteriores y nuevos) y devuelve datos que se combinarán con el estado del flujo de trabajo.
`config()`	Configuración opcional. Admite las claves `max_attempts` y `backoff`; `backoff` acepta un nombre de estrategia (`exponential`, `linear`, `fixed`), ya que actualmente se descarta una lista de demoras (#16).

Cómo se persisten los fragmentos

Cada vez que el generador produce un valor, Queuety llama a ChunkStore::append_chunk() para escribirlo en la tabla queuety_chunks. No se agrupan ni almacenan en búfer. Cada fragmento generado equivale a una operación INSERT.

La fila del fragmento almacena job_id, workflow_id, step_index, chunk_index (desde cero) y la cadena content.

Reanudación

Cuando se reintenta un paso de transmisión tras un fallo, el worker carga todos los fragmentos persistidos y los pasa como parámetro $existing_chunks. Tu implementación determina cómo reanudar:

Omitir los elementos ya procesados comprobando el número de fragmentos existentes
Avanzar en un API paginado usando el número de fragmentos como desplazamiento de página
Empezar desde cero ignorando por completo $existing_chunks (los fragmentos del intento anterior siguen en la base de datos y se añadirán al principio)

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

Señales de actividad durante la transmisión

Las transmisiones largas deben enviar señales de actividad para evitar que el detector de trabajos obsoletos recupere el trabajo. Llama a Heartbeat::beat() dentro del bucle del generador:

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 Señales de actividad para obtener más información sobre la API de actividad.

Limpieza de fragmentos

Cuando la transmisión termina correctamente y on_complete() devuelve un resultado, el worker llama a ChunkStore::clear_chunks() para eliminar todos los fragmentos del trabajo. El resultado combinado de on_complete() es el registro duradero. Los fragmentos son temporales por diseño.

Si el paso falla y se reintenta, los fragmentos no se eliminan. Se vuelven a pasar mediante $existing_chunks en el siguiente intento.

Ejemplo completo: transmisión de 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 ];
    }
}

Úsalo en un flujo de trabajo como cualquier otro paso:

PHP

use Queuety\Queuety;

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

Ejemplo completo: procesamiento de archivos

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

Clase ChunkStore API

La clase ChunkStore gestiona la persistencia de fragmentos. Accede a ella mediante Queuety::chunk_store().

PHP

$store = Queuety::chunk_store();

Método	Devuelve	Descripción
`get_chunks( int $job_id )`	`string[]`	Obtiene todos los fragmentos de un trabajo, ordenados por índice de fragmento
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Añade un único fragmento
`clear_chunks( int $job_id )`	`void`	Elimina todos los fragmentos de un trabajo
`chunk_count( int $job_id )`	`int`	Cuenta los fragmentos de un trabajo
`get_accumulated( int $job_id )`	`string`	Concatena todos los fragmentos en una sola cadena

Normalmente no necesitas llamar directamente a ChunkStore. El worker gestiona automáticamente la persistencia y la limpieza al procesar un StreamingStep. API está disponible para casos avanzados, como inspeccionar fragmentos en curso o realizar una limpieza manual.