워크플로우

Markdown으로 보기 GitHub에서 수정 문서 위치 2cc7b2b

워크플로우는 장기 실행 작업을 개별 단계로 나눕니다. 각 단계는 출력을 데이터베이스의 공유 상태 저장소에 저장합니다. PHP가 단계 도중 종료되면 작업자가 이전 상태를 모두 보존한 채 해당 단계를 재시도합니다.

워크플로우란?

워크플로우는 순서가 정해진 단계의 연속입니다. 각 단계는 PHP 클래스로, 이전 모든 단계에서 누적된 상태를 받아 새 데이터를 반환해 상태에 병합합니다. 워크플로우는 한 번에 한 단계씩 진행되며, 각 전이는 원자적인 MySQL 트랜잭션으로 처리됩니다.

PHP

use Queuety\Queuety;

$workflow_id = Queuety::workflow( 'generate_report' )
    ->then( FetchDataHandler::class )
    ->then( CallLLMHandler::class )
    ->then( FormatOutputHandler::class )
    ->dispatch( [ 'user_id' => 42 ] );

상태 누적

dispatch()에 전달하는 초기 페이로드가 시작 상태가 됩니다. 각 단계에서 반환한 키는 상태에 병합됩니다. _로 시작하는 키는 _next_step 같은 제어 값으로 예약되며 공개 상태에 병합되지 않습니다:

코드

dispatch(['user_id' => 42])

Step 0: FetchDataHandler   -> returns ['user_name' => 'Alice']
  state: {user_id: 42, user_name: 'Alice'}

Step 1: CallLLMHandler     -> returns ['llm_response' => '...']
  state: {user_id: 42, user_name: 'Alice', llm_response: '...'}

Step 2: FormatOutputHandler -> returns ['report_url' => '/reports/42.pdf']
  state: {user_id: 42, user_name: 'Alice', llm_response: '...', report_url: '/reports/42.pdf'}

모든 단계는 누적된 전체 상태를 확인할 수 있습니다. 외부 저장소 없이 단계 간에 데이터를 전달하기 쉽습니다.

재개 가능성

각 단계가 끝날 때마다 상태가 데이터베이스에 저장되므로 워크플로우는 내구성이 있습니다. 작업자가 1단계 도중 중단되면 0단계의 출력이 상태에 보존된 채 워크플로우가 1단계부터 다시 시작합니다.

단계 경계는 다음 작업을 원자적으로 처리하는 단일 MySQL 트랜잭션입니다:

단계의 출력으로 워크플로우 상태를 업데이트합니다.
현재 단계의 작업을 완료로 표시합니다.
다음 단계의 작업을 대기열에 추가합니다.

이렇게 하면 충돌, 시간 초과 또는 PHP 치명적 오류가 발생해도 상태가 손실되지 않습니다.

워크플로우 생명 주기

코드

dispatch() -> running -> step 0 -> step 1 -> ... -> step N -> completed
                  |                   |
                  |                   +-> failed -> retry -> running (from failed step)
                  |
                  +-> paused -> resumed -> running
                  |
                  +-> waiting_for_signal -> signal received -> running
                  |
                  +-> waiting_for_workflows -> dependency settles -> running
                  |
                  +-> cancelled (via cancel_workflow)

상태	설명
`running`	단계를 활발히 처리 중
`completed`	모든 단계를 성공적으로 완료함
`failed`	재시도를 모두 소진한 뒤 단계 실패
`paused`	수동으로 일시 중지됨. 현재 진행 중인 작업은 완료될 수 있지만, 다시 시작할 때까지 후속 단계는 대기열에 추가되지 않음
`waiting_for_signal`	`Queuety::signal()`를 통해 외부 신호를 기다리는 중
`waiting_for_workflows`	`wait_for_workflow()` 또는 `wait_for_workflows()`을 통해 하나 이상의 다른 워크플로우가 완료되기를 기다리는 중
`cancelled`	`Queuety::cancel_workflow()`을 통해 명시적으로 취소됨. 대기 중인 작업은 묻히고 선택적 정리 핸들러가 실행됨.

워크플로우 관리하기

상태 확인:

PHP

