Sub-Alur Kerja

Membuat Sub-Workflow

Di C#, Anda menyusun sub-alur kerja dengan dua cara:

• Pengikatan langsung — gunakan BindAsExecutor() untuk menyematkan alur kerja secara langsung sebagai pelaksana dalam alur kerja induk. Ini mempertahankan jenis input/output asli sub-alur kerja.
• Pembungkusan agen — gunakan AsAIAgent() untuk mengonversi alur kerja menjadi agen, lalu tambahkan agen ke alur kerja induk. Ini berguna ketika alur kerja induk menggunakan pelaksana berbasis agen.

Pengikatan Langsung dengan BindAsExecutor

Metode BindAsExecutor() ekstensi mengonversi alur kerja menjadi ExecutorBinding yang dapat ditambahkan langsung ke alur kerja induk:

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

Dengan BindAsExecutor, tipe input dan output sub-alur kerja dipertahankan — alur kerja induk merutekan pesan berdasarkan jenis aktual yang diharapkan dan dihasilkan oleh sub-alur kerja.

Pembungkusan Agen dengan AsAIAgent

Saat alur kerja induk menggunakan pelaksana berbasis agen, konversikan alur kerja dalam ke agen menggunakan AsAIAgent(). WorkflowBuilder secara otomatis membungkus agen dalam seorang pelaksana.

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

Alur kerja internal dijalankan dalam satu langkah dari sudut pandang alur kerja induk. Koordinator mengirim pesan ke alur analisis, yang dijalankan specialist1 → specialist2secara internal , lalu meneruskan hasilnya ke peninjau.

Jenis Input dan Output

Saat alur kerja digunakan sebagai sub-alur kerja, alur kerja tersebut mempertahankan kontrak tipe dari eksekutor internalnya.

Dengan BindAsExecutor, eksekutor sub-alur kerja menerima jenis input yang sama dengan pelaksana mulai alur kerja dalam, dan mengirimkan jenis output yang sama dengan yang dihasilkan oleh alur kerja dalam. Tepi alur kerja induk harus menghubungkan pelaksana yang jenis outputnya cocok dengan jenis input sub-alur kerja yang diharapkan, dan jenis output sub-alur kerja harus cocok dengan input eksekutor hilir yang diharapkan.

Dengan AsAIAgent, sub-alur kerja dikemas sebagai sebuah agen dan mengikuti kontrak input/output Agent Executor (string, ChatMessage, IEnumerable<ChatMessage>).

Perilaku Output

Secara default, ketika sub-alur kerja menghasilkan output (melalui YieldOutputAsync), output tersebut diteruskan sebagai pesan ke pelaksana yang terhubung di alur kerja induk. Ini memungkinkan pelaksana hilir untuk memproses hasil sub-alur kerja.

Kelas ExecutorOptions mengontrol perilaku ini:

Saat AutoYieldOutputHandlerResultObject diaktifkan, output sub-alur kerja melewati perutean internal induk dan dikirimkan langsung ke pemanggil alur kerja induk.

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

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

Permintaan dan Respons

Sub-alur kerja sepenuhnya mendukung mekanisme permintaan dan respons . Ketika eksekutor di dalam sub-alur kerja mengirimkan permintaan (contohnya, untuk meminta masukan manusia), WorkflowHostExecutor meneruskan permintaan tersebut ke alur kerja induk dengan melampirkan RequestInfoEvent — di mana ID eksekutor sub-alur kerja ditambahkan di depan ID port (contohnya, ).

Kualifikasi ini memastikan bahwa ketika alur kerja induk menerima respons, alur kerja dapat merutekan respons kembali ke instans sub-alur kerja yang benar. Alur kerja induk menangani permintaan sub-alur kerja menggunakan mekanisme respons yang sama dengan permintaan lainnya:

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

Cara Kerjanya

Saat alur kerja induk merutekan pesan ke pelaksana sub-alur kerja:

1. Pengiriman input — pesan diteruskan ke pelaksana yang memulai alur kerja internal. Dengan BindAsExecutor, jenis pesan harus cocok dengan jenis eksekutor awal yang diharapkan. Dengan AsAIAgent, pesan dinormalisasi ke ChatMessage format.
2. Eksekusi dalam — alur kerja dalam menjalankan loop superstepnya sendiri.
3. Pengumpulan output — peristiwa output dari alur kerja internal dikumpulkan. Dengan BindAsExecutor, output mempertahankan jenis aslinya. Dengan AsAIAgent, output dikonversi ke pesan respons agen.
4. Penerusan permintaan — jika alur kerja dalam memiliki permintaan yang tertunda, alur kerja tersebut diteruskan ke alur kerja induk untuk penanganan (lihat Permintaan dan Respons).
5. Pengiriman hilir — pesan yang dihasilkan dikirim ke pelaksana berikutnya dalam alur kerja induk.

