サブワークフロー

Sub-Workflow の作成

C# では、次の 2 つの方法でサブワークフローを作成します。

• 直接バインド — BindAsExecutor() を使用して、ワークフローを親ワークフローの Executor として直接埋め込みます。 これにより、サブワークフローのネイティブ入力/出力型が保持されます。
• エージェント ラッピング — AsAIAgent() を使用してワークフローをエージェントに変換し、その後、そのエージェントを親ワークフローに追加します。 これは、親ワークフローでエージェント ベースの Executor を使用する場合に便利です。

BindAsExecutor での直接バインディング

BindAsExecutor()拡張メソッドは、ワークフローを親ワークフローに直接追加できるExecutorBindingに変換します。

using Microsoft.Agents.AI.Workflows;

// Create executors for the inner workflow
UppercaseExecutor uppercase = new();
ReverseExecutor reverse = new();
AppendSuffixExecutor append = new(" [PROCESSED]");

// Build the inner workflow
var innerWorkflow = new WorkflowBuilder(uppercase)
    .AddEdge(uppercase, reverse)
    .AddEdge(reverse, append)
    .WithOutputFrom(append)
    .Build();

// Bind the inner workflow as an executor
ExecutorBinding subWorkflowExecutor = innerWorkflow.BindAsExecutor("TextProcessingSubWorkflow");

// Build the parent workflow using the sub-workflow executor
PrefixExecutor prefix = new("INPUT: ");
PostProcessExecutor postProcess = new();

var parentWorkflow = new WorkflowBuilder(prefix)
    .AddEdge(prefix, subWorkflowExecutor)
    .AddEdge(subWorkflowExecutor, postProcess)
    .WithOutputFrom(postProcess)
    .Build();

BindAsExecutorでは、サブワークフローの型指定された入力と出力の種類が保持されます。親ワークフローは、サブワークフローが期待して生成する実際の型に基づいてメッセージをルーティングします。

AsAIAgent を使用したエージェント ラッピング

親ワークフローでエージェント ベースの Executor を使用する場合は、 AsAIAgent()を使用して内部ワークフローをエージェントに変換します。 WorkflowBuilderは、エージェントを Executor に自動的にラップします。

using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Workflows;

// Create agents for the inner workflow
AIAgent specialist1 = chatClient.AsAIAgent("You are specialist 1. Analyze the data.");
AIAgent specialist2 = chatClient.AsAIAgent("You are specialist 2. Validate the analysis.");

// Build the inner workflow
var innerWorkflow = new WorkflowBuilder(specialist1)
    .AddEdge(specialist1, specialist2)
    .Build();

// Convert the inner workflow to an agent
AIAgent innerWorkflowAgent = innerWorkflow.AsAIAgent(
    id: "analysis-pipeline",
    name: "Analysis Pipeline",
    description: "A sub-workflow that analyzes and validates data"
);

// Create agents for the parent workflow
AIAgent coordinator = chatClient.AsAIAgent("You are a coordinator. Delegate tasks to the team.");
AIAgent reviewer = chatClient.AsAIAgent("You are a reviewer. Review the final output.");

// Build the parent workflow with the sub-workflow
var parentWorkflow = new WorkflowBuilder(coordinator)
    .AddEdge(coordinator, innerWorkflowAgent)
    .AddEdge(innerWorkflowAgent, reviewer)
    .Build();

内部ワークフローは、親ワークフローの観点から 1 つのステップとして実行されます。 コーディネーターは、 specialist1 → specialist2内部で実行される分析パイプラインにメッセージを送信し、その結果をレビュー担当者に転送します。

入力と出力の種類

ワークフローがサブワークフローとして使用される場合、ワークフローは内部実行プログラムの型コントラクトを保持します。

BindAsExecutorでは、サブワークフロー実行プログラムは、内部ワークフローの開始 Executor と同じ入力型を受け入れ、内部ワークフローが生成するのと同じ出力の種類を送信します。 親ワークフローのエッジは、サブワークフローの予想される入力型と一致する出力の種類を持つ Executor を接続する必要があり、サブワークフローの出力の種類はダウンストリームの Executor の予想される入力と一致する必要があります。

AsAIAgentでは、サブワークフローはエージェントとしてラップされ、Agent Executor の入出力コントラクト (string、ChatMessage、IEnumerable<ChatMessage>) に従います。

