ワークフロー実行モード

.NET でワークフローを実行する場合、 実行モード はスーパーステップの処理方法と、コンシューマーにイベントを配信する方法を制御します。 InProcessExecution クラスは、OffThread と Lockstep の 2 つの実行モードを公開します。

概要

OffThread

OffThread は既定の実行モード です 。 スーパーステップはバックグラウンド スレッドで実行され、イベントはチャネル ベースの実装を介して発生するとすぐにストリーム アウトされます。

// 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);

どのように機能するのか

1. バックグラウンド タスクは、メッセージが保留中の間、スーパーステップを継続的に実行します。
2. Executor が出力またはイベントを生成すると、結果の WorkflowEvent オブジェクトは無制限の Channel<WorkflowEvent>に書き込まれます。
3. コンシューマーは、 WatchStreamAsyncを介してチャネルからイベントを読み取り、生成されたイベントをリアルタイムで受信します。
4. すべてのスーパーステップが完了し、メッセージが残っていない場合、実行は Idle 状態または PendingRequests 状態で停止します。

スーパーステップ ループとコンシューマーは同時に実行されるため、イベントは発生するとすぐに表示されます。バッファリングの遅延はありません。 これにより、OFFThread は、UI でのトークンごとの更新の表示など、待機時間の短いイベント配信が重要なストリーミング シナリオに最適です。

同時実行

OffThread では、複数の実行で同じワークフロー インスタンスを同時に共有できる 同時実行 バリアントもサポートされています。

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

Lockstep

ロックステップ モードでは、スーパーステップはバックグラウンド タスクではなく コンシューマーのスレッド で実行されます。 イベントは各スーパーステップの間に蓄積され、スーパーステップの完了後にバッチとして出力されます。

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

どのように機能するのか

1. コンシューマーは WatchStreamAsyncを呼び出し、実行ループを駆動します。
2. スーパーステップは完了まで実行され、イベントはキューに蓄積されます。
3. スーパーステップが完了すると、キューに登録されたすべてのイベントがコンシューマーに渡されます。
4. 次のスーパーステップは、コンシューマーが前のスーパーステップからすべてのイベントを受信した後にのみ開始されます。

この交互のパターンは、コンシューマーとワークフロー エンジンが同時に実行されることはないことを意味します。 イベント配信は決定論的です。スーパーステップからのすべてのイベントは、次のスーパーステップからのイベントの前に到着することが保証されます。

Lockstep を使用するタイミング

Lockstep は、次の場合に便利です。

• テスト — 決定論的なイベントの順序付けにより、アサーションが簡単になります。
• デバッグ — コンシューマーのスレッドで実行を維持すると、ステップスルー デバッグが簡単になります。
• 順序付けされた処理 — 次のスーパーステップが開始される前に、1 つのスーパーステップのイベントを完全に処理する必要があるシナリオ。

実行モードの選択

ほとんどの運用シナリオでは、既定の OffThread モードをお勧めします。 最適な応答性を提供し、コンシューマーがイベントを処理している間もワークフローの処理を続行できます。

単体テストやデバッグ セッションなど、パフォーマンスよりも決定的な動作が重要な場合は、 Lockstep を使用します。

// 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);

非ストリーミング実行

どちらの実行モードでも、 RunAsyncを介した非ストリーミング実行がサポートされます。 非ストリーミング モードでは、ワークフローは完了まで実行され、すべてのイベントを段階的にストリーミングするのではなく、 Run オブジェクトに収集します。

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

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

ストリーミング以外の実行では完了後にすべてのイベントが収集されるため、OffThread のリアルタイム イベント配信の利点は適用されません。 ストリーミング以外のシナリオでのモードの主な違いは スレッドです。OffThread はバックグラウンド スレッドでスーパーステップを実行し、完了を待っている間に呼び出し元のスレッドを解放します。一方、Lockstep は呼び出し元のスレッドでスーパーステップを実行し、ワークフローが完了するまでブロックします。

非ストリーミング実行では、既定の OffThread モードが使用されます。 ストリーミング以外の実行で Lockstep を使用するには:

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

次のステップ

実行モードは、Python ワークフローには適用されません。 Python ワークフローでは、非同期ジェネレーターを介したスーパーステップ処理とイベント配信を処理する単一の実行モデルが使用されます。 このモデルは .NET ロックステップ モードに似ています。コンシューマーがジェネレーターからアクティブにイベントをプルしない限り、手順は進まない。

Python ワークフローの実行については、「 ワークフロー ビルダーと実行」を参照してください。

Go でワークフローを実行する場合、実行環境はスーパーステップの処理方法と、コンシューマーにイベントを配信する方法を制御します。 workflow/inproc パッケージは、Default/OffThread、Lockstep、Concurrentの 3 つの環境を公開します。

概要

OffThread

OffThread は既定の実行モードです。 これらは同等です。

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

どのように機能するのか

1. メッセージが保留中の間、バックグラウンド の goroutine はスーパーステップを実行します。
2. Executor が出力またはイベントを生成すると、ワークフロー イベントがストリームに書き込まれます。
3. コンシューマーは、 WatchStreamを使用してイベントを読み取り、生成されたイベントを受信します。
4. すべてのスーパーステップが完了し、メッセージが残っていない場合、実行はアイドル状態または保留中の要求状態で停止します。

同時実行

ワークフロー内のすべての Executor バインドで同時共有実行がサポートされている場合は、 inproc.Concurrent を使用します。

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

Lockstep

ロックステップ モードでは、コンシューマーがストリームから読み取るにつれてワークフローの実行が進みます。 これにより、テストとデバッグのイベント順序が決定論的になります。

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
}

どのように機能するのか

1. コンシューマーは WatchStreamを呼び出し、実行ループを駆動します。
2. スーパーステップの実行が完了し、イベントが累積されます。
3. 蓄積されたイベントはコンシューマーに渡されます。
4. 次のスーパーステップは、コンシューマーが前のスーパーステップのイベントを受信した後にのみ開始されます。

Lockstep を使用するタイミング

確定的な動作が、単体テスト、デバッグ、次のスーパーステップの開始前に 1 つのスーパーステップのイベントを完全に処理するシナリオなど、待機時間の短いストリーミングよりも重要な場合は、Lockstep を使用します。

実行モードの選択

ほとんどの運用シナリオでは、 inproc.Default または inproc.OffThreadを使用します。 テストなど、ストリーミング待機時間よりも決定的なイベント順序が重要な場合は、 inproc.Lockstep を使用します。 inproc.Concurrentは、ワークフロー内のすべてのバインドで同時共有実行がサポートされている場合にのみ使用します。

// 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)

非ストリーミング実行

また、すべての実行環境では、ストリーミング以外の Runもサポートされています。これは次の停止まで実行され、返された実行で生成されたイベントを格納します。

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)
    }
}

次のステップ