Etapas de streaming

Ver como Markdown Editar no GitHub Documentação em 2cc7b2b

As etapas de streaming permitem que um fluxo de trabalho produza saída incrementalmente. Cada bloco gerado é gravado imediatamente no banco de dados. Assim, se a etapa falhar durante o streaming, os blocos já persistidos são preservados e ficam disponíveis na nova tentativa.

Isso é útil para respostas em streaming de LLMs, processamento de arquivos grandes e qualquer situação em que o progresso parcial não deva ser perdido.

A interface StreamingStep

Uma etapa de streaming 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	Descrição
`stream( $state, $existing_chunks )`	Gerador que produz blocos de texto. Cada bloco gerado aciona uma gravação imediata no banco de dados.
`on_complete( $chunks, $state )`	Chamado após o fim do streaming. Recebe todos os blocos (anteriores e novos) e retorna dados para mesclar ao estado do fluxo de trabalho.
`config()`	Configuração opcional. Aceita as chaves `max_attempts` e `backoff`; `backoff` recebe o nome de uma estratégia (`exponential`, `linear`, `fixed`), pois uma lista de atrasos é descartada atualmente (#16).

Como os blocos são persistidos

Sempre que o gerador produz um valor, Queuety chama ChunkStore::append_chunk() para gravá-lo na tabela queuety_chunks. Não há agrupamento nem armazenamento em buffer. Cada bloco gerado corresponde a um INSERT.

A linha do bloco armazena job_id, workflow_id, step_index, chunk_index (com índice iniciado em zero) e a string content.

Capacidade de retomada

Quando uma etapa de streaming é repetida após uma falha, o worker carrega todos os blocos persistidos anteriormente e os passa como parâmetro $existing_chunks. Sua implementação decide como retomar:

Ignorar itens já processados verificando a quantidade de blocos existentes
Avançar em um API paginado usando a quantidade de blocos como deslocamento da página
Recomeçar do zero ignorando completamente $existing_chunks (os blocos da tentativa anterior ainda estão no banco de dados e serão antepostos)

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

Heartbeats durante o streaming

Streams de longa duração devem enviar heartbeats para impedir que o detector de tarefas obsoletas recupere a tarefa. Chame Heartbeat::beat() dentro do loop do gerador:

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

Consulte Heartbeats para saber mais sobre a API de heartbeat.

Limpeza de blocos

Quando o streaming termina com sucesso e on_complete() retorna, o worker chama ChunkStore::clear_chunks() para excluir todos os blocos da tarefa. O resultado mesclado de on_complete() é o registro durável. Os blocos são temporários por definição.

Se a etapa falhar e for repetida, os blocos não serão excluídos. Eles serão repassados por $existing_chunks na próxima tentativa.

Exemplo completo: streaming 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 ];
    }
}

Use-o em um fluxo de trabalho como qualquer outra etapa:

PHP

use Queuety\Queuety;

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

Exemplo completo: processamento de arquivos

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

Classe API ChunkStore

A classe ChunkStore gerencia a persistência de blocos. Acesse-a por Queuety::chunk_store().

PHP

$store = Queuety::chunk_store();

Método	Retorna	Descrição
`get_chunks( int $job_id )`	`string[]`	Busca todos os blocos de uma tarefa, ordenados pelo índice do bloco
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Acrescenta um único bloco
`clear_chunks( int $job_id )`	`void`	Exclui todos os blocos de uma tarefa
`chunk_count( int $job_id )`	`int`	Conta a quantidade de blocos de uma tarefa
`get_accumulated( int $job_id )`	`string`	Concatena todos os blocos em uma string

Normalmente, não é preciso chamar ChunkStore diretamente. O worker gerencia a persistência e a limpeza automaticamente ao processar um StreamingStep. API está disponível para casos de uso avançados, como inspecionar blocos em andamento ou fazer a limpeza manual.