出力動作

既定では、サブワークフローが ( YieldOutputAsync 経由で) 出力を生成すると、それらの出力は、親ワークフローの接続された Executor にメッセージとして転送されます。 これにより、ダウンストリームの Executor でサブワークフローの結果を処理できます。

ExecutorOptions クラスは、この動作を制御します。

AutoYieldOutputHandlerResultObjectが有効になっている場合、サブワークフロー出力は親の内部ルーティングをバイパスし、親ワークフローの呼び出し元に直接配信されます。

var options = new ExecutorOptions
{
    AutoYieldOutputHandlerResultObject = true,
};

ExecutorBinding subWorkflowExecutor = innerWorkflow.BindAsExecutor("SubWorkflow", options);

要求と応答

サブワークフローは、 要求と応答 のメカニズムを完全にサポートします。 サブワークフロー内の Executor が要求を送信する場合 (たとえば、人間の入力を要求する)、WorkflowHostExecutorは修飾されたポート ID を使ってRequestInfoEventを親ワークフローに転送します。サブワークフロー実行者の ID はポート ID の前に付加されます (たとえば、)。

この修飾により、親ワークフローが応答を受け取ったときに、応答を正しいサブワークフロー インスタンスに戻すことができます。 親ワークフローは、他の要求と同じ応答メカニズムを使用してサブワークフロー要求を処理します。

await using StreamingRun handle = await InProcessExecution.RunStreamingAsync(parentWorkflow, input);
await foreach (WorkflowEvent evt in handle.WatchStreamAsync())
{
    switch (evt)
    {
        case RequestInfoEvent requestInfoEvt:
            // The request may originate from the sub-workflow
            // Handle it and send the response back
            var response = requestInfoEvt.Request.CreateResponse(myResponseData);
            await handle.SendResponseAsync(response);
            break;

        case WorkflowOutputEvent outputEvt:
            Console.WriteLine($"Output: {outputEvt.Data}");
            break;
    }
}

しくみ

親ワークフローがサブワークフロー 実行プログラムにメッセージをルーティングする場合:

1. 入力配信 — メッセージは、内部ワークフローの開始 Executor に転送されます。 BindAsExecutorでは、メッセージの種類は、開始 Executor の予期される型と一致する必要があります。 AsAIAgentでは、メッセージはChatMessage形式に正規化されます。
2. 内部実行 — 内部ワークフローは、独自のスーパーステップ ループを実行します。
3. 出力コレクション — 内部ワークフローの出力イベントが収集されます。 BindAsExecutorでは、出力は元の型を保持します。 AsAIAgentでは、出力はエージェント応答メッセージに変換されます。
4. 要求の転送 — 内部ワークフローに保留中の要求がある場合は、処理のために親ワークフローに転送されます ( 要求と応答を参照)。
5. ダウンストリーム ディスパッチ — 結果のメッセージは、親ワークフローの次の Executor に送信されます。

内部ワークフローは独自の実行コンテキストを保持するため、その状態は親ワークフローから独立しています。

複数階層のネスト

サブワークフローは、任意の深さに入れ子にすることができます。 各レベルは、独自の実行コンテキストを保持します。

// Level 1: Data preparation pipeline
var dataPipeline = new WorkflowBuilder(fetcher)
    .AddEdge(fetcher, cleaner)
    .Build();

AIAgent dataPipelineAgent = dataPipeline.AsAIAgent(
    id: "data-pipeline",
    name: "Data Pipeline"
);

// Level 2: Analysis pipeline (contains the data pipeline)
var analysisPipeline = new WorkflowBuilder(dataPipelineAgent)
    .AddEdge(dataPipelineAgent, analyzer)
    .Build();

AIAgent analysisPipelineAgent = analysisPipeline.AsAIAgent(
    id: "analysis-pipeline",
    name: "Analysis Pipeline"
);

// Level 3: Top-level orchestration
var topWorkflow = new WorkflowBuilder(coordinator)
    .AddEdge(coordinator, analysisPipelineAgent)
    .AddEdge(analysisPipelineAgent, reporter)
    .Build();

エラー処理

サブワークフローが失敗すると、エラーは SubworkflowErrorEventとして親ワークフローに伝達されます。 親ワークフローは、イベント ストリームを通じて次のエラーを観察できます。

