Modalità di esecuzione del flusso di lavoro

Quando si esegue un flusso di lavoro in .NET, la modalità di esecuzione controlla come vengono elaborati i superstep e il modo in cui gli eventi vengono recapitati al consumer. La InProcessExecution classe espone due modalità di esecuzione: OffThread e Lockstep.

Informazioni generali

OffThread

OffThread è la modalità di esecuzione predefinita . I superstep vengono eseguiti su un thread in secondo piano e gli eventi vengono trasmessi immediatamente quando vengono generati tramite un'implementazione basata su canali.

// OffThread is the default — these are equivalent:
await using StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, input);
await using StreamingRun run = await InProcessExecution.OffThread.RunStreamingAsync(workflow, input);

Come funziona

1. Un'attività in background esegue continuamente i supersteps mentre i messaggi sono in sospeso.
2. Poiché gli executor producono output o eventi, gli oggetti risultanti WorkflowEvent vengono scritti in un Channel<WorkflowEvent> illimitato.
3. Il consumatore legge gli eventi dal canale tramite WatchStreamAsync, ricevendoli in tempo reale mentre vengono prodotti.
4. Quando tutti i superpassi sono completi e non rimangono messaggi, l'esecuzione si interrompe con uno stato Idle o PendingRequests.

Poiché il ciclo superstep e il consumer vengono eseguiti simultaneamente, gli eventi vengono visualizzati non appena vengono generati, senza alcun ritardo di buffering. Questo rende OffThread ideale per scenari di streaming in cui il recapito di eventi a bassa latenza è importante, ad esempio la visualizzazione degli aggiornamenti token per token in un'interfaccia utente.

Esecuzioni simultanee

OffThread supporta anche una variante simultanea che consente a più esecuzioni di condividere contemporaneamente la stessa istanza del flusso di lavoro:

await using StreamingRun run = await InProcessExecution.Concurrent.RunStreamingAsync(workflow, input);

Lockstep

In modalità Lockstep i superstep vengono eseguiti nel thread del consumer anziché su un'attività in background. Gli eventi vengono accumulati durante ogni superpasso e rilasciati come un lotto al termine del superpasso.

await using StreamingRun run = await InProcessExecution.Lockstep.RunStreamingAsync(workflow, input);

Come funziona

1. Il consumer chiama WatchStreamAsync, che avvia il ciclo di esecuzione.
2. Un passaggio superiore viene eseguito fino al completamento e gli eventi vengono accumulati in una coda.
3. Al termine del superstep, tutti gli eventi in coda vengono restituiti al consumatore.
4. Il prossimo superstep inizia solo dopo che il consumatore ha ricevuto tutti gli eventi da quello precedente.

Lo schema alternato implica che il consumer e il motore del flusso di lavoro non siano mai attivi contemporaneamente. Il recapito degli eventi è deterministico: tutti gli eventi da un superpasso arriveranno sicuramente prima di qualsiasi evento dal superpasso successivo.

Quando usare Lockstep

Lockstep è utile quando:

• Test : l'ordinamento deterministico degli eventi rende le asserzioni semplici.
• Debug : il debug dettagliato è più semplice quando l'esecuzione rimane nel thread del consumer.
• Elaborazione ordinata : scenari in cui è necessario elaborare completamente gli eventi di un superstep prima dell'inizio del passaggio successivo.

Scelta di una modalità di esecuzione

Per la maggior parte degli scenari di produzione, è consigliabile usare la modalità OffThread predefinita. Offre la massima velocità di risposta e consente al flusso di lavoro di continuare l'elaborazione mentre il consumer gestisce gli eventi.

Usare Lockstep quando il comportamento deterministico è più importante delle prestazioni, ad esempio negli unit test o nelle sessioni di debug.

// Production: OffThread (default)
await using StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, input);

// Testing: Lockstep for deterministic behavior
await using StreamingRun run = await InProcessExecution.Lockstep.RunStreamingAsync(workflow, input);

Esecuzione non in streaming

Entrambe le modalità di esecuzione supportano l'esecuzione non in streaming tramite RunAsync. In modalità non di streaming, il flusso di lavoro viene eseguito fino al completamento e raccoglie tutti gli eventi in un Run oggetto anziché trasmetterli in modo incrementale:

Run run = await InProcessExecution.RunAsync(workflow, input);

// Access all emitted events
foreach (WorkflowEvent evt in run.OutgoingEvents)
{
    // Process events
}

Poiché l'esecuzione non in streaming raccoglie tutti gli eventi dopo il completamento, il vantaggio di recapito degli eventi in tempo reale di OffThread non si applica. La differenza principale tra le modalità negli scenari non di streaming è il threading: OffThread esegue i superstep in un thread in background, liberando il thread chiamante mentre attende il completamento, mentre Lockstep esegue i superstep sul thread del chiamante, bloccandolo fino al termine del flusso di lavoro.

