単純なシーケンシャルワークフローを作成する

このチュートリアルでは、Agent Framework ワークフローを使用して単純なシーケンシャルワークフローを作成する方法について説明します。

シーケンシャルワークフローは、複雑な AI エージェントシステムを構築する基盤です。このチュートリアルでは、各ステップでデータを処理し、次の手順に渡す単純な 2 段階認証ワークフローを作成する方法について説明します。

概要

このチュートリアルでは、2 つの Executor を含むワークフローを作成します。

大文字変換エグゼキューター - 入力テキストを大文字に変換します
テキストの逆引き実行プログラム - テキストを反転し、最終的な結果を出力します

このワークフローは、次のような主要な概念を示しています。

1 つのハンドラーでカスタム Executor を作成する
関数からカスタム Executor を作成する
WorkflowBuilderを使用して Executor とエッジを接続する
順次ステップによるデータの処理
イベントを使用したワークフローの実行の監視

対象となる概念

[前提条件]

.NET 8.0 SDK 以降
この基本的な例では、外部 AI サービスは必要ありません
新しいコンソールアプリケーション

ステップバイステップの実装

以降のセクションでは、順次ワークフローを段階的に構築する方法について説明します。

手順 1: NuGet パッケージをインストールする

まず、.NET プロジェクトに必要なパッケージをインストールします。

dotnet add package Microsoft.Agents.AI.Workflows --prerelease

手順 2: アッパーケース・エグゼキュータを定義する

テキストを大文字に変換する Executor を定義します。

using System;
using System.Linq;
using System.Threading.Tasks;
using Microsoft.Agents.AI.Workflows;

/// <summary>
/// First executor: converts input text to uppercase.
/// </summary>
Func<string, string> uppercaseFunc = s => s.ToUpperInvariant();
var uppercase = uppercaseFunc.BindExecutor("UppercaseExecutor");

キーポイント:

文字列を受け取り、大文字のバージョンを返す関数を作成する
BindExecutor()を使用して関数から Executor を作成する

手順 3: リバーステキスト Executor を定義する

テキストを反転させる Executor を定義します。

/// <summary>
/// Second executor: reverses the input text and completes the workflow.
/// </summary>
internal sealed class ReverseTextExecutor() : Executor<string, string>("ReverseTextExecutor")
{
    public override ValueTask<string> HandleAsync(string input, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        // Reverse the input text
        return ValueTask.FromResult(new string(input.Reverse().ToArray()));
    }
}

ReverseTextExecutor reverse = new();

キーポイント:

から継承するクラスを作成する Executor<TInput, TOutput>
入力を処理して出力を返す HandleAsync() を実装する

手順 4: ワークフローを構築して接続する

WorkflowBuilderを使用して Executor を接続します。

// Build the workflow by connecting executors sequentially
WorkflowBuilder builder = new(uppercase);
builder.AddEdge(uppercase, reverse).WithOutputFrom(reverse);
var workflow = builder.Build();

キーポイント:

WorkflowBuilder コンストラクターは開始用エグゼキュータを受け取ります
AddEdge() は、大文字からリバースへの直接接続を作成します。
WithOutputFrom() ワークフロー出力を生成する Executor を指定します
Build() は、変更できないワークフローを作成します

手順 5: ワークフローを実行する

ワークフローを実行し、結果を確認します。

// Execute the workflow with input data
await using Run run = await InProcessExecution.RunAsync(workflow, "Hello, World!");
foreach (WorkflowEvent evt in run.NewEvents)
{
    switch (evt)
    {
        case ExecutorCompletedEvent executorComplete:
            Console.WriteLine($"{executorComplete.ExecutorId}: {executorComplete.Data}");
            break;
    }
}

手順 6: ワークフロー出力について

ワークフローを実行すると、次のような出力が表示されます。

UppercaseExecutor: HELLO, WORLD!
ReverseTextExecutor: !DLROW ,OLLEH

入力 "Hello, World!" は、最初に大文字 ("HELLO, WORLD!") に変換され、次に反転 ("!DLROW,OLLEH")。

主な概念の説明