await foreach (WorkflowEvent evt in handle.WatchStreamAsync())
{
    if (evt is SubworkflowErrorEvent subError)
    {
        Console.WriteLine($"Sub-workflow '{subError.ExecutorId}' failed: {subError.Data}");
    }
}

サブワークフローで未処理の例外が発生した場合、親ワークフローの実行は続行されますが、サブワークフロー実行プログラムはそれ以降のメッセージの処理を停止します。

チェックポイント処理

親ワークフローでチェックポイントが取得されると、サブワークフロー エージェントのセッション状態が親 Executor のチェックポイント データの一部としてシリアル化されます。 復元時にセッション状態が逆シリアル化され、サブワークフローの状態をそのまま使用して親ワークフローを再開できます。

CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();

// Run the parent workflow with checkpointing
StreamingRun run = await InProcessExecution
    .RunStreamingAsync(parentWorkflow, input, checkpointManager);

await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    // Process events, including those from sub-workflows
}

// Resume from a checkpoint
CheckpointInfo checkpoint = run.Checkpoints[^1];
StreamingRun resumedRun = await InProcessExecution
    .ResumeStreamingAsync(parentWorkflow, checkpoint, checkpointManager);

Sub-Workflow の作成

Python では、サブワークフローを作成するには、WorkflowでWorkflowExecutorをラップし、親ワークフローに追加します。

from agent_framework import WorkflowBuilder, WorkflowExecutor

# Create agents for the inner workflow
specialist1 = client.as_agent(name="Specialist1", instructions="Analyze the data.")
specialist2 = client.as_agent(name="Specialist2", instructions="Validate the analysis.")

# Build the inner workflow
inner_workflow = (
    WorkflowBuilder(start_executor=specialist1)
    .add_edge(specialist1, specialist2)
    .build()
)

# Wrap as an executor
inner_workflow_executor = WorkflowExecutor(
    workflow=inner_workflow,
    id="analysis-pipeline",
)

# Create agents for the parent workflow
coordinator = client.as_agent(name="Coordinator", instructions="Delegate tasks to the team.")
reviewer = client.as_agent(name="Reviewer", instructions="Review the final output.")

# Build the parent workflow with the sub-workflow
parent_workflow = (
    WorkflowBuilder(start_executor=coordinator)
    .add_edge(coordinator, inner_workflow_executor)
    .add_edge(inner_workflow_executor, reviewer)
    .build()
)

内部ワークフローは、親ワークフローの観点から 1 つのステップとして実行されます。 コーディネーターは、 specialist1 → specialist2内部で実行される分析パイプラインにメッセージを送信し、その結果をレビュー担当者に転送します。

WorkflowExecutor パラメーター

サブワークフローの折り返し

Workflowインスタンスを親ワークフローに追加する前に、WorkflowExecutorで明示的にラップします。 エージェントは WorkflowBuilderに直接渡すことができますが、生の Workflow インスタンスにはこのラッパーが必要です。

from agent_framework import WorkflowExecutor

inner_workflow_executor = WorkflowExecutor(inner_workflow, id="analysis_pipeline")

parent_workflow = (
    WorkflowBuilder(start_executor=coordinator)
    .add_edge(coordinator, inner_workflow_executor)
    .add_edge(inner_workflow_executor, reviewer)
    .build()
)

明示的な折り返しを使用すると、次のことができます。

• 複数のエッジで参照する特定の Executor ID を割り当てます。
• グラフ全体で同じ WorkflowExecutor インスタンスを再利用します。

# Explicit wrapping — create the WorkflowExecutor yourself
inner_workflow_executor = WorkflowExecutor(
    workflow=inner_workflow,
    id="analysis-pipeline",
)

parent_workflow = (
    WorkflowBuilder(start_executor=coordinator)
    .add_edge(coordinator, inner_workflow_executor)
    .add_edge(inner_workflow_executor, reviewer)
    .build()
)

入力と出力の種類

WorkflowExecutorは、ラップされたワークフローから型シグネチャを継承します。

• 入力の種類 は、ラップされたワークフローの開始 Executor 入力の種類と一致します (さらに、転送された要求への応答を処理するための SubWorkflowResponseMessage )。
• 出力の種類 は、ラップされたワークフローの出力の種類と一致します。 サブワークフロー内の実行プログラムが要求応答に対応している場合、 SubWorkflowRequestMessage も出力の種類として含まれます。

