Folyamatos adatfolyam-lépések

Megtekintés Markdownként Szerkesztés itt: GitHub Dokumentáció itt: 2cc7b2b

Az adatfolyam-lépések lehetővé teszik, hogy a munkafolyamat fokozatosan állítson elő kimenetet. Minden kibocsátott adatdarab azonnal bekerül az adatbázisba, így ha a lépés adatfolyam közben hibázik, a már mentett adatdarabok megmaradnak, és újrapróbálkozáskor rendelkezésre állnak.

Ez hasznos az LLM-adatfolyam-válaszokhoz, nagy fájlok feldolgozásához és minden olyan helyzetben, amikor a részleges előrehaladás nem veszhet el.

A StreamingStep interfész

Az adatfolyam-lépés megvalósítja a Queuety\Contracts\StreamingStep interfészt:

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

Metódus	Leírás
`stream( $state, $existing_chunks )`	Karakterlánc-adatdarabokat kibocsátó generátor. Minden kibocsátás azonnali adatbázis-írást indít.
`on_complete( $chunks, $state )`	Az adatfolyam befejeződése után hívódik meg. Megkapja az összes (korábbi és új) adatdarabot, és visszaadja a munkafolyamat állapotába egyesítendő adatokat.
`config()`	Opcionális konfiguráció. A `max_attempts` és `backoff` kulcsokat támogatja; a `backoff` stratégianevet vár (`exponential`, `linear`, `fixed`), mivel a késleltetések listáját jelenleg eldobja (#16).

Az adatdarabok mentése

Minden alkalommal, amikor a generátor értéket bocsát ki, a Queuety meghívja a ChunkStore::append_chunk() metódust, hogy az értéket a queuety_chunks táblába írja. Nincs kötegelt írás vagy pufferelés. Egy kibocsátás egy INSERT műveletet jelent.

Az adatdarab sora a következőket tárolja: job_id, workflow_id, step_index, chunk_index (nullától kezdve), valamint a content karakterláncot.

Folytathatóság

Ha egy adatfolyam-lépés hiba után újra megpróbálkozik, a munkafolyamat-kezelő betölti az összes korábban mentett adatdarabot, és átadja őket a $existing_chunks paraméterként. A megvalósítás dönti el, hogyan folytatja a feldolgozást:

A már feldolgozott elemek kihagyása a meglévő adatdarabok számának ellenőrzésével
Lapozott API eltolása az adatdarabszám használatával oldalkezdő pozícióként
Újrakezdés az elejéről a $existing_chunks teljes figyelmen kívül hagyásával (az előző próbálkozás adatdarabjai továbbra is az adatbázisban vannak, és eléjük kerülnek)

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

Szívverések adatfolyam közben

A hosszan futó adatfolyamoknak szívveréseket kell küldeniük, hogy az elavult feladatokat figyelő érzékelő ne vegye át a feladatot. Hívja meg a Heartbeat::beat() metódust a generátor ciklusán belül:

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

Részletekért lásd a Szívverések szakaszt, amely a Heartbeat API-ről szól.

Adatdarabok törlése

Ha az adatfolyam sikeresen befejeződik, és a on_complete() visszatér, a munkafolyamat-kezelő meghívja a ChunkStore::clear_chunks() metódust a feladat összes adatdarabjának törléséhez. A on_complete() egyesített eredménye a tartós rekord. Az adatdarabok kialakításuk szerint ideiglenesek.

Ha a lépés hibázik, és újrapróbálkozik, az adatdarabokat nem törli. A következő próbálkozáskor a $existing_chunks segítségével adja vissza őket.

Teljes példa: LLM-adatfolyam

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

Használja egy munkafolyamatban, mint bármely más lépést:

PHP

use Queuety\Queuety;

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

Teljes példa: fájlfeldolgozás

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

ChunkStore API

A ChunkStore osztály kezeli az adatdarabok mentését. A Queuety::chunk_store() segítségével érhető el.

PHP

$store = Queuety::chunk_store();

Metódus	Visszatérési érték	Leírás
`get_chunks( int $job_id )`	`string[]`	Lekéri egy feladat összes adatdarabját az adatdarab-index szerinti sorrendben
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Hozzáfűz egyetlen adatdarabot
`clear_chunks( int $job_id )`	`void`	Törli egy feladat összes adatdarabját
`chunk_count( int $job_id )`	`int`	Megszámolja egy feladat adatdarabjait
`get_accumulated( int $job_id )`	`string`	Egyetlen karakterlánccá fűzi össze az összes adatdarabot

Általában nincs szükség arra, hogy közvetlenül meghívja a ChunkStore metódust. A munkafolyamat-kezelő automatikusan gondoskodik a mentésről és a törlésről egy StreamingStep feldolgozásakor. A API speciális felhasználási esetekben, például folyamatban lévő adatdarabok vizsgálatához vagy kézi törléshez áll rendelkezésre.