Karena alur kerja dalam mempertahankan konteks eksekusinya sendiri, statusnya independen dari alur kerja induk.

Penyusunan Bertingkat Ganda

Sub-alur kerja dapat disarangkan ke kedalaman arbitrer. Setiap tingkat mempertahankan konteks eksekusinya sendiri:

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

Penanganan Kesalahan

Ketika sub-alur kerja gagal, kesalahan disebarluaskan ke alur kerja induk sebagai SubworkflowErrorEvent. Alur kerja induk dapat mengamati kesalahan ini melalui aliran peristiwanya:

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

Jika sub-alur kerja mengalami pengecualian yang tidak tertangani, eksekusi alur kerja induk berlanjut tetapi pelaksana sub-alur kerja berhenti memproses pesan lebih lanjut.

Titik Pemeriksaan

Ketika titik pemeriksaan diambil pada alur kerja induk, status sesi agen sub-alur kerja diserialisasikan sebagai bagian dari data titik pemeriksaan pelaksana induk. Saat dipulihkan, status sesi dideserialisasi, memungkinkan alur kerja induk dilanjutkan dengan status sub-alur kerja utuh.

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

Membuat Sub-Workflow

Di Python, Anda membuat sub-alur kerja dengan membungkus Workflow dalam WorkflowExecutor dan menambahkannya ke alur kerja induk.

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

Alur kerja internal dijalankan dalam satu langkah dari sudut pandang alur kerja induk. Koordinator mengirim pesan ke alur analisis, yang dijalankan specialist1 → specialist2secara internal , lalu meneruskan hasilnya ke peninjau.

Parameter-parameter untuk WorkflowExecutor

Membungkus sub-alur kerja

Bungkus Workflow instans secara eksplisit dalam WorkflowExecutor sebelum menambahkannya ke alur kerja induk. Agent dapat diteruskan langsung ke WorkflowBuilder, tetapi instance Workflow mentah memerlukan wrapper ini.

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

Pembungkusan eksplisit memungkinkan Anda:

• Tetapkan ID pelaksana tertentu untuk referensi di beberapa tepi.
• Gunakan kembali instans yang sama WorkflowExecutor di seluruh grafik.

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

Jenis Input dan Output

Mewarisi WorkflowExecutor tanda tangan jenisnya dari alur kerja yang dibungkus:

• Jenis input harus sesuai dengan jenis input pelaksana mulai dari alur kerja yang dibungkus (ditambah SubWorkflowResponseMessage untuk menangani tanggapan terhadap permintaan yang diteruskan).
• Jenis output cocok dengan jenis output alur kerja yang dibungkus. Jika ada eksekutor dalam sub-alur kerja yang mampu merespons permintaan, SubWorkflowRequestMessage juga disertakan sebagai jenis output.

Ini berarti tepi alur kerja induk harus menghubungkan pelaksana yang jenis outputnya cocok dengan jenis input sub-alur kerja yang diharapkan. Demikian pula, eksekutor hilir harus menerima jenis yang dihasilkan sub-alur kerja:

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

Perilaku Output

Secara default (allow_direct_output=False), ketika sub-alur kerja menghasilkan output melalui yield_output, output tersebut diteruskan sebagai pesan ke pelaksana yang terhubung di alur kerja induk menggunakan send_message. Ini memungkinkan pelaksana hilir untuk memproses hasil sub-alur kerja sebagai bagian dari grafik induk.

Ketika allow_direct_output=True, output sub-alur kerja dihasilkan langsung ke aliran peristiwa alur kerja induk. Output dari sub-alur kerja menjadi output dari alur kerja induk, tanpa melalui perutean eksekutor internal dari alur kerja induk.

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

Emisi menengah dari alur kerja anak

"intermediate" peristiwa yang dihasilkan di dalam alur kerja anak secara otomatis diteruskan melalui aliran peristiwa induk. Mereka diatribusikan ke WorkflowExecutor itu sendiri id (bukan ke eksekutor internal yang semula menghasilkannya), yang menjaga enkapsulasi. Yang terpenting, peristiwa ini tetap mempertahankan "intermediate" label terlepas dari bagaimana elemen induk menetapkan WorkflowExecutor dalam daftar output_from atau intermediate_output_from miliknya sendiri.

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

Permintaan dan Respons

Sub-alur kerja sepenuhnya mendukung mekanisme permintaan dan respons . Ketika eksekutor di dalam sub-alur kerja memanggil ctx.request_info(), WorkflowExecutor mencegat permintaan dan menanganinya berdasarkan pengaturan propagate_request.