つまり、親ワークフローのエッジは、サブワークフローの予想される入力型と一致する出力の種類を持つ Executor を接続する必要があります。 同様に、ダウンストリーム Executor は、サブワークフローによって生成される型を受け入れる必要があります。

# The sub-workflow's start executor accepts TextProcessingRequest
# So the parent executor must send TextProcessingRequest
class Orchestrator(Executor):
    @handler
    async def start(self, texts: list[str], ctx: WorkflowContext[TextProcessingRequest]) -> None:
        for text in texts:
            await ctx.send_message(TextProcessingRequest(text=text))

# The sub-workflow yields TextProcessingResult
# So the downstream executor must handle TextProcessingResult
class ResultCollector(Executor):
    @handler
    async def collect(self, result: TextProcessingResult, ctx: WorkflowContext) -> None:
        print(f"Received: {result}")

出力動作

既定 (allow_direct_output=False) では、サブワークフローが yield_outputを介して出力を生成すると、それらの出力は、 send_messageを使用して親ワークフローの接続された Executor にメッセージとして転送されます。 これにより、ダウンストリームの Executor は、親グラフの一部としてサブワークフローの結果を処理できます。

allow_direct_output=Trueすると、サブワークフローの出力が親ワークフローのイベント ストリームに直接生成されます。 サブワークフローの出力は親ワークフローの出力になり、親の内部 Executor ルーティングはバイパスされます。

# Outputs go directly to parent's event stream
sub_workflow_executor = WorkflowExecutor(
    workflow=inner_workflow,
    id="analysis-pipeline",
    allow_direct_output=True,
)

# The caller receives sub-workflow outputs directly
async for event in parent_workflow.run(input_data, stream=True):
    if event.type == "output":
        # This output came from the sub-workflow
        print(event.data)

子ワークフローからの中間排出量

"intermediate" 子ワークフロー内で生成されたイベントは、親のイベント ストリームを自動的にバブル アップします。 これらは、WorkflowExecutor 自身の id によるものとされます（もともとそれらを発生させた内部エグゼキューターによるものではありません）。これにより、カプセル化が保たれます。 重要なのは、これらのイベントは、親が独自の"intermediate"またはリストでWorkflowExecutorを指定する方法に関係なく、output_fromintermediate_output_fromすることです。

async for event in parent_workflow.run(input_data, stream=True):
    if event.type == "intermediate":
        # Attributed to the WorkflowExecutor id, e.g. "analysis-pipeline"
        print(f"[{event.executor_id}] intermediate: {event.data}")
    elif event.type == "output":
        print(f"Terminal output: {event.data}")

要求と応答

サブワークフローは、 要求と応答 のメカニズムを完全にサポートします。 サブワークフロー内の Executor が ctx.request_info()を呼び出すと、 WorkflowExecutor は要求をインターセプトし、 propagate_request 設定に基づいて処理します。

親ワークフロー (デフォルト) でのリクエストのインターセプト

propagate_request=False (既定) では、サブワークフローからの要求はSubWorkflowRequestMessageでラップされ、親ワークフローの接続された Executor に送信されます。 これにより、親 Executor は要求をローカルで処理できます。

from agent_framework import (
    SubWorkflowRequestMessage,
    SubWorkflowResponseMessage,
)


class ParentHandler(Executor):
    @handler
    async def handle_request(
        self,
        request: SubWorkflowRequestMessage,
        ctx: WorkflowContext[SubWorkflowResponseMessage],
    ) -> None:
        # Inspect the original request from the sub-workflow
        original_data = request.source_event.data

        # Create and send a response back to the sub-workflow
        response = request.create_response(my_response_data)
        await ctx.send_message(response, target_id=request.executor_id)

create_response() メソッドは、応答データ型が元の要求の予想される型と一致することを検証します。 型が一致しない場合は、 TypeError が発生します。

外部呼び出し元への要求の伝達

propagate_request=Trueでは、標準のrequest_info メカニズムを使用して、サブワークフローからの要求が親ワークフローのイベント ストリームに伝達されます。 親ワークフローの呼び出し元は、これらの要求を他の人間のループ内要求と同じ方法で処理します。

sub_workflow_executor = WorkflowExecutor(
    workflow=inner_workflow,
    id="analysis-pipeline",
    propagate_request=True,
)