$state = Queuety::workflow_status( $workflow_id );
echo $state->status->value;   // 'running'
echo $state->current_step;    // 1
echo $state->total_steps;     // 3
print_r( $state->waiting_for ); // ['42'] while waiting on another workflow
print_r( $state->wait_details ); // matched/remaining blockers, result key, correlation info

일시 중지 및 다시 시작:

PHP

Queuety::pause_workflow( $workflow_id );
Queuety::resume_workflow( $workflow_id );

실패한 단계부터 실패한 워크플로우 재시도:

PHP

Queuety::retry_workflow( $workflow_id );

워크플로우 취소 및 정리 핸들러 실행:

PHP

Queuety::cancel_workflow( $workflow_id );

정리 핸들러 및 워크플로우 취소 시 동작에 대한 자세한 내용은 취소를 참조하세요.

워크플로우 옵션

워크플로우의 모든 단계에 적용할 대기열, 우선순위 및 재시도 동작을 설정하세요:

PHP

use Queuety\Enums\Priority;

Queuety::workflow( 'important_pipeline' )
    ->on_queue( 'pipelines' )
    ->with_priority( Priority::High )
    ->max_attempts( 5 )
    ->on_cancel( PipelineCleanupHandler::class )
    ->prune_state_after( 3 )
    ->then( StepA::class )
    ->then( StepB::class )
    ->dispatch( $payload );

on_cancel() 메서드는 워크플로우가 취소될 때 실행되는 정리 핸들러를 등록합니다. 취소를 참조하세요.

prune_state_after() 메서드는 워크플로우 상태를 일정하게 유지하기 위해 오래된 단계 출력을 자동으로 제거합니다. 상태 정리를 참조하세요.

compensate_with()를 사용해 단계별 보상 핸들러를 지정하세요. 명시적인 취소뿐만 아니라 워크플로우 실패 후 완료된 단계의 보상을 자동으로 실행하려면 compensate_on_failure()를 사용하세요. 자세한 내용은 보상을 참조하세요.

장기 실행 오케스트레이션의 안전성을 높이려면 워크플로우 안전 장치를 사용하세요. version()는 정의에 애플리케이션 수준 버전 태그를 지정하고, Queuety는 각 실행의 결정론적인 정의 해시도 저장하며, idempotency_key()는 재시도나 중복 요청에도 디스패치를 안정적으로 처리합니다. max_transitions(), max_for_each_items() 및 max_state_bytes()는 폭주하기 전에 실행을 실패 처리합니다. 자세한 내용은 안전 장치를 참조하세요.

단계 유형

Queuety는 워크플로우 내에서 다음 11가지 유형의 단계를 지원합니다:

순차 체인: 한 번에 하나씩 실행하도록 ->then() 사용
병렬 그룹: 동시에 실행하도록 ->parallel() 사용
동적 For-Each: 런타임에 발견한 분기 작업과 명시적인 완료 조건에 ->for_each() 사용
비동기 핸드오프: 런타임에 발견한 항목에서 독립적인 하위 워크플로우를 디스패치하도록 ->start_workflows() 또는 ->start_agents() 사용
조건부 분기: 이름이 지정된 단계 및 _next_step 사용
반복: 앞서 이름이 지정된 단계로 내구성 있는 역방향 분기를 만들도록 ->repeat_until() 및 ->repeat_while() 사용
워크플로우 실행: 중첩 구성에 ->run_workflow() 사용
내구성 있는 지연: 시간 기반 지연에 ->delay() 사용
신호: 외부 이벤트 트리거에 ->wait_for_signal(), ->wait_for_signals(), ->wait_for_approval(), ->wait_for_input() 및 ->wait_for_decision() 사용
워크플로우 종속성: 워크플로우 간 조정에 ->wait_for_workflow(), ->wait_for_workflows() 및 ->wait_for_agents() 사용
스트리밍 단계: 청크 기반 출력을 내구성 있게 처리하도록 StreamingStep 인터페이스 사용

일반적인 패턴

자율 또는 반자율 시스템을 구축한다면 먼저 에이전트 오케스트레이션을 확인하세요. 실제 플래너/실행기 및 사람 검토 흐름에서 for_each(), start_agents(), wait_for_agents(), wait_for_workflow() 및 wait_for_decision()가 어떻게 함께 작동하는지 보여줍니다.