Streaming-Schritte

Als Markdown anzeigen Auf GitHub bearbeiten Dokumentation zu 2cc7b2b

Mit Streaming-Schritten kann ein Workflow Ausgaben schrittweise erzeugen. Jeder ausgegebene Chunk wird sofort in die Datenbank geschrieben. Schlägt der Schritt mitten im Stream fehl, bleiben die bereits gespeicherten Chunks erhalten und stehen beim erneuten Versuch zur Verfügung.

Das ist nützlich für Streaming-Antworten von LLMs, die Verarbeitung großer Dateien und alle Szenarien, in denen Teilergebnisse nicht verloren gehen sollen.

Die StreamingStep-Schnittstelle

Ein Streaming-Schritt implementiert 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;
}

Methode	Beschreibung
`stream( $state, $existing_chunks )`	Generator, der String-Chunks ausgibt. Jeder Aufruf von yield löst sofort einen DB-Schreibvorgang aus.
`on_complete( $chunks, $state )`	Wird nach Abschluss des Streams aufgerufen. Erhält alle Chunks (vorherige und neue) und gibt Daten zurück, die mit dem Workflow-Zustand zusammengeführt werden.
`config()`	Optionale Konfiguration. Unterstützt die Schlüssel `max_attempts` und `backoff`. `backoff` erwartet einen Strategienamen (`exponential`, `linear`, `fixed`), da eine Liste von Verzögerungen derzeit verworfen wird (#16).

So werden Chunks gespeichert

Jedes Mal, wenn dein Generator einen Wert ausgibt, ruft Queuety ChunkStore::append_chunk() auf, um ihn in die Tabelle queuety_chunks zu schreiben. Es gibt keine Stapelverarbeitung und keine Pufferung. Ein Aufruf von yield entspricht einem INSERT.

Der Chunk-Datensatz speichert job_id, workflow_id, step_index, chunk_index (nullbasiert) und den String content.

Fortsetzen

Wird ein Streaming-Schritt nach einem Fehler erneut versucht, lädt der Worker alle zuvor gespeicherten Chunks und übergibt sie als Parameter $existing_chunks. Deine Implementierung entscheidet, wie sie fortgesetzt wird:

Bereits verarbeitete Elemente überspringen, indem du die Anzahl vorhandener Chunks prüfst
Zu einer paginierten API springen, wobei die Chunk-Anzahl als Seitenoffset dient
Von vorn beginnen, indem du $existing_chunks vollständig ignorierst (Chunks des vorherigen Versuchs sind weiterhin in der DB und werden vorangestellt)

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 während des Streamings

Lange Streams sollten Heartbeats senden, damit der Erkennung veralteter Jobs den Job nicht zurückfordert. Rufe Heartbeat::beat() innerhalb deiner Generatorschleife auf:

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

Weitere Informationen zur Heartbeat-API findest du unter Heartbeats.

Chunks bereinigen

Wird der Stream erfolgreich abgeschlossen und gibt on_complete() zurück, ruft der Worker ChunkStore::clear_chunks() auf, um alle Chunks des Jobs zu löschen. Das zusammengeführte Ergebnis von on_complete() ist der dauerhaft gespeicherte Datensatz. Chunks sind absichtlich nur vorübergehend vorhanden.

Schlägt der Schritt fehl und wird erneut versucht, werden die Chunks nicht gelöscht. Beim nächsten Versuch werden sie über $existing_chunks zurückgegeben.

Vollständiges Beispiel: LLM-Streaming

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

Verwende es wie jeden anderen Schritt in einem Workflow:

PHP

use Queuety\Queuety;

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

Vollständiges Beispiel: Dateiverarbeitung

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

Die Klasse ChunkStore verwaltet die Speicherung von Chunks. Du kannst über Queuety::chunk_store() darauf zugreifen.

PHP

$store = Queuety::chunk_store();

Methode	Rückgabewert	Beschreibung
`get_chunks( int $job_id )`	`string[]`	Ruft alle Chunks eines Jobs ab, sortiert nach Chunk-Index
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Fügt einen einzelnen Chunk hinzu
`clear_chunks( int $job_id )`	`void`	Löscht alle Chunks eines Jobs
`chunk_count( int $job_id )`	`int`	Zählt die Chunks eines Jobs
`get_accumulated( int $job_id )`	`string`	Verbindet alle Chunks zu einem String

Normalerweise musst du ChunkStore nicht direkt aufrufen. Der Worker übernimmt die Speicherung und Bereinigung automatisch, wenn er ein StreamingStep verarbeitet. Für fortgeschrittene Anwendungsfälle wie das Prüfen von Chunks während des Streamings oder eine manuelle Bereinigung steht API zur Verfügung.