# Run the parent workflow and handle propagated requests
result = await parent_workflow.run(input_data)
request_info_events = result.get_request_info_events()
if request_info_events:
    responses = {}
    for event in request_info_events:
        # Handle each request (e.g., ask a human)
        responses[event.request_id] = get_human_response(event.data)
    result = await parent_workflow.run(responses=responses)

しくみ

親ワークフローがメッセージを WorkflowExecutorにルーティングする場合:

1. 入力配信 — メッセージは、内部ワークフローの開始 Executor に転送されます。 メッセージの種類は、開始 Executor の予想される入力型と一致する必要があります。
2. 内部実行 — 内部ワークフローは、完了するまで、または外部入力が必要になるまで、独自のスーパーステップ ループを実行します。
3. 出力コレクション — 内部ワークフローの出力イベントは、 allow_direct_output 設定に基づいて収集および転送されます。
4. 要求の転送 — 内部ワークフローに保留中の要求がある場合は、 propagate_request 設定に基づいて転送されます ( 要求と応答を参照)。
5. 応答の累積 — WorkflowExecutor は、特定の実行に対して予想されるすべての応答を受信した場合にのみ、応答を収集し、サブワークフローを再開します。
6. ダウンストリーム ディスパッチ — 出力は、親ワークフローの次の Executor に送信されます。

サブワークフローは、親とは別に独自の内部状態を維持します。 メッセージは、 WorkflowExecutor を親グラフの残りの部分に接続するエッジを介してのみルーティングされます。入れ子になったレベル間でメッセージブロードキャストを行うメッセージはありません。

複数階層のネスト

サブワークフローは、任意の深さに入れ子にすることができます。 各レベルは、独自の実行コンテキストを保持します。

# Level 1: Data preparation pipeline
data_pipeline = (
    WorkflowBuilder(start_executor=fetcher)
    .add_edge(fetcher, cleaner)
    .build()
)

data_pipeline_executor = WorkflowExecutor(data_pipeline, id="data_pipeline")

# Level 2: Analysis pipeline (contains the data pipeline)
analysis_pipeline = (
    WorkflowBuilder(start_executor=data_pipeline_executor)
    .add_edge(data_pipeline_executor, analyzer)
    .build()
)

analysis_pipeline_executor = WorkflowExecutor(analysis_pipeline, id="analysis_pipeline")

# Level 3: Top-level orchestration
top_workflow = (
    WorkflowBuilder(start_executor=coordinator)
    .add_edge(coordinator, analysis_pipeline_executor)
    .add_edge(analysis_pipeline_executor, reporter)
    .build()
)

エラー処理

サブワークフローが失敗すると、エラーは親ワークフローに反映されます。 WorkflowExecutorは、サブワークフローから失敗したイベントをキャプチャし、親コンテキストでエラー イベントに変換します。

async for event in parent_workflow.run(input_data, stream=True):
    if event.type == "error":
        print(f"Sub-workflow failed: {event.details.message}")
    elif event.type == "output":
        print(event.data)

サブワークフローで未処理の例外が検出された場合、親ワークフローは、サブワークフローの ID を含む例外の詳細を含むエラー イベントを受け取ります。

チェックポイント処理

サブワークフローはチェックポイント処理をサポートします。 親ワークフローでチェックポイントが取得されると、 WorkflowExecutor は内部ワークフローの実行進行状況やキャッシュされたメッセージなど、内部状態をシリアル化します。 復元時にこの状態が逆シリアル化され、親ワークフローがサブワークフローをそのまま使用して再開できるようになります。

from agent_framework import FileCheckpointStorage, WorkflowBuilder

checkpoint_storage = FileCheckpointStorage(storage_path="./checkpoints")

# Build the parent workflow with checkpointing
parent_workflow = (
    WorkflowBuilder(
        start_executor=coordinator,
        checkpoint_storage=checkpoint_storage,
    )
    .add_edge(coordinator, inner_workflow_executor)
    .add_edge(inner_workflow_executor, reviewer)
    .build()
)

# Run with automatic checkpointing
async for event in parent_workflow.run("Analyze the dataset", stream=True):
    if event.type == "output":
        print(event.data)

