Поточно предаване на стъпки

Преглед като Markdown Редактиране в GitHub Документация за 2cc7b2b

Стъпките с поточно предаване позволяват на работния процес да генерира резултат постепенно. Всяка генерирана част се записва незабавно в базата данни, така че ако стъпката се провали по средата на потока, вече запазените части се запазват и са налични при повторен опит.

Това е полезно за поточно предаване на отговори от LLM, обработка на големи файлове и всеки сценарий, при който частичният напредък не бива да се губи.

Интерфейсът StreamingStep

Стъпка с поточно предаване реализира 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;
}

Метод	Описание
`stream( $state, $existing_chunks )`	Генератор, който извежда низове на части. Всяко извеждане задейства незабавен запис в DB.
`on_complete( $chunks, $state )`	Извиква се след приключване на потока. Получава всички части (предишните и новите) и връща данни за обединяване със състоянието на работния процес.
`config()`	Незадължителна конфигурация. Поддържа ключовете `max_attempts` и `backoff`; `backoff` приема име на стратегия (`exponential`, `linear`, `fixed`), тъй като списък със закъснения в момента се отхвърля (#16).

Как се запазват частите

Всеки път, когато генераторът ви изведе стойност, Queuety извиква ChunkStore::append_chunk(), за да я запише в таблицата queuety_chunks. Няма групиране или буфериране. Едно извеждане е равно на един INSERT.

Редът на частта съхранява job_id, workflow_id, step_index, chunk_index (започващ от нула) и низа content.

Възможност за подновяване

Когато стъпка с поточно предаване бъде изпълнена повторно след неуспех, работникът зарежда всички преди това запазени части и ги подава като параметъра $existing_chunks. Вашата реализация определя как да продължи:

Пропуснете вече обработените елементи чрез проверка на броя на съществуващите части
Преминете до нужната позиция в странициран API, като използвате броя на частите като отместване на страницата
Започнете отначало, като напълно игнорирате $existing_chunks (частите от предишния опит все още са в DB и ще бъдат добавени отпред)

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

Сигнали за активност по време на поточно предаване

Дългите потоци трябва да изпращат сигнали за активност, за да не може детекторът за остарели задачи да прибере задачата обратно. Извикайте Heartbeat::beat() вътре в цикъла на генератора:

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

Вижте Сигнали за активност за подробности относно API за сигнали за активност.

Почистване на частите

Когато потокът приключи успешно и on_complete() върне резултат, работникът извиква ChunkStore::clear_chunks(), за да изтрие всички части за задачата. Обединеният резултат от on_complete() е трайно запазеният запис. Частите са временни по замисъл.

Ако стъпката се провали и бъде изпълнена повторно, частите не се изтриват. При следващия опит те се подават обратно чрез $existing_chunks.

Пълен пример: поточно предаване от 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 ];
    }
}

Използвайте го в работен процес като всяка друга стъпка:

PHP

use Queuety\Queuety;

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

Пълен пример: обработка на файл

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

Класът ChunkStore управлява запазването на частите. Достъпете го чрез Queuety::chunk_store().

PHP

$store = Queuety::chunk_store();

Метод	Връща	Описание
`get_chunks( int $job_id )`	`string[]`	Извлича всички части за задача, подредени по индекс на частта
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Добавя една част
`clear_chunks( int $job_id )`	`void`	Изтрива всички части за задача
`chunk_count( int $job_id )`	`int`	Брои частите за задача
`get_accumulated( int $job_id )`	`string`	Обединява всички части в един низ

Обикновено не е необходимо да извиквате ChunkStore директно. Работникът автоматично се грижи за запазването и почистването при обработка на StreamingStep. API е наличен за по-специални случаи, като например проверка на части в процес на обработка или ръчно почистване.