L'esecuzione non in streaming usa la modalità OffThread predefinita. Per usare Lockstep con esecuzione non di streaming:

Run run = await InProcessExecution.Lockstep.RunAsync(workflow, input);

Passaggi successivi

Le modalità di esecuzione non sono applicabili ai flussi di lavoro Python. I flussi di lavoro Python usano un singolo modello di esecuzione che gestisce l'elaborazione dei superstep e la consegna degli eventi tramite un generatore asincrono. Questo modello è simile alla modalità Lockstep .NET. I passaggi non avanzano a meno che il consumer non esegua attivamente il pull degli eventi dal generatore.

Per informazioni sull'esecuzione di flussi di lavoro Python, vedere Generatore flussi di lavoro ed esecuzione.

Quando si esegue un flusso di lavoro in Go, l'ambiente di esecuzione controlla la modalità di elaborazione dei superstep e il modo in cui gli eventi vengono recapitati al consumer. Il workflow/inproc pacchetto espone tre ambienti: Default/OffThread, Lockstepe .Concurrent

Informazioni generali

OffThread

OffThread è la modalità di esecuzione predefinita. Questi sono equivalenti:

stream, err := inproc.Default.RunStreaming(ctx, wf, input)
stream, err := inproc.OffThread.RunStreaming(ctx, wf, input)

Come funziona

1. Una goroutine in background esegue i passaggi avanzati mentre i messaggi sono in sospeso.
2. Man mano che gli esecutori generano output o eventi, gli eventi del flusso di lavoro vengono scritti nello stream.
3. Il consumer legge gli eventi tramite WatchStream, ricevendoli man mano che vengono prodotti.
4. Quando tutti i passaggi sovrapposti sono completi e non rimangono messaggi, l'esecuzione si interrompe con uno stato di inattività o di richiesta in sospeso.

Esecuzioni simultanee

Usare inproc.Concurrent quando tutte le associazioni dell'executor nel flusso di lavoro supportano l'esecuzione condivisa simultanea:

stream, err := inproc.Concurrent.RunStreaming(ctx, wf, input)
if err != nil {
    return err
}
defer stream.Close(ctx)

Lockstep

In modalità Lockstep, l'esecuzione del flusso di lavoro avanza man mano che il consumer legge dallo stream. Ciò rende deterministico l'ordinamento degli eventi per i test e il debug.

stream, err := inproc.Lockstep.RunStreaming(ctx, wf, input)
if err != nil {
    return err
}
defer stream.Close(ctx)

for evt, err := range stream.WatchStream(ctx) {
    if err != nil {
        return err
    }
    // inspect event
}

Come funziona

1. Il consumer chiama WatchStream, che avvia il ciclo di esecuzione.
2. Un superstep viene eseguito fino al completamento e gli eventi vengono accumulati.
3. Gli eventi accumulati vengono passati al consumatore.
4. Il superstep successivo inizia solo dopo che il consumer riceve gli eventi del superstep precedente.

Quando usare Lockstep

Usare Lockstep quando il comportamento deterministico conta più del flusso a bassa latenza, ad esempio unit test, debug o scenari in cui si vogliono elaborare completamente gli eventi di un superstep prima dell'inizio del passaggio successivo.

Scelta di una modalità di esecuzione

Per la maggior parte degli scenari di produzione, usare inproc.Default o inproc.OffThread. Usare inproc.Lockstep quando l'ordinamento di eventi deterministici è più importante della latenza di streaming, ad esempio nei test. Usare inproc.Concurrent solo quando ogni associazione nel flusso di lavoro supporta l'esecuzione condivisa simultanea.

// Production: OffThread (default)
stream, err := inproc.Default.RunStreaming(ctx, wf, input)
if err != nil {
    return err
}
defer stream.Close(ctx)

// Testing: Lockstep for deterministic behavior
testStream, err := inproc.Lockstep.RunStreaming(ctx, wf, input)
if err != nil {
    return err
}
defer testStream.Close(ctx)

Esecuzione non in streaming

Tutti gli ambienti di esecuzione supportano anche la modalità non in streaming Run, che viene eseguita fino al successivo arresto e memorizza gli eventi emessi nell'esecuzione restituita.

run, err := inproc.Default.Run(ctx, wf, input)
if err != nil {
    return err
}

for evt := range run.NewEvents() {
    if output, ok := evt.(workflow.OutputEvent); ok {
        fmt.Printf("Final result: %v\n", output.Output)
    }
}

Passaggi successivi