# Resume from a checkpoint
checkpoints = await checkpoint_storage.list_checkpoints(workflow_name=parent_workflow.name)
async for event in parent_workflow.run(
    checkpoint_id=checkpoints[-1].checkpoint_id,
    checkpoint_storage=checkpoint_storage,
    stream=True,
):
    if event.type == "output":
        print(event.data)

Sub-Workflow の作成

Go では、 *workflow.Workflow を作成し、 inproc.BindSubworkflowAsExecutorを使用して親ワークフローにバインドすることで、サブワークフローを作成します。

package main

import (
    "context"
    "fmt"
    "slices"
    "strings"

    "github.com/microsoft/agent-framework-go/workflow"
    "github.com/microsoft/agent-framework-go/workflow/inproc"
)

func buildParentWorkflow() (*workflow.Workflow, error) {
    uppercase := workflow.NewExecutor("UppercaseExecutor", strings.ToUpper).Bind()
    reverse := workflow.NewExecutor("ReverseExecutor", reverseString).Bind()
    appendSuffix := workflow.NewExecutor("AppendSuffixExecutor", func(input string) string {
        return input + " [PROCESSED]"
    }).Bind()

    textProcessing, err := workflow.NewBuilder(uppercase).
        AddEdge(uppercase, reverse).
        AddEdge(reverse, appendSuffix).
        WithOutputFrom(appendSuffix).
        Build()
    if err != nil {
        return nil, err
    }

    textProcessingExecutor := inproc.BindSubworkflowAsExecutor(
        textProcessing,
        "TextProcessingSubWorkflow",
    )

    prefix := workflow.NewExecutor("PrefixExecutor", func(input string) string {
        return "INPUT: " + input
    }).Bind()
    postProcess := workflow.NewExecutor("PostProcessExecutor", func(input string) string {
        return "[FINAL] " + input + " [END]"
    }).Bind()

    return workflow.NewBuilder(prefix).
        AddEdge(prefix, textProcessingExecutor).
        AddEdge(textProcessingExecutor, postProcess).
        WithOutputFrom(postProcess).
        Build()
}

func reverseString(input string) string {
    runes := []rune(input)
    slices.Reverse(runes)
    return string(runes)
}

func runWorkflow(ctx context.Context, parentWorkflow *workflow.Workflow) error {
    run, err := inproc.Default.RunStreaming(ctx, parentWorkflow, "hello")
    if err != nil {
        return err
    }
    defer run.Close(ctx)

    for event, err := range run.WatchStream(ctx) {
        if err != nil {
            return err
        }
        if output, ok := event.(workflow.OutputEvent); ok {
            fmt.Println(output.Output)
        }
    }
    return nil
}

バインドされた子ワークフローは、親ワークフローの観点から 1 つの Executor として実行されます。 メッセージはバインディングを通じて入力され、子ワークフローは独自の内部グラフを実行し、子ワークフローの出力は親グラフにルーティングされます。

入力と出力の種類

バインドは、ラップされたワークフローのプロトコルを継承します。 親ワークフローは、子ワークフローの受け入れられた入力型と一致するランタイム型のメッセージを送信でき、バインドによって、子ワークフローの生成された出力の種類がメッセージの種類と出力の種類の両方として公開されます。

つまり、親ワークフローのエッジは、サブワークフローの受け入れられた入力と一致する出力型を持つ Executor を接続する必要があり、ダウンストリーム の Executor は子ワークフローによって生成される型を処理する必要があります。

type TextProcessingRequest struct {
    Text string
}

type TextProcessingResult struct {
    Text string
}

orchestrator := workflow.NewExecutor("Orchestrator", func(ctx *workflow.Context, texts []string) error {
    for _, text := range texts {
        if err := ctx.SendMessage("", TextProcessingRequest{Text: text}); err != nil {
            return err
        }
    }
    return nil
}).Bind()

collector := workflow.NewExecutor("Collector", func(result TextProcessingResult) {
    fmt.Println(result.Text)
}).Bind()

出力動作

子ワークフローが出力を生成すると、サブワークフロー バインドは、その出力をバインドから親ワークフローの接続された Executor にメッセージとして送信します。 親ワークフローもサブワークフロー バインドを WithOutputFrom でマークする場合、workflow.OutputEventがサブワークフロー バインド ID である親ExecutorIDとして同じ値が出力されます。

