Aggiorna i checkpoint del flusso di lavoro Python alla versione 1.13.0

Agent Framework 1.13.0 contiene lievi modifiche incompatibili all'esecuzione dei flussi di lavoro Python. La maggior parte delle applicazioni non richiede modifiche. Le modifiche influiscono sulle applicazioni che dipendono dal numero esatto di passaggi sovrapposti o dai numeri di iterazione, impostati max_iterations al limite di convergenza, controllano l'ID origine del messaggio iniziale o fanno ipotesi sul posizionamento e l'ordinamento dei checkpoint.

Background

Prima della versione 1.13.0, il checkpoint non soddisfava completamente la promessa di acquisire lo stato del flusso di lavoro necessario per riprendere l'esecuzione da qualsiasi limite registrato. L'esecutore di avvio è stato eseguito prima del ciclo di superstep e checkpoint, quindi il primo checkpoint conteneva l'output dell'esecutore di avvio e lo stato aggiornato, ma non l'input originale del flusso di lavoro. Analogamente, le risposte agli eventi di richiesta sono state recapitate ed elaborate senza prima essere registrate in un checkpoint. Di conseguenza, nessun checkpoint avrebbe potuto rieseguire l'esecutore iniziale a partire dall'input originale né riprodurre una continuazione con intervento umano dalla risposta restituita.

Modifiche del comportamento

La versione 1.13.0 chiude queste lacune. L’executor di avvio ora viene eseguito nel primo superstep, un checkpoint di ingresso registra l’input iniziale prima di tale superstep e un checkpoint di ingresso delle risposte registra le risposte consegnate prima che vengano elaborate. Nel complesso, queste modifiche rendono un flusso di lavoro con checkpoint interamente rieseguibile a partire dal suo input, comprese le continuazioni con intervento umano.

Modifiche che potrebbero richiedere un'azione

Miglioramenti della riproducibilità

Aggiornare la gestione degli eventi superstep

Una nuova esecuzione del flusso di lavoro produce ora un'ulteriore coppia di eventi di superstep perché l'esecutore iniziale viene eseguito nel superstep 1:

• superstep_started con iteration == 1
• superstep_completed con iteration == 1

I successivi turni di lavoro dell'executor si spostano di un passaggio superiore. Aggiornate i test, la telemetria, gli indicatori di avanzamento o altro codice che presupponga un numero esatto di eventi o associ un determinato executor a un'iterazione fissa.

Il codice che risponde ai tipi di evento senza basarsi sul conteggio o sull'iterazione non deve cambiare.

Esaminare il limite massimo di iterazione

Il limite max_iterations ora comprende anche il superstep che esegue l'esecutore di avvio. Se in precedenza un flusso di lavoro usava il limite completo, aumentare il valore configurato di uno:

from agent_framework import WorkflowBuilder

workflow = WorkflowBuilder(
    start_executor=start_executor,
    max_iterations=previous_max_iterations + 1,
).build()

Non è necessaria alcuna modifica se il flusso di lavoro converge già prima di raggiungere il limite configurato.

Aggiornare i controlli iniziali dell'origine dei messaggi

Se un esecutore di avvio consuma l'ID di origine del messaggio iniziale, sostituisci il valore "Workflow" codificato in modo statico con l'ID di origine dell'arco interno dell'esecutore di avvio.

Prima della versione 1.13.0:

is_workflow_input = ctx.source_executor_ids != ["Workflow"]

Nella versione 1.13.0 e successive:

from agent_framework import INTERNAL_SOURCE_ID

is_workflow_input = ctx.source_executor_ids != [INTERNAL_SOURCE_ID(self.id)]

INTERNAL_SOURCE_ID(executor_id) attualmente restituisce "internal:<executor_id>". Usare l'helper invece di costruire questa stringa in modo che il codice segua il formato ID di origine del framework.

Gestione dei checkpoint di aggiornamento

Punti di controllo iniziali dell'immissione

Quando il checkpointing è abilitato, a ogni nuova esecuzione viene ora creato un checkpoint iniziale in iteration_count == 0. Questo checkpoint contiene l'input originale come messaggio in transito indirizzato all'esecutore di avvio. Ripristinandolo, viene rieseguito l'esecutore di avvio e viene riprodotta l'intera esecuzione del flusso di lavoro.

Dopo ogni superstep completato, il framework continua a generare un checkpoint. Per un'esecuzione con $N$ superstep, prevedete $N + 1$ checkpoint: il checkpoint iniziale seguito da un checkpoint per ogni superstep completato.

Esaminare il codice che presuppone che il checkpoint dell’iterazione 0 contenga lo stato generato dallo start executor. Tale stato viene ora visualizzato nel checkpoint creato dopo il passaggio superiore 1.

Punti di controllo delle richieste e risposte

Quando si continua un flusso di lavoro con workflow.run(responses=...), il framework crea ora un checkpoint di inserimento delle risposte dopo aver accodato le risposte e prima di eseguire il superstep che le consuma. Il ripristino di questo checkpoint recapita nuovamente le risposte registrate e riproduce il resto del flusso di lavoro.

Il checkpoint di response-entry presenta lo stesso iteration_count del checkpoint precedente che contiene la richiesta in sospeso. Si tratta di un checkpoint separato il cui previous_checkpoint_id punta a quel checkpoint della richiesta in sospeso.

Elenco di controllo per la migrazione

• Aggiorna le asserzioni e i consumer di eventi che dipendono dal numero esatto di superstep o dai numeri delle iterazioni.
• Aumentare max_iterations di uno solo per i flussi di lavoro che hanno raggiunto il limite precedente.
• Sostituire i controlli iniziali dell'ID sorgente per "Workflow" con INTERNAL_SOURCE_ID(start_executor.id).
• Considera il checkpoint dell'iterazione 0 come checkpoint di input pre-esecuzione.
• Ordina i checkpoint con intervento umano in base alla discendenza anziché presumere che iteration_count sia univoco.
• Verificare che la riesecuzione di un checkpoint di ingresso e di un checkpoint di ingresso della risposta produca l'output e gli effetti collaterali previsti.

Per i dettagli di implementazione, vedi Consentire la ripetibilità completa del checkpoint del flusso di lavoro.