ワークフローは 、エグゼキューター と エッジ を有向グラフに結び付け、実行を管理します。 Executor の呼び出し、メッセージ ルーティング、イベント ストリーミングを調整します。
ワークフローの構築
ワークフローは、ワークフロー構造を定義するための fluent API を提供する WorkflowBuilder クラスを使用して構築されます。
using Microsoft.Agents.AI.Workflows;
var processor = new DataProcessor();
var validator = new Validator();
var formatter = new Formatter();
// Build workflow
WorkflowBuilder builder = new(processor); // Set starting executor
builder.AddEdge(processor, validator);
builder.AddEdge(validator, formatter);
var workflow = builder.Build();
ワークフローは、 WorkflowBuilder クラスを使用して構築されます。
from agent_framework import WorkflowBuilder
processor = DataProcessor()
validator = Validator()
formatter = Formatter()
# Build workflow
builder = WorkflowBuilder(start_executor=processor)
builder.add_edge(processor, validator)
builder.add_edge(validator, formatter)
workflow = builder.build()
ビルド時の検証時に、ハンドラーは、 list、 dict、 tuple、 Sequenceなどのベア コンテナー注釈を使用できます。
list[str]など、対応するパラメーター化されたコンテナーを使用する出力を受け取ることができます。
workflow パッケージは、エグゼキューターがエッジによって接続されるグラフベースの実行モデルを提供します。
- Executor - 入力を受信して出力を生成する処理装置
- Edge - 1 つの Executor の出力を別の Executor の入力に接続します
- ビルダー - Executor とエッジを定義してワークフローを構築する
- 実行 - 指定された入力を使用してワークフローを実行します
この関数は、モデルや資格情報を必要とせずにテキスト処理ワークフローを構築します。
import (
"slices"
"strings"
"github.com/microsoft/agent-framework-go/workflow"
)
func buildWorkflow() (*workflow.Workflow, error) {
uppercase := workflow.NewExecutor(
"UppercaseExecutor", strings.ToUpper,
).Bind()
reverse := workflow.NewExecutor(
"ReverseExecutor", func(input string) string {
runes := []rune(input)
slices.Reverse(runes)
return string(runes)
},
).Bind()
return workflow.NewBuilder(uppercase).
AddEdge(uppercase, reverse).
WithOutputFrom(reverse).
Build()
}
WithOutputFrom(reverse) は、リバース Executor の結果をワークフロー出力として公開します。 エッジは、出力の指定に関係なく、大文字の結果をリバース Executor に引き続き配信します。
buildWorkflowによって返されたワークフローを、以下のいずれかの実行関数に渡します。 入力 "Hello, World!" の場合、どちらも Workflow output: !DLROW ,OLLEH を出力します。
ワークフローの実行
ワークフローでは、ストリーミング実行モードと非ストリーミング実行モードの両方がサポートされています。
using Microsoft.Agents.AI.Workflows;
// Streaming execution — get events as they happen
StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, inputMessage);
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
if (evt is ExecutorCompletedEvent executorComplete)
{
Console.WriteLine($"{executorComplete.ExecutorId}: {executorComplete.Data}");
}
if (evt is WorkflowOutputEvent outputEvt)
{
Console.WriteLine($"Workflow completed: {outputEvt.Data}");
}
}
// Non-streaming execution — wait for completion
Run result = await InProcessExecution.RunAsync(workflow, inputMessage);
foreach (WorkflowEvent evt in result.NewEvents)
{
if (evt is WorkflowOutputEvent outputEvt)
{
Console.WriteLine($"Final result: {outputEvt.Data}");
}
}
# Streaming execution — get events as they happen
async for event in workflow.run(input_message, stream=True):
if event.type == "output":
print(f"Workflow completed: {event.data}")
# Non-streaming execution — wait for completion
events = await workflow.run(input_message)
print(f"Final result: {events.get_outputs()}")
イベントを発生時に受け取る場合は、RunStreaming を使用します。 呼び出し元のコンテキストを渡して、取り消しがワークフローに到達するようにし、イテレーションの終了後にエラーを確認し、完了したら実行を閉じます。
import (
"context"
"errors"
"fmt"
"github.com/microsoft/agent-framework-go/workflow"
"github.com/microsoft/agent-framework-go/workflow/inproc"
)
func runStreaming(
ctx context.Context, wf *workflow.Workflow,
) (err error) {
stream, err := inproc.Default.RunStreaming(ctx, wf, "Hello, World!")
if err != nil {
return err
}
defer func() {
err = errors.Join(err, stream.Close(context.WithoutCancel(ctx)))
}()
for evt, streamErr := range stream.WatchStream(ctx) {
if streamErr != nil {
return streamErr
}
switch event := evt.(type) {
case workflow.OutputEvent:
fmt.Printf("Workflow output: %v\n", event.Output)
case workflow.ErrorEvent:
return event.Error
case workflow.ExecutorFailedEvent:
return fmt.Errorf("executor %q failed: %w",
event.ExecutorID, event.Error)
}
}
return ctx.Err()
}
WatchStream は、外部要求が保留中の間待機します。 人間による入力が必要なワークフローの場合は、要求イベントを処理するか、WatchUntilHalt を使用して一時停止時に戻ります。 「ワークフロー イベント」を参照してください。
Runを使用して次の停止に進み、収集されたイベントを検査します。 実行が完了したときだけでなく、外部入力の待機中にも実行が停止する可能性があります。
import (
"context"
"errors"
"fmt"
"github.com/microsoft/agent-framework-go/workflow"
"github.com/microsoft/agent-framework-go/workflow/inproc"
)
func runToHalt(
ctx context.Context, wf *workflow.Workflow,
) (err error) {
run, err := inproc.Default.Run(ctx, wf, "Hello, World!")
if err != nil {
return err
}
defer func() {
err = errors.Join(err, run.Close(context.WithoutCancel(ctx)))
}()
for evt := range run.NewEvents() {
switch event := evt.(type) {
case workflow.OutputEvent:
fmt.Printf("Workflow output: %v\n", event.Output)
case workflow.ErrorEvent:
return event.Error
case workflow.ExecutorFailedEvent:
return fmt.Errorf("executor %q failed: %w",
event.ExecutorID, event.Error)
}
}
for evt := range run.OutgoingEvents() {
if evt, ok := evt.(workflow.ExecutorCompletedEvent); ok {
fmt.Printf("%s: %v\n", evt.ExecutorID, evt.Result)
}
}
return ctx.Err()
}
NewEvents() は、呼び出すたびにブックマークを進めます。 2 つ目のループでは、OutgoingEvents() を使用して、NewEvents() で再度消費しようとするのではなく、累積された Executor イベントを再生します。
これらの関数は、結果の確認後に実行を終了します。 アプリケーションでさらにメッセージを送信したり、一時停止後に再開したりする必要がある場合は、代わりに run を開いたままにします。
コンシューマーの詳細、出力の分類、一時停止のセマンティクスについては、「ワークフロー イベント」を参照してください。
Tip
実行可能な完全な サンプルについては、ワークフローの例 を参照してください。
ワークフローの検証
ワークフロー ビルダーは、実行前にグラフを検証します。 正確なチェックは、SDK と使用可能な型メタデータによって異なります。
- 型の互換性: メタデータが使用可能な接続された Executor 間で宣言されたメッセージ プロトコルを確認します。
- グラフ接続: 登録された Executor が開始 Executor から到達可能であることを確認します。
- Executor バインド: 必要な Executor 登録が存在することを確認します。
- エッジ検証: ビルダーのルールに従って、無効な接続と重複するエッジをチェックします。
Go Build() は、未解決のバインド、到達できない登録された Executor、重複する無条件エッジ、および互換性のない宣言されたメッセージの種類を拒否します。 条件付きルートと動的プロトコルには、引き続きランタイム テストが必要です。
型検証では、一時的な Executor インスタンスを作成して、それらのプロトコルを検査できます。 ファクトリまたはプロトコルを検査できない場合、その互換性チェックはスキップされます。 ビルドが成功しても、実行時に Executor の作成やすべてのメッセージ配信が成功することは保証されません。 実行の作成、イベントの反復、ワークフローの失敗イベントからのエラーを処理します。
BuildWithOptions(workflow.BuildOptions{DisableOrphanValidation: true}) は orphan-executor チェックのみを無効にします。 その他の検証は引き続き実行されます。
実行モデル: スーパーステップ
このフレームワークでは、変更された Pregel 実行モデル (スーパーステップ ベースの処理を含む一括同期並列 (BSP) アプローチ) を使用します。
スーパーステップのしくみ
ワークフローの実行は、個別のスーパーステップに編成されます。 各スーパーステップ:
- 前のスーパーステップから保留中のすべてのメッセージを収集します
- エッジ定義に基づいてターゲット Executor にメッセージをルーティングする
- スーパーステップ内ですべてのターゲット Executor を同時に実行します。
- すべての Executor が完了するまで待機してから進めます (同期バリア)
- Executor によって出力された新しいメッセージをキューに入れ、次のスーパーステップに進みます。
Superstep N:
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Collect All │───▶│ Route Messages │───▶│ Execute All │
│ Pending │ │ Based on Type │ │ Target │
│ Messages │ │ & Conditions │ │ Executors │
└─────────────────┘ └─────────────────┘ └─────────────────┘
│
│ (barrier: wait for all)
┌─────────────────┐ ┌─────────────────┐ │
│ Start Next │◀───│ Emit Events & │◀────────────┘
│ Superstep │ │ New Messages │
└─────────────────┘ └─────────────────┘
同期バリア
最も重要な特性は、スーパーステップ間の同期バリアです。 トリガーされた Executor はスーパーステップ内で同時に実行できますが、それらが完了するまで、ワークフローは次のスーパーステップに進むことはありません。
これはファンアウト パターンに影響します。複数のパス (1 つは実行時間の長い Executor のチェーンを持ち、もう 1 つは実行時間の長い Executor を持つパス) にファンアウトする場合、チェーンパスは実行時間の長い Executor が完了するまで進めません。
スーパーステップの理由
BSP モデルは、次の重要な保証を提供します。
- 定義されたステップ境界: スーパーステップ中に出力されたメッセージは、後で処理するためにキューに入れられます。
- チェックポイント境界: チェックポイントが有効なワークフローは、スーパーステップ境界で実行状態を保存します。
- バッファー状態の更新: キューに置かれたワークフロー状態の変更は、スーパーステップ間でコミットされます。
これらの保証によって、1 つのスーパーステップ内で同時に実行される Executor 間の実行やイベントの順序が固定されるわけではありません。 スーパーステップ バリアがすべてのレースコンディションを防ぐと想定するのではなく、アプリケーション所有の共有状態へのアクセスを同期します。
スーパーステップ モデルの操作
相互にブロックしない完全に独立した並列パスが必要な場合は、シーケンシャル ステップを 1 つの Executor に統合します。
step1 → step2 → step3をチェーンするのではなく、そのロジックを 1 つの Executor に結合します。 その後、両方の並列パスが 1 つのスーパーステップ内で実行されます。
次のステップ
関連トピック: