子工作流程

創建一個子工作流程

在 C# 中，你可以用兩種方式組合子工作流程：

• 直接綁定 — 使用 將工作流程嵌入至父工作流程中作為執行器。 這會保留子工作流程的原生輸入/輸出類型。
• 代理包裹 — 用 AsAIAgent() 來將一個工作流程轉換成代理，然後將代理加入父工作流程。 當父流程使用基於代理的執行者時，這非常有用。

使用 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 進行代理包裝

當父工作流程使用基於代理的執行者時，將內部工作流程轉換為代理，使用 AsAIAgent()。 此時 WorkflowBuilder 會自動將代理人包裹在執行者中：

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

從父工作流程的角度來看，內部工作流程是單步進行的。 協調者會將訊息傳送到分析管線，該管線內部執行 specialist1 → specialist2，然後將結果轉發給審查者。

輸入與輸出類型

當工作流程被用作子工作流程時，它會保留其內部執行者的型別契約。

當 BindAsExecutor時，子工作流程執行器接受與內部工作流程起始執行器相同的輸入類型，並傳送與內部工作流程產生的輸出類型相同。 父工作流程的邊必須連接輸出型別與子工作流程預期輸入型態相符的執行者，而子流程的輸出型別必須與下游執行者預期輸入相符。

當 AsAIAgent時，子工作流程被包裝成代理，並遵循 代理執行者 輸入/輸出合約（string， ChatMessage， IEnumerable<ChatMessage>）。

輸出行為

預設情況下，當子工作流程產生輸出（透過 YieldOutputAsync），這些輸出會以訊息形式轉發給父工作流程中連接的執行者。 這使下游執行者能夠處理子工作流程的結果。

該 ExecutorOptions 類別控制此行為：

啟用 AutoYieldOutputHandlerResultObject 時，子工作流程的輸出會繞過父工作流程的內部路由，直接傳送給父工作流程的呼叫者。

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

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

請求和回應

子工作流程完全支援 請求與回應 機制。 當子工作流程中的執行器發送請求（例如請求人工輸入）時，WorkflowHostExecutor 會將該 RequestInfoEvent 轉發給具有 合格埠碼 的父工作流程——該埠碼的 ID 前會加上子工作流程執行器的 ID（例如，SubWorkflow.GuessNumber）。

此限定確保父工作流程收到回應時，能將回應路由回正確的子工作流程實例。 父工作流程處理子工作流程請求時，使用與其他請求相同的回應機制：

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. 輸入傳遞 — 訊息會轉發到內部工作流程的啟動執行器。 當 BindAsExecutor，訊息類型必須符合啟動執行者的預期類型。 透過AsAIAgent，訊息會正規化為ChatMessage格式。
2. 內部執行 ——內部工作流程運行自己的超步驟迴圈。
3. 輸出收集 — 收集內部工作流程的輸出事件。 當 時 BindAsExecutor，輸出會保留其原始型別。 當 ， AsAIAgent輸出會轉換成代理回應訊息。
4. 請求轉發 — 若內部工作流程有待處理請求，則會轉交給母工作流程處理（參見 請求與回應）。
5. 下游派遣 — 產生的訊息會送達父工作流程中的下一個執行者。

由於內部工作流程維護自己的執行上下文，其狀態獨立於父工作流程。

多層巢狀結構

子工作流程可以任意深度地巢狀化。 每個層級都維持其獨立的執行上下文：

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

若子工作流程遇到未處理的例外，父工作流程的執行會繼續，但子工作流程執行器會停止處理後續訊息。

檢查點處理

當父工作流程設定檢查點時，子工作流程代理的會話狀態會序列化，作為父進程執行者的檢查點資料的一部分。 在還原時，會話狀態被反序列化，讓父工作流程能以子工作流程的狀態完整恢復。

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

創建一個子工作流程

在 Python 中，你會透過將 a Workflow 包裹在 a 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()
)

從父工作流程的角度來看，內部工作流程是單步進行的。 協調者會將訊息傳送到分析管線，該管線內部執行 specialist1 → specialist2，然後將結果轉發給審查者。

工作流程執行器參數

子工作流程的包裝

在將 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()
)

明確包裝可以讓你：

• 指定具體的執行者 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 繼承其型別簽名，來自封裝的工作流程。

• 輸入型別 與包裝後工作流程的起始執行者輸入類型相符（此外，也 SubWorkflowResponseMessage 用於處理轉發請求的回應）。
• 輸出類型 與包裝工作流程的輸出類型相符。 若子工作流程中任何執行器具備請求-回應功能， SubWorkflowRequestMessage 也會作為輸出型態被包含。

這表示父工作流程的邊必須連接那些輸出類型與子工作流程預期輸入類型匹配的執行者。 同樣地，下游執行者必須接受子工作流程產生的類型：

# 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。 這使得下游執行者能將子工作流程結果作為父圖的一部分處理。

當 allow_direct_output=True時，子工作流程的輸出會直接傳送到父工作流程的事件串流。 子工作流程的輸出會成為父工作流程的輸出，繞過父流程的內部執行者路由：

# 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_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}")

請求和回應

子工作流程完全支援 請求與回應 機制。 當子工作流程中的執行者呼叫 ctx.request_info() 時，WorkflowExecutor 攔截請求，並根據 propagate_request 設定進行處理。

在父工作流程中攔截請求（預設）

預設 propagate_request=False 情況下，子工作流程的請求會被 a SubWorkflowRequestMessage 包裹並傳送給父工作流程中已連接的執行者。 這讓父執行者能在本地處理請求：

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. 輸入傳遞 — 訊息會轉發到內部工作流程的啟動執行器。 訊息類型必須與啟動執行者的預期輸入類型相符。
2. 內部執行 — 內部工作流程會執行自己的超步驟迴圈直到完成，或直到需要外部輸入。
3. 輸出收集 — 根據設定收集並轉 allow_direct_output 發內部工作流程的輸出事件。
4. 請求轉發 — 若內部工作流程有待處理請求，則依設定轉發 propagate_request （參見 請求與回應）。
5. 回應累積 — WorkflowExecutor 收集回應，僅在所有預期回應都已收到後才恢復子工作流程。
6. 下游派遣 — 輸出會送至父工作流程中的下一個執行者。

子工作流程獨立於父工作流程維持自身的內部狀態。 訊息僅透過連接 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)

創建一個子工作流程

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

從父工作流程的角度來看，已繫結的子工作流程會作為單一執行器運作。 訊息透過綁定進入，子工作流程運行自己的內部圖，而子工作流程的輸出則被路由回父圖。

輸入與輸出類型

此綁定會繼承該封裝工作流程的協議。 父工作流程可以傳送其執行時類型與子工作流程接受的輸入類型相符的訊息，而綁定則將子工作流程產生的輸出類型同時以訊息類型與輸出類型的形式呈現。

這表示父工作流程的邊緣必須連接輸出類型與子工作流程接受輸入相符的執行者，而下游執行者則必須處理子工作流程所產生的類型：

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

輸出行為

當子工作流程產生輸出時，子工作流程綁定會將該輸出作為綁定的訊息發送給父工作流程中連接的執行者。 若父工作流程也以 WithOutputFrom 標記子工作流程繫結，則會以父層 workflow.OutputEvent 的形式輸出相同的值，其 ExecutorID 為子工作流程繫結 ID。

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

在子工作流程內發出的自訂工作流程事件，會被轉送到父工作流程的事件串流中。 子工作流程自身的開始與超步驟生命週期事件會被保留在內部，因此父流程能持續聚焦於外部有意義的事件。

請求和回應

子工作流程支援 請求與回應 機制。 當子工作流程中的執行器發布外部請求時，子工作流程綁定會先加上綁定 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. 輸入傳遞 — 綁定接受與子工作流程接受的輸入類型相符的訊息，並將其排入子工作流程的啟動執行器。
2. 內部執行 — 子工作流程在同一執行中執行環境中運行，並維持自己的超步驟迴圈。
3. 輸出轉發 — 子項的 workflow.OutputEvent 值會由繫結以訊息形式傳送至下游父執行器；如果該繫結列於 WithOutputFrom 中，這些值也會由父執行器產生。
4. 請求轉送 — 子 workflow.RequestInfoEvent 請求會以限定埠 ID 重新送出，並可經由父 RequestPort 繫結進行路由傳送。
5. 事件轉發 — 自訂子工作流程事件會被加入父串流。 錯誤會以父 workflow.ErrorEvent 值呈現，並記錄子工作流程 ID。
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 值。

檢查點處理

子工作流程支援檢查點。 當父工作流程建立檢查點時，子工作流程繫結會將子工作流程的檢查點管理器，以及任何待處理的符合條件回應埠對應關係，儲存在父執行器狀態中。 還原後，子工作流程可在其巢狀執行狀態保持完整的情況下繼續執行，包括待處理的請求。

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)

如果在子工作流程仍有待處理的請求時還原檢查點，還原後的父工作流程執行個體會重新發布合格的請求資訊事件。 呼叫者可以根據該重新發佈的請求建立回應，並透過父執行控制代碼將其傳回。