Executor インターフェイス

関数からのエグゼキュータ

BindExecutor()を使用して関数から Executor を作成する

Executor は、 Executor<TInput, TOutput>を実装します。

TInput: この Executor が受け入れるデータの型
TOutput: この Executor によって生成されるデータの型
HandleAsync: 入力を処理し、出力を返すメソッド

.NET ワークフロービルダーパターン

WorkflowBuilderは、ワークフローを構築するための fluent API を提供します。

コンストラクター: 初期実行者を受け取ります
AddEdge(): Executor 間にダイレクト接続を作成します
WithOutputFrom(): ワークフロー出力を生成する Executor を指定します
Build(): 最終的な変更できないワークフローを作成します

.NET イベントの種類

実行中に、次のイベントの種類を確認できます。

ExecutorCompletedEvent - Executor の処理が完了したとき

完全な .NET の例

完全ですぐに実行できる実装については、Agent Framework リポジトリの 01_ExecutorsAndEdgesサンプルを参照してください。

このサンプルには次のものが含まれます。

すべての using ステートメントとクラス構造を使用した完全な実装
ワークフローの概念を説明する追加のコメント
プロジェクトのセットアップと構成を完了する

概要

このチュートリアルでは、2 つの Executor を含むワークフローを作成します。

大文字変換実行機 - 入力テキストを大文字に変換します
テキストの逆引き実行プログラム - テキストを反転し、最終的な結果を出力します

このワークフローは、次のような主要な概念を示しています。

@executor デコレーターを使用してワークフローノードを作成する
Executor を WorkflowBuilder に接続する
ステップ間でのデータの受け渡し ctx.send_message()
ctx.yield_output() を使用して最終的な出力を生成する
リアルタイムの可観測性のためのストリーミングイベント

対象となる概念

[前提条件]

Python 3.10 以降
インストールされている Agent Framework Core Python パッケージ: pip install agent-framework-core --pre
この基本的な例では、外部 AI サービスは必要ありません

ステップバイステップの実装

以降のセクションでは、順次ワークフローを段階的に構築する方法について説明します。

手順 1: 必要なモジュールをインポートする

まず、Agent Framework から必要なモジュールをインポートします。

import asyncio
from typing_extensions import Never
from agent_framework import WorkflowBuilder, WorkflowContext, WorkflowOutputEvent, executor

手順 2: 最初の Executor を作成する

ハンドラーメソッドを用いてテキストを大文字に変換するエグゼキュータを作成し、実装します。

class UpperCase(Executor):
    def __init__(self, id: str):
        super().__init__(id=id)

    @handler
    async def to_upper_case(self, text: str, ctx: WorkflowContext[str]) -> None:
        """Convert the input to uppercase and forward it to the next node.

        Note: The WorkflowContext is parameterized with the type this handler will
        emit. Here WorkflowContext[str] means downstream nodes should expect str.
        """
        result = text.upper()

        # Send the result to the next executor in the workflow.
        await ctx.send_message(result)

キーポイント:

@executorデコレーターは、この関数をワークフローノードとして登録します
WorkflowContext[str] は、この Executor が最初の型パラメーターを指定してダウンストリームに文字列を送信することを示します
ctx.send_message() データを次の手順に渡します

手順 3: 2 つ目の Executor を作成する

テキストを反転させ、 @executorで修飾されたメソッドから最終的な出力を生成する Executor を作成します。

@executor(id="reverse_text_executor")
async def reverse_text(text: str, ctx: WorkflowContext[Never, str]) -> None:
    """Reverse the input and yield the workflow output."""
    result = text[::-1]

    # Yield the final output for this workflow run
    await ctx.yield_output(result)

キーポイント:

WorkflowContext[Never, str] これは、最初の型パラメーターとして Never を指定してメッセージを送信せず、2 番目のパラメーターとして str を指定してワークフロー出力を生成するターミナル実行プログラムであることを示します
ctx.yield_output() は、最終的なワークフロー結果を提供します
停止状態になるとワークフローが完了する

手順 4: ワークフローを構築する