Mencegat Permintaan di Alur Kerja Induk (Default)

Dengan propagate_request=False (default), permintaan dari sub-alur kerja dibungkus dalam SubWorkflowRequestMessage dan dikirim ke pelaksana yang terhubung dalam alur kerja induk. Ini memungkinkan eksekutor induk untuk menangani permintaan secara lokal:

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)

Metode create_response() memvalidasi bahwa jenis data respons cocok dengan jenis yang diharapkan dari permintaan asli. Jika jenis tidak cocok, akan TypeError dinaikkan.

Mendistribusikan Permintaan ke Pemanggil Eksternal

Dengan propagate_request=True, permintaan dari sub-alur kerja disebarluaskan ke aliran peristiwa alur kerja induk menggunakan mekanisme standar request_info . Pemanggil alur kerja induk menangani permintaan ini dengan cara yang sama seperti permintaan human-in-the-loop lainnya:

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)

Cara Kerjanya

Saat alur kerja induk merutekan pesan ke WorkflowExecutor:

1. Pengiriman input — pesan diteruskan ke pelaksana yang memulai alur kerja internal. Jenis pesan harus cocok dengan jenis input yang diharapkan pelaksana awal.
2. Eksekusi dalam — alur kerja internal menjalankan loop superstepnya sendiri hingga selesai, atau sampai membutuhkan masukan eksternal.
3. Pengumpulan keluaran — peristiwa keluaran dari dalam alur kerja dikumpulkan dan diteruskan berdasarkan pengaturan allow_direct_output.
4. Penerusan permintaan — jika alur kerja dalam memiliki permintaan yang tertunda, mereka diteruskan berdasarkan propagate_request pengaturan (lihat Permintaan dan Respons).
5. Akumulasi respons — WorkflowExecutor mengumpulkan respons dan melanjutkan sub-alur kerja hanya ketika semua respons yang diharapkan untuk eksekusi tertentu telah diterima.
6. Pengiriman hilir — output dikirim ke pelaksana berikutnya dalam alur kerja induk.

Sub-alur kerja mempertahankan status internalnya sendiri secara independen dari induk. Pesan hanya dirutekan melalui tepi yang menghubungkan WorkflowExecutor dengan sisa grafik induk — tidak ada penyiaran pesan di seluruh tingkat pewarisan.

Penyusunan Bertingkat Ganda

Sub-alur kerja dapat disarangkan ke kedalaman arbitrer. Setiap tingkat mempertahankan konteks eksekusinya sendiri:

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

Penanganan Kesalahan

Saat sub-alur kerja gagal, kesalahan disebarluaskan ke alur kerja induk. Menangkap peristiwa gagal dari sub-alur kerja dan mengonversinya menjadi peristiwa kesalahan dalam konteks induk:

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)

Jika sub-alur kerja mengalami pengecualian yang tidak tertangani, alur kerja induk menerima peristiwa kesalahan dengan detail pengecualian, termasuk ID sub-alur kerja.

Titik Pemeriksaan

Alur kerja sub mendukung titik pemeriksaan. Ketika titik pemeriksaan dilakukan pada alur kerja induk, WorkflowExecutor menyerialkan status internalnya, termasuk kemajuan eksekusi alur kerja bagian dalam dan segala pesan yang di-cache. Saat dipulihkan, status ini dideserialisasi, memungkinkan alur kerja induk dilanjutkan dengan sub-alur kerja tetap utuh.

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)

Membuat Sub-Workflow

Di Go, Anda membuat sub-alur kerja dengan membangun *workflow.Workflow dan mengikatnya ke dalam alur kerja induk dengan 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
}

Alur kerja anak terikat berjalan sebagai satu pelaksana dari perspektif alur kerja induk. Pesan masuk melalui pengikatan, alur kerja anak menjalankan grafik internalnya sendiri, dan output alur kerja anak dirutekan kembali ke grafik induk.

Jenis Input dan Output

Binding ini mewarisi protokol dari workflow yang dibungkus. Alur kerja induk dapat mengirim pesan yang jenis runtime-nya cocok dengan jenis input yang diterima alur kerja anak, dan pengikatan mengekspos jenis output yang dihasilkan alur kerja anak sebagai jenis pesan dan jenis output.

Ini berarti tepi alur kerja induk harus menghubungkan pelaksana yang jenis outputnya cocok dengan input yang diterima sub-alur kerja, dan eksekutor hilir harus menangani jenis yang dihasilkan oleh alur kerja anak:

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

Perilaku Output

Saat alur kerja anak menghasilkan output, pengikatan sub-alur kerja mengirimkan output tersebut sebagai pesan dari pengikatan ke pelaksana yang terhubung di alur kerja induk. Jika alur kerja induk juga menandai pengikatan sub-alur kerja dengan WithOutputFrom, nilai yang sama dihasilkan sebagai workflow.OutputEvent induk dengan ExecutorID berupa ID pengikatan sub-alur kerja.

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

Peristiwa alur kerja kustom yang dihasilkan dalam alur kerja anak diteruskan ke aliran peristiwa induk. Peristiwa siklus hidup saat mulai dan superstep milik sub-workflow itu sendiri tetap bersifat internal agar stream induk tetap berfokus pada peristiwa yang bermakna secara eksternal.

Permintaan dan Respons

Sub-alur kerja mendukung mekanisme permintaan dan respons . Saat eksekutor di dalam sub-workflow mengirim permintaan eksternal, binding sub-workflow membuat ID port permintaan menjadi fully qualified dengan menambahkan ID binding di depannya. Misalnya, port permintaan anak bernama ApprovalPort menjadi ApprovalSubWorkflow.ApprovalPort dalam alur kerja induk.

Untuk mengekspos permintaan anak melalui alur kerja induk, tambahkan RequestPort induk dengan ID yang memenuhi syarat dan rutekan permintaan serta respons antara pengikatan sub-alur kerja dan port tersebut:

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

Pemanggil menangani permintaan dari aliran alur kerja induk dan mengirimkan respons kembali melalui handle eksekusi yang sama. Pengikatan sub-alur kerja menghapus prefiks yang memenuhi syarat sebelum mengirimkan respons ke alur kerja anak.

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

Gunakan predikat untuk menjaga tepi permintaan dan respons tetap sempit:

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

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

Cara Kerjanya

Saat alur kerja induk mengarahkan pesan ke pengikatan sub-alur kerja:

1. Pengiriman input — binding menerima pesan yang sesuai dengan tipe input yang diterima alur kerja anak dan memasukkannya ke antrean eksekutor awal alur kerja anak.
2. Eksekusi internal — alur kerja turunan berjalan dalam lingkungan eksekusi dalam proses yang sama dan mempertahankan loop superstep-nya sendiri.
3. Penerusan output — nilai turunan workflow.OutputEvent dikirim sebagai pesan dari binding ke eksekutor induk downstream, dan juga dikembalikan oleh induk jika binding tercantum dalam WithOutputFrom.
4. Penerusan permintaan — permintaan anak workflow.RequestInfoEvent dipancarkan ulang dengan ID port yang memenuhi syarat dan dapat dirutekan melalui binding induk RequestPort.
5. Penerusan peristiwa — peristiwa alur kerja anak kustom ditambahkan ke aliran induk. Kesalahan ditampilkan sebagai nilai induk workflow.ErrorEvent dengan ID subalur kerja yang tercatat.
6. Penerusan downstream — pesan yang dihasilkan terus melalui cabang alur kerja induk.

Alur kerja anak menjaga status dan perutean pesannya terpisah dari induk. Pesan melewati batas hanya melalui tepi yang tersambung ke pengikatan sub-alur kerja.

Penyusunan Bertingkat Ganda

Sub-alur kerja dapat disarangkan ke kedalaman arbitrer. Setiap alur kerja anak terikat sebelum ditambahkan ke alur kerja yang berisinya:

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

Penanganan Kesalahan

Saat alur kerja anak menghasilkan error, pengikatan sub-alur kerja meneruskannya ke alur kerja induk sebagai workflow.ErrorEvent dan menetapkan SubWorkflowID ke ID pengikatan. Alur kerja induk dapat mengamati kesalahan ini melalui aliran peristiwa yang sama yang digunakannya untuk kesalahan alur kerja tingkat atas:

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

Kesalahan yang terjadi saat meneruskan event turunan juga dikonversi menjadi nilai workflow.ErrorEvent induk dengan ID sub-workflow terlampir.

Titik Pemeriksaan

Alur kerja sub mendukung titik pemeriksaan. Ketika alur kerja induk mengambil titik pemeriksaan, pengikatan sub-alur kerja menyimpan manajer titik pemeriksaan alur kerja anak dan pemetaan port respons yang memenuhi syarat yang tertunda dalam status eksekutor induk. Saat dipulihkan, alur kerja anak dapat melanjutkan dengan status eksekusi bertingkat yang tetap utuh, termasuk permintaan yang tertunda.

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)

Jika titik pemeriksaan dipulihkan saat alur kerja anak memiliki permintaan yang tertunda, eksekusi induk yang dipulihkan menerbitkan ulang peristiwa info permintaan yang memenuhi syarat. Pemanggil dapat membuat respons dari permintaan yang dipublikasikan ulang tersebut dan mengirimkannya kembali melalui handle proses induk.