하위 워크플로우

Sub-Workflow 만들기

C#에서는 다음 두 가지 방법으로 하위 워크플로를 작성합니다.

• 직접 바인딩 - 워크플로를 부모 워크플로에 직접 실행기로 포함하는 데 사용합니다 BindAsExecutor() . 이렇게 하면 하위 워크플로의 네이티브 입력/출력 형식이 유지됩니다.
• 에이전트 래핑 - 워크플로를 에이전트로 변환한 다음 부모 워크플로에 에이전트를 추가하는 데 사용합니다 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 앞에 추가되어 전송됩니다(예: ).

이 정규화는 부모 워크플로가 응답을 받을 때 응답을 올바른 하위 워크플로 인스턴스로 다시 라우팅할 수 있도록 합니다. 부모 워크플로는 다른 요청과 동일한 응답 메커니즘을 사용하여 하위 워크플로 요청을 처리합니다.

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

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

내부 워크플로는 부모 워크플로의 관점에서 단일 단계로 실행됩니다. 코디네이터는 내부적으로 실행되는 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()
)

명시적 래핑을 사용하면 다음을 수행할 수 있습니다:

• 여러 에지에서 참조에 대한 특정 실행기 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 (기본값)을 사용하면 하위 워크플로의 요청이 부모 워크플로의 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() 응답 데이터 형식이 원래 요청의 예상 형식과 일치하는지 확인합니다. 형식이 일치하지 않으면 a 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)

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
}

바인딩된 자식 워크플로는 부모 워크플로의 관점에서 하나의 실행기로 실행됩니다. 메시지는 바인딩을 통해 입력되고, 자식 워크플로는 자체 내부 그래프를 실행하고, 자식 워크플로의 출력은 부모 그래프로 다시 라우팅됩니다.

입력 및 출력 형식

바인딩은 래핑된 워크플로의 프로토콜을 상속합니다. 부모 워크플로는 런타임 형식이 자식 워크플로의 허용된 입력 형식과 일치하는 메시지를 보낼 수 있으며, 바인딩은 자식 워크플로의 생성 출력 형식을 메시지 형식과 출력 형식 모두로 노출합니다.

즉, 부모 워크플로의 에지는 출력 형식이 하위 워크플로의 허용된 입력과 일치하는 실행기를 연결해야 하며 다운스트림 실행기는 자식 워크플로에서 생성한 형식을 처리해야 합니다.

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로 표시하면, 동일한 값이 ExecutorID가 하위 워크플로 바인딩 ID인 부모 workflow.OutputEvent로 출력됩니다.

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를 붙여 해당 요청 포트 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. 내부 실행 - 자식 워크플로는 동일한 In-process 실행 환경에서 실행되며 자체 슈퍼스텝 루프를 유지 관리합니다.
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 전달하고 바인딩 ID로 설정합니다 SubWorkflowID . 부모 워크플로는 최상위 워크플로 오류에 사용하는 것과 동일한 이벤트 스트림을 통해 이러한 오류를 관찰할 수 있습니다.

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)

자식 워크플로에 보류 중인 요청이 있는 상태에서 체크포인트가 복원되면, 복원된 부모 워크플로 실행은 해당 요청 정보 이벤트를 재게시합니다. 호출자는 해당 재게시된 요청에서 응답을 생성하여 부모 실행 핸들을 통해 이를 돌려보낼 수 있습니다.