subWorkflowExecutor := inproc.BindSubworkflowAsExecutor(textProcessing, "TextProcessingSubWorkflow")
postProcess := workflow.NewExecutor("PostProcessExecutor", func(input string) string {
    return "[FINAL] " + input
}).Bind()

parentWorkflow, err := workflow.NewBuilder(subWorkflowExecutor).
    AddEdge(subWorkflowExecutor, postProcess).
    WithOutputFrom(subWorkflowExecutor).
    WithOutputFrom(postProcess).
    Build()

子ワークフロー内で生成されたカスタム ワークフロー イベントは、親イベント ストリームに転送されます。 サブワークフロー独自の開始およびスーパーステップ ライフサイクル イベントは内部に保持されるため、親ストリームは外部的に意味のあるイベントに集中し続けます。

要求と応答

サブワークフローは、 要求と応答 のメカニズムをサポートします。 サブワークフロー内の Executor が外部要求を投稿すると、サブワークフロー バインドは、バインド ID をプリペンドすることによって要求ポート ID を修飾します。 たとえば、 ApprovalPort という名前の子要求ポートは、親ワークフローで ApprovalSubWorkflow.ApprovalPort になります。

親ワークフローを介して子要求を表示するには、修飾 ID を持つ親 RequestPort を追加し、サブワークフロー バインドとそのポートの間で要求と応答をルーティングします。

import "reflect"

approvalPort := workflow.RequestPort{
    ID:       "ApprovalPort",
    Request:  reflect.TypeFor[string](),
    Response: reflect.TypeFor[bool](),
}

approvalWorkflow, err := workflow.NewBuilder(approvalPort.Bind()).
    Build()
if err != nil {
    return err
}

approvalSubWorkflow := inproc.BindSubworkflowAsExecutor(
    approvalWorkflow,
    "ApprovalSubWorkflow",
)

qualifiedApprovalPort := workflow.RequestPort{
    ID:       "ApprovalSubWorkflow.ApprovalPort",
    Request:  approvalPort.Request,
    Response: approvalPort.Response,
}
qualifiedApproval := qualifiedApprovalPort.Bind()

parentWorkflow, err := workflow.NewBuilder(approvalSubWorkflow).
    AddDirectEdge(approvalSubWorkflow, qualifiedApproval, false, externalRequestOnly).
    AddDirectEdge(qualifiedApproval, approvalSubWorkflow, false, externalResponseOnly).
    Build()

呼び出し元は、親ワークフロー ストリームからの要求を処理し、同じ実行ハンドルを介して応答を返します。 サブワークフロー バインドは、子ワークフローに応答を配信する前に、修飾されたプレフィックスを削除します。

run, err := inproc.Default.RunStreaming(ctx, parentWorkflow, "Approve deployment?")
if err != nil {
    return err
}
defer run.Close(ctx)

for event, err := range run.WatchStream(ctx) {
    if err != nil {
        return err
    }
    switch event := event.(type) {
    case workflow.RequestInfoEvent:
        response, err := event.Request.CreateResponse(true)
        if err != nil {
            return err
        }
        if err := run.SendResponse(ctx, response); err != nil {
            return err
        }
    case workflow.OutputEvent:
        fmt.Println(event.Output)
    }
}

述語を使用して、要求と応答のエッジを狭く保ちます。

func externalRequestOnly(msg any) bool {
    _, ok := msg.(*workflow.ExternalRequest)
    return ok
}

func externalResponseOnly(msg any) bool {
    _, ok := msg.(*workflow.ExternalResponse)
    return ok
}

しくみ

親ワークフローがメッセージをサブワークフロー バインドにルーティングする場合:

1. 入力配信 — バインディングは、子ワークフローの受け入れられた入力の種類に一致するメッセージを受け入れ、子ワークフローの開始 Executor にエンキューします。
2. 内部実行 — 子ワークフローは、同じインプロセス実行環境で実行され、独自のスーパーステップ ループを維持します。
3. 出力転送 — 子 workflow.OutputEvent 値は、バインドからダウンストリームの親エグゼキューターへのメッセージとして送信され、バインドが WithOutputFromにリストされている場合は親からも生成されます。
4. 要求の転送 — 子 workflow.RequestInfoEvent 要求は、修飾されたポート ID を使用して再出力され、親 RequestPort バインドを介してルーティングできます。
5. イベント転送 — カスタムの子ワークフロー イベントが親ストリームに追加されます。 エラーは、サブワークフロー ID が記録された親 workflow.ErrorEvent 値として表示されます。
6. ダウンストリームディスパッチ — 結果のメッセージは、親ワークフローエッジを通して続行されます。

