스트리밍 단계

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, 0부터 시작하는 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 클래스는 청크 저장을 관리합니다. 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는 진행 중인 청크 검사나 수동 정리 같은 고급 사용 사례에 사용할 수 있습니다.