# Streaming-Schritte

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](https://github.com/inline0/queuety/issues/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](https://queuety.inline0.dev/de/docs/features/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.