子ワークフローは、その状態とメッセージ ルーティングを親とは別に保持します。 メッセージは、サブワークフロー バインドに接続されているエッジを介してのみ境界を越えます。

複数階層のネスト

サブワークフローは、任意の深さに入れ子にすることができます。 各子ワークフローは、子ワークフローを含むワークフローに追加される前にバインドされます。

fraudCheck, err := workflow.NewBuilder(analyzePatterns).
    AddEdge(analyzePatterns, calculateRiskScore).
    WithOutputFrom(calculateRiskScore).
    Build()
if err != nil {
    return err
}

fraudCheckExecutor := inproc.BindSubworkflowAsExecutor(fraudCheck, "FraudCheck")

payment, err := workflow.NewBuilder(validatePayment).
    AddEdge(validatePayment, fraudCheckExecutor).
    AddEdge(fraudCheckExecutor, chargePayment).
    WithOutputFrom(chargePayment).
    Build()
if err != nil {
    return err
}

paymentExecutor := inproc.BindSubworkflowAsExecutor(payment, "Payment")
shippingExecutor := inproc.BindSubworkflowAsExecutor(shipping, "Shipping")

orderWorkflow, err := workflow.NewBuilder(orderReceived).
    AddEdge(orderReceived, paymentExecutor).
    AddEdge(paymentExecutor, shippingExecutor).
    AddEdge(shippingExecutor, orderCompleted).
    WithOutputFrom(orderCompleted).
    Build()

エラー処理

子ワークフローがエラーを出力すると、サブワークフロー バインドによって親ワークフローに workflow.ErrorEvent として転送され、 SubWorkflowID がバインド ID に設定されます。 親ワークフローは、最上位のワークフロー エラーに使用するのと同じイベント ストリームを通じて、これらのエラーを観察できます。

for event, err := range run.WatchStream(ctx) {
    if err != nil {
        return err
    }
    switch event := event.(type) {
    case workflow.ErrorEvent:
        if event.SubWorkflowID != "" {
            return fmt.Errorf("sub-workflow %q failed: %w", event.SubWorkflowID, event.Error)
        }
        return event.Error
    case workflow.ExecutorFailedEvent:
        return fmt.Errorf("executor %q failed: %w", event.ExecutorID, event.Error)
    }
}

子イベントの転送中に発生したエラーは、サブワークフロー ID がアタッチされた親 workflow.ErrorEvent 値にも変換されます。

チェックポイント処理

サブワークフローはチェックポイント処理をサポートします。 親ワークフローがチェックポイントを取得すると、サブワークフロー バインドは、子ワークフローのチェックポイント マネージャーと、保留中の修飾された応答ポート マッピングを親 Executor 状態に格納します。 復元すると、子ワークフローは、保留中のリクエストを含むネストされた実行状態を保持したまま再開できます。

checkpointManager := checkpoint.NewInMemoryManager()
environment := inproc.Default.WithCheckpointing(checkpointManager)

var checkpoints []workflow.CheckpointInfo
run, err := environment.RunStreaming(ctx, parentWorkflow, "hello")
if err != nil {
    return err
}
defer run.Close(ctx)

for event, err := range run.WatchStream(ctx) {
    if err != nil {
        return err
    }
    if completed, ok := event.(workflow.SuperStepCompletedEvent); ok {
        if completed.CompletionInfo != nil && completed.CompletionInfo.CheckpointInfo != nil {
            checkpoints = append(checkpoints, *completed.CompletionInfo.CheckpointInfo)
        }
    }
}

if len(checkpoints) == 0 {
    return fmt.Errorf("no checkpoints were created")
}

resumedRun, err := environment.ResumeStreaming(ctx, parentWorkflow, checkpoints[len(checkpoints)-1])
if err != nil {
    return err
}
defer resumedRun.Close(ctx)

子ワークフローに保留中のリクエストがある状態でチェックポイントが復元されると、復元された親ワークフロー実行は、修飾子付きリクエスト情報イベントを再公開します。 呼び出し元は、その再公開されたリクエストに対するレスポンスを作成し、親実行ハンドルを介して送り返すことができます。