Proteções de fluxos de trabalho

Fluxos de trabalho no estilo de agentes são flexíveis, mas também facilitam a criação de execuções duplicadas ou permitem que um planejador produza mais trabalho do que o previsto. Queuety oferece proteções leves no nível do fluxo de trabalho para que você mantenha essa flexibilidade sem perder o controle.

Para conhecer o modelo mais amplo de recursos para admissão de trabalhadores e metadados de custo por trabalho, consulte Gerenciamento de recursos.

Versionar uma definição de fluxo de trabalho

Use version() para anexar um rótulo de versão no nível da aplicação a uma definição de fluxo de trabalho:

Queuety::workflow( 'research_run' )
    ->version( 'research.v2' )
    ->then( PlanResearchStep::class )
    ->then( DraftResearchSummaryStep::class )
    ->dispatch();

A versão é armazenada junto ao estado do fluxo de trabalho, disponibilizada por workflow_status(), exibida por wp queuety workflow status e incluída nas exportações do fluxo de trabalho.

Queuety também armazena um definition_hash determinístico para cada execução do fluxo de trabalho. O hash é derivado da definição serializada do fluxo de trabalho e é útil quando você precisa identificar exatamente qual estrutura de fluxo de trabalho uma execução em andamento está usando após uma implantação.

Tornar o despacho idempotente

Use idempotency_key() quando os chamadores puderem repetir a mesma solicitação:

Queuety::workflow( 'research_run' )
    ->idempotency_key( 'brief:42:research' )
    ->then( PlanResearchStep::class )
    ->dispatch( [ 'brief_id' => 42 ] );

Se um fluxo de trabalho for despachado novamente com uma chave já usada, Queuety retorna o ID original do fluxo de trabalho em vez de criar uma execução duplicada. As chaves são globais entre os nomes dos fluxos de trabalho; portanto, inclua o nome do fluxo de trabalho na chave se fluxos diferentes puderem compartilhar uma.

Isso é especialmente útil para:

• endpoints API que podem ser repetidos por clientes ou gateways
• consumidores de webhook que podem receber entregas duplicadas
• etapas planejadoras que devem criar apenas uma execução de orquestração por entidade externa

As chaves de idempotência são armazenadas na tabela queuety_workflow_dispatch_keys.

Definir orçamentos de fluxo de trabalho

Use orçamentos de fluxo de trabalho para interromper execuções que excedam os limites planejados:

Queuety::workflow( 'research_run' )
    ->max_transitions( 20 )
    ->max_for_each_items( 12 )
    ->max_state_bytes( 32768 )
    ->max_cost_units( 40 )
    ->max_started_workflows( 8 )
    ->then( PlanResearchStep::class )
    ->for_each( 'tasks', ExecuteResearchTask::class, 'results' )
    ->dispatch();

max_transitions()

Limita quantas etapas do fluxo de trabalho podem ser concluídas antes que a execução falhe. Isso é útil para proteger fluxos de trabalho com ramificações dinâmicas ou que podem revisitar a mesma área por meio de _next_step.

max_for_each_items()

Limita quantos itens descobertos em tempo de execução uma única etapa for_each() pode expandir. Isso é útil para sistemas de planejamento e execução, nos quais um agente pode descobrir muito mais trabalho de ramificação do que o previsto.

max_state_bytes()

Limita o tamanho codificado do estado público do fluxo de trabalho. Isso evita que fluxos de trabalho de longa duração cresçam até se tornarem grandes blocos, lentos para inspecionar, exportar ou repetir.

max_cost_units()

Limita as unidades de custo totais consumidas por etapas concluídas. As unidades de custo vêm dos metadados da etapa ou do trabalho que Queuety resolve ao enfileirar trabalho. Trabalhos regulares do usuário usam 1 por padrão; espaços reservados de orquestração, como atrasos e esperas, usam 0; e os manipuladores podem definir um valor positivo maior por meio de config() ou das propriedades do trabalho. Apenas cost_units de uma definição de etapa pode definir um valor de 0.

Isso é útil quando um fluxo de trabalho pode revisitar a mesma área ou expandir itens em trabalho cujo custo não é capturado apenas pela contagem de etapas. Um planejador que continua executando uma chamada cara a um modelo pode ficar dentro de max_transitions() e ainda assim exceder o limite de custo planejado.

max_started_workflows()

Limita quantos fluxos de trabalho de nível superior uma execução pode iniciar por meio de start_workflows() ou start_agents(). O limite é verificado antes do despacho dos fluxos de trabalho filhos, de modo que o fluxo de trabalho pai falha sem criar parcialmente um grupo que já exceda o orçamento.

Quando um limite é excedido, Queuety interrompe o fluxo de trabalho imediatamente, em vez de tentar executar a etapa novamente. Uma exceção: quando Queuety::signal() conclui uma espera e a transição resultante excede um limite, a exceção é lançada para quem chamou signal(), a inserção do sinal é revertida e o fluxo de trabalho continua aguardando.

Monitoramento das proteções

O status do fluxo de trabalho expõe:

• definition_version
• definition_hash
• idempotency_key
• budget

O conteúdo de budget inclui os limites configurados e o uso atual, inclusive o tamanho atual do estado público em bytes.

Exemplo:

$state = Queuety::workflow_status( $workflow_id );

$state->definition_version; // 'research.v2'
$state->definition_hash;    // '4c7a...'
$state->idempotency_key;    // 'brief:42:research'
$state->budget;             // [ 'max_transitions' => 20, ... ]