WorkflowBuilderを使用して Executor を接続します。

upper_case = UpperCase(id="upper_case_executor")

workflow = (
    WorkflowBuilder()
    .add_edge(upper_case, reverse_text)
    .set_start_executor(upper_case)
    .build()
)

キーポイント:

add_edge() は、Executor 間にダイレクト接続を作成します
set_start_executor() エントリポイントを定義します。
build() ワークフローを最終処理します

手順 5: ストリーミングを使用してワークフローを実行する

ワークフローを実行し、リアルタイムでイベントを観察します。

async def main():
    # Run the workflow and stream events
    async for event in workflow.run_stream("hello world"):
        print(f"Event: {event}")
        if isinstance(event, WorkflowOutputEvent):
            print(f"Workflow completed with result: {event.data}")

if __name__ == "__main__":
    asyncio.run(main())

手順 6: 出力について

ワークフローを実行すると、次のようなイベントが表示されます。

Event: ExecutorInvokedEvent(executor_id=upper_case_executor)
Event: ExecutorCompletedEvent(executor_id=upper_case_executor)
Event: ExecutorInvokedEvent(executor_id=reverse_text_executor)
Event: ExecutorCompletedEvent(executor_id=reverse_text_executor)
Event: WorkflowOutputEvent(data='DLROW OLLEH', source_executor_id=reverse_text_executor)
Workflow completed with result: DLROW OLLEH

主な概念の説明

ワークフローコンテキストの種類

WorkflowContextジェネリック型は、Executor 間のデータフローを定義します。

WorkflowContext[str] - 次の Executor に文字列を送信します
WorkflowContext[Never, str] - 文字列型のワークフロー出力を生成するターミナル実行プログラム

イベントの種類

ストリーミングの実行中に、次のイベントの種類を確認します。

ExecutorInvokedEvent - Executor が処理を開始したとき
ExecutorCompletedEvent - Executor の処理が完了したとき
WorkflowOutputEvent - 最終的なワークフロー結果が含まれます

Python ワークフロービルダーパターン

WorkflowBuilderは、ワークフローを構築するための fluent API を提供します。

add_edge(): Executor 間のダイレクト接続を作成します
set_start_executor(): ワークフローエントリポイントを定義します
build(): 変更できないワークフローオブジェクトを最終処理して返します

コード例全体

完全ですぐに実行できる実装については、Agent Framework リポジトリのサンプルを参照してください。

このサンプルには次のものが含まれます。

すべてのインポートとドキュメントを使用した完全な実装
ワークフローの概念を説明する追加のコメント
予想される結果を示すサンプル出力

次のステップ

単純な同時実行ワークフローの作成について説明します

フィードバック

このページはお役に立ちましたか?

Last updated on 2025-11-21

次の方法で共有

単純なシーケンシャル ワークフローを作成する

概要

対象となる概念

[前提条件]

ステップ バイ ステップの実装

手順 1: NuGet パッケージをインストールする

手順 2: アッパーケース・エグゼキュータを定義する

手順 3: リバース テキスト Executor を定義する

手順 4: ワークフローを構築して接続する

手順 5: ワークフローを実行する

手順 6: ワークフロー出力について

主な概念の説明

Executor インターフェイス

.NET ワークフロー ビルダー パターン

.NET イベントの種類

完全な .NET の例

概要

対象となる概念

[前提条件]

ステップ バイ ステップの実装

手順 1: 必要なモジュールをインポートする

手順 2: 最初の Executor を作成する

手順 3: 2 つ目の Executor を作成する

手順 4: ワークフローを構築する

手順 5: ストリーミングを使用してワークフローを実行する

手順 6: 出力について

主な概念の説明

ワークフロー コンテキストの種類

イベントの種類

Python ワークフロー ビルダー パターン

コード例全体

次のステップ

フィードバック

その他のリソース

単純なシーケンシャルワークフローを作成する

ステップバイステップの実装

手順 3: リバーステキスト Executor を定義する

.NET ワークフロービルダーパターン

ステップバイステップの実装

ワークフローコンテキストの種類

Python ワークフロービルダーパターン