Podprocesy

Vytvoření Sub-Workflow

V jazyce C# vytváříte dílčí pracovní postupy dvěma způsoby:

• Přímá vazba – slouží BindAsExecutor() k vložení pracovního postupu přímo jako výkonný mechanismus do nadřazeného pracovního postupu. Tím se zachová nativní typy vstupu a výstupu dílčího pracovního postupu.
• Zabalení agenta – slouží AsAIAgent() k převodu pracovního postupu na agenta a následné přidání agenta do nadřazeného pracovního postupu. To je užitečné, pokud nadřazený pracovní postup používá vykonavatele založené na agentech.

Přímá vazba s BindAsExecutorem

Metoda BindAsExecutor() rozšíření převede pracovní postup na ExecutorBinding pracovní postup, který lze přidat přímo do nadřazeného pracovního postupu:

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

S BindAsExecutor jsou zachovány typované vstupní a výstupní typy dílčího pracovního postupu — nadřazený pracovní postup směruje zprávy na základě skutečných typů, které dílčí pracovní postup očekává a generuje.

Zabalení agenta pomocí agenta AsAIAgent

Pokud nadřazený pracovní postup používá exekutory založené na agentech, převeďte vnitřní pracovní postup na agenta pomocí AsAIAgent(). Automaticky WorkflowBuilder zabalí agenta do exekutoru:

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

Vnitřní pracovní postup se spouští jako jeden krok z pohledu nadřazeného pracovního postupu. Koordinátor odesílá zprávy do analytické pipeline, která interně spouští specialist1 → specialist2, a poté předá výsledek recenzentovi.

Vstupní a výstupní typy

Pokud se pracovní postup používá jako dílčí pracovní postup, zachová typové smlouvy svých interních exekutorů.

V případě BindAsExecutor výkonný modul pro dílčí pracovní postup přijímá stejné vstupní typy jako spouštěcí modul vnitřního pracovního postupu a odesílá stejné výstupní typy, které vnitřní pracovní postup vytváří. Hrany nadřazeného pracovního postupu musí propojit exekutory, jejichž výstupní typy odpovídají očekávaným vstupním typům dílčího pracovního postupu a výstupní typy dílčího pracovního postupu musí odpovídat očekávaným vstupům podřízených exekutorů.

Dílčí pracovní postup je zabalen jako agent a řídí se vstupními/výstupními kontrakty Agent Executor (AsAIAgent, , string).

Chování systému vůči výstupu

Pokud dílčí pracovní postup ve výchozím nastavení vytváří výstupy (prostřednictvím YieldOutputAsync), tyto výstupy se přeposílají jako zprávy připojeným exekutorům v nadřazeným pracovním postupu. To umožňuje podřízeným exekutorům zpracovávat výsledky dílčího pracovního postupu.

Třída ExecutorOptions řídí toto chování.

Pokud je možnost AutoYieldOutputHandlerResultObject povolena, výstupy dílčího pracovního postupu obcházejí interní směrování nadřazeného pracovního postupu a jsou doručovány přímo volajícímu nadřazeného pracovního postupu.

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

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

Žádosti a odpovědi

Dílčí pracovní postupy plně podporují mechanismus požadavků a odpovědí . Když exekutor uvnitř dílčího pracovního postupu odešle požadavek (například k vyžádání lidského vstupu), WorkflowHostExecutor přepošle RequestInfoEvent nadřazenému pracovnímu postupu s kvalifikovaným ID portu – ID exekutoru dílčího pracovního postupu je předloženo ID portu (například SubWorkflow.GuessNumber).

Tato kvalifikace zajišťuje, že když nadřazený pracovní postup obdrží odpověď, může odpověď směrovat zpět do správné instance dílčího pracovního postupu. Nadřazený pracovní postup zpracovává požadavky dílčího pracovního postupu pomocí stejného mechanismu odpovědi jako jakýkoli jiný požadavek:

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

Jak to funguje

Když nadřazený pracovní postup směruje zprávu do exekutoru dílčího pracovního postupu:

1. Doručení vstupu – zpráva se přesměruje do spouštěče vnitřního pracovního postupu. Při BindAsExecutorpoužití musí typ zprávy odpovídat očekávaným typům spouštěcího exekutoru. U AsAIAgent, zprávy jsou normalizovány do ChatMessage formátu.
2. Vnitřní spuštění – vnitřní pracovní postup spouští vlastní smyčku superkroků.
3. Výstupní kolekce – shromažďují se výstupní události vnitřního pracovního postupu. Výstupy s BindAsExecutor, zachovávají své původní typy. Výstupy jsou převedeny na zprávy odpovědí agenta pomocí AsAIAgent.
4. Předávání žádostí – pokud má vnitřní pracovní postup čekající žádosti, jsou přeposlány do nadřazeného pracovního postupu ke zpracování (viz Požadavky a odpovědi).
5. Sestupné předání – výsledné zprávy se odesílají do dalšího exekutora v mateřském pracovním postupu.

Vzhledem k tomu, že vnitřní pracovní postup udržuje svůj vlastní kontext spuštění, je jeho stav nezávislý na nadřazeném pracovním postupu.

Víceúrovňové vnoření

Dílčí pracovní postupy lze vnořit do libovolné hloubky. Každá úroveň udržuje svůj vlastní kontext spuštění:

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

Zpracování chyb

Pokud dílčí pracovní postup selže, chyba se rozšíří do nadřazeného pracovního postupu jako .SubworkflowErrorEvent Nadřazený pracovní postup může sledovat tyto chyby prostřednictvím streamu událostí:

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

Pokud dílčí pracovní postup narazí na neošetřenou výjimku, provádění nadřazeného pracovního postupu pokračuje, ale exekutor dílčího pracovního postupu přestane zpracovávat další zprávy.

Vytváření kontrolních bodů

Když se v nadřazeném pracovním postupu uloží kontrolní bod, stav relace agenta dílčího pracovního postupu se serializuje jako součást dat kontrolního bodu nadřazeného workflowu. Při obnovení je stav relace deserializován, což umožňuje nadřazenému pracovnímu postupu pokračovat s podřízeným pracovním postupem beze změny stavu.

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

Vytvoření Sub-Workflow

V Pythonu vytvoříte dílčí pracovní postup tak, že zapouzdříte Workflow do WorkflowExecutor a přidáte ho do nadřazeného pracovního postupu.

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

Vnitřní pracovní postup se spouští jako jeden krok z pohledu nadřazeného pracovního postupu. Koordinátor odesílá zprávy do analytické pipeline, která interně spouští specialist1 → specialist2, a poté předá výsledek recenzentovi.

Parametry vykonavatele workflows

Zabalení dílčích pracovních postupů

Před přidáním instancí Workflow do nadřazeného pracovního postupu je výslovně zabalte do WorkflowExecutor. Agenti lze předat přímo do WorkflowBuilder, ale surové instance Workflow vyžadují tento obal.

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

Explicitní zabalení umožňuje:

• Přiřaďte konkrétní ID vykonavatele pro použití na více hranách.
• Znovu použijte stejnou WorkflowExecutor instanci v grafu.

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

Vstupní a výstupní typy

WorkflowExecutor zdědí podpis svého typu ze zabaleného workflowu.

• Vstupní typy odpovídají vstupním typům spouštěcího exekutoru zabaleného pracovního postupu (plus SubWorkflowResponseMessage pro zpracování odpovědí na předávané požadavky).
• Typy výstupu odpovídají výstupním typům zabaleného pracovního postupu. Pokud některý exekutor v dílčím pracovním postupu je schopný reagovat na požadavky, SubWorkflowRequestMessage je také zahrnut jako výstupní typ.

To znamená, že hrany nadřazeného pracovního postupu musí připojit exekutory, jejichž výstupní typy odpovídají očekávaným vstupním typům dílčího pracovního postupu. Podobně podřízené exekutory musí přijímat typy, které dílčí pracovní postup vytváří:

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

Chování systému vůči výstupu

Ve výchozím nastavení (allow_direct_output=False), když dílčí pracovní postup vytváří výstupy prostřednictvím yield_output, tyto výstupy se přeposílají jako zprávy připojeným exekutorům v nadřazeného pracovním postupu pomocí send_message. To umožňuje podřízeným exekutorům zpracovávat výsledky dílčího pracovního postupu jako součást nadřazeného grafu.

Když allow_direct_output=True jsou výstupy z dílčího pracovního postupu předány přímo na datový proud událostí nadřazeného pracovního postupu. Výstupy dílčího pracovního postupu se stanou výstupy nadřazeného pracovního postupu, které obcházejí vnitřní logiku směrování prováděcího nástroje nadřazeného pracovního postupu.

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

Přechodné emise z podřízených pracovních postupů

"intermediate" události vytvořené uvnitř podřízeného pracovního postupu se automaticky propagují do proudu událostí nadřazeného pracovního postupu. Připisují se vlastnímu WorkflowExecutor prvku id (nikoli vnitřnímu exekutoru, který je původně emitoval), což zachovává zapouzdření. Důležité je, že tyto události si zachovávají "intermediate" označení bez ohledu na to, jak nadřazená položka označí WorkflowExecutor ve svých seznamech output_from a intermediate_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}")

Žádosti a odpovědi

Dílčí pracovní postupy plně podporují mechanismus požadavků a odpovědí . Když prováděč uvnitř dílčího pracovního postupu zavolá ctx.request_info(), WorkflowExecutor zachytí požadavek a zpracuje ho na základě propagate_request nastavení.

Zachycení požadavků v nadřazeném pracovním postupu (výchozí)

S propagate_request=False (výchozím nastavením) se požadavky z dílčího pracovního postupu zabalí do SubWorkflowRequestMessage a odešlou se připojeným exekutorům v nadřazeném pracovním postupu. To umožňuje nadřazené exekutory zpracovávat požadavek místně:

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)

Metoda create_response() ověří, že datový typ odpovědi odpovídá očekávanému typu z původního požadavku. Pokud se typy neshodují, je TypeError vyvolán.

Šíření požadavků na externí volající

Požadavky z dílčího pracovního postupu se pomocí standardního propagate_request=True mechanismu přenášejí do streamu událostí nadřazeného pracovního postupu. Volající nadřazeného pracovního procesu zpracovává tyto požadavky stejným způsobem jako jakýkoli jiný požadavek zahrnující člověka.

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)

Jak to funguje

Když nadřazený pracovní postup směruje zprávu do WorkflowExecutor:

1. Doručení vstupu – zpráva se přesměruje do spouštěče vnitřního pracovního postupu. Typ zprávy se musí shodovat s očekávanými vstupními typy spouštěcího exekutoru.
2. Vnitřní vykonávání – vnitřní pracovní postup běží ve vlastní smyčce superkroků až do dokončení, nebo dokud nepotřebuje externí vstup.
3. Výstupní kolekce – výstupní události vnitřního pracovního postupu se shromažďují a přeposílají na allow_direct_output základě nastavení.
4. Předávání žádostí – pokud má vnitřní pracovní postup čekající požadavky, přeposílají se na propagate_request základě nastavení (viz Požadavky a odpovědi).
5. Akumulace odpovědí – WorkflowExecutor shromažďuje odpovědi a pokračuje v dílčím pracovním postupu pouze v případě, že byly přijaty všechny očekávané odpovědi pro dané spuštění.
6. Následující odeslání – výstupy se odesílají do dalšího vykonavatele v nadřazeném pracovním postupu.

Dílčí pracovní postup udržuje svůj vlastní interní stav nezávisle na nadřazeném. Zprávy se směrují pouze přes hrany, které spojují WorkflowExecutor se zbytkem nadřazeného grafu – mezi úrovněmi vnoření nedochází k žádnému vysílání zpráv.

Víceúrovňové vnoření

Dílčí pracovní postupy lze vnořit do libovolné hloubky. Každá úroveň udržuje svůj vlastní kontext spuštění:

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

Zpracování chyb

Pokud dílčí pracovní postup selže, chyba se rozšíří do nadřazeného pracovního postupu. Zachytí WorkflowExecutor neúspěšnou událost z dílčího pracovního postupu a převede ji na chybovou událost v nadřazeném kontextu:

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)

Pokud dílčí pracovní postup narazí na neošetřenou výjimku, nadřazený pracovní postup obdrží chybovou událost s podrobnostmi o výjimce, včetně ID dílčího pracovního postupu.

Vytváření kontrolních bodů

Dílčí pracovní postupy podporují vytváření kontrolních bodů. Při přijetí kontrolního bodu v nadřazeném pracovním postupu WorkflowExecutor serializuje jeho vnitřní stav, včetně průběhu provádění vnitřního pracovního postupu a všech zpráv uložených v mezipaměti. Při obnovení je tento stav deserializován, což umožňuje hlavnímu pracovnímu postupu pokračovat s podřízeným pracovním postupem beze změn.

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)

Vytvoření Sub-Workflow

V Go vytvoříte dílčí workflow sestavením *workflow.Workflow a jeho navázáním na nadřazený workflow pomocí 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
}

Vázaný podřízený pracovní postup funguje z pohledu nadřazeného pracovního postupu jako jeden vykonavatel. Zprávy vstupují přes vazbu, podřízený pracovní postup běží ve vlastním interním grafu a výstupy podřízeného pracovního postupu jsou směrovány zpět do nadřazeného grafu.

Vstupní a výstupní typy

Vazba dědí protokol zabaleného pracovního postupu. Nadřazený pracovní postup může odesílat zprávy, jejichž typy za běhu odpovídají vstupním typům, které podřízený pracovní postup přijímá, a vazba zpřístupňuje typy výstupů poskytovaných podřízeným pracovním postupem jak jako typy zpráv, tak jako výstupní typy.

To znamená, že hrany nadřazeného pracovního postupu musí propojovat executory, jejichž výstupní typy odpovídají vstupům, které dílčí pracovní postup přijímá, a navazující executory musí zpracovávat typy, které dílčí pracovní postup generuje:

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

Chování systému vůči výstupu

Když podřízený pracovní postup vygeneruje výstup, vazba podřízeného pracovního postupu odešle tento výstup jako zprávu z této vazby připojeným vykonavatelům v nadřazeném pracovním postupu. Pokud nadřazený pracovní postup také označí vazbu dílčího pracovního postupu pomocí WithOutputFrom, je vygenerována tatáž hodnota jako nadřazený workflow.OutputEvent, jehož ExecutorID je ID vazby dílčího pracovního postupu.

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

Vlastní události pracovního postupu generované uvnitř podřízeného pracovního postupu se předávají do nadřazeného datového proudu událostí. Události spuštění a životního cyklu samotného dílčího pracovního postupu i superkroku se uchovávají interně, takže nadřazený proud zůstává zaměřený na události, které mají význam navenek.

Žádosti a odpovědi

Dílčí pracovní postupy podporují mechanismus žádosti a odpovědi . Když vykonavatel uvnitř dílčího pracovního postupu odešle externí požadavek, vazba dílčího pracovního postupu upřesní ID portu požadavku tím, že před něj přidá ID vazby. Například port požadavku pojmenovaný ApprovalPort v podřízeném pracovním postupu se v nadřazeném pracovním postupu změní na ApprovalSubWorkflow.ApprovalPort.

Chcete-li zpřístupnit podřízený požadavek prostřednictvím nadřazeného pracovního postupu, přidejte nadřazený port RequestPort s kvalifikovaným ID a směrujte požadavky a odpovědi mezi vazbou podřízeného pracovního postupu a tímto portem:

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

Volající proces zpracuje požadavek z proudu nadřazeného workflow a odešle odpověď zpět prostřednictvím stejného identifikátoru běhu. Vazba dílčího pracovního postupu před doručením odpovědi podřízeného pracovního postupu odebere kvalifikovanou předponu.

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

Pomocí predikátů udržujte hrany požadavků a odpovědí úzké:

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

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

Jak to funguje

Když nadřazený pracovní postup směruje zprávu do vazby dílčího pracovního postupu:

1. Vstupní doručení – vazba přijímá zprávy, které odpovídají přijatým vstupním typům podřízeného pracovního postupu, a zarovná je do spouštěcího exekutoru podřízeného pracovního postupu.
2. Vnitřní provádění – podřízený pracovní postup běží ve stejném prostředí provádění v rámci procesu a udržuje vlastní smyčku superkroků.
3. Přeposílání výstupu — hodnoty potomka workflow.OutputEvent se odesílají jako zprávy z vazby následným nadřazeným exekutorům a jsou také vraceny nadřazeným exekutorem, pokud je vazba uvedena v WithOutputFrom.
4. Přeposílání požadavků — požadavky potomků workflow.RequestInfoEvent jsou znovu odesílány s úplnými identifikátory portů a lze je směrovat prostřednictvím vazeb nadřazeného prvku RequestPort.
5. Předávání událostí – do nadřazeného datového proudu se přidají vlastní podřízené události pracovního postupu. Chyby jsou vykazovány jako nadřazené hodnoty workflow.ErrorEvent s zaznamenaným ID dílčího pracovního postupu.
6. Podřízené odesílání – výsledné zprávy pokračují přes nadřazené hrany pracovního postupu.

Podřízený pracovní postup uchovává svůj stav a směrování zpráv odděleně od nadřazeného pracovního postupu. Zprávy procházejí přes hranici pouze prostřednictvím hran připojených k vazbě podřízeného pracovního postupu.

Víceúrovňové vnoření

Dílčí pracovní postupy lze vnořit do libovolné hloubky. Každý podřízený pracovní postup je svázán před jeho přidáním do pracovního postupu, který ho obsahuje:

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

Zpracování chyb

Když podřízený pracovní postup vyvolá chybu, vazba podřízeného pracovního postupu ji předá nadřazenému pracovnímu postupu jako workflow.ErrorEvent a nastaví SubWorkflowID na ID vazby. Nadřazený pracovní postup může sledovat tyto chyby prostřednictvím stejného datového proudu událostí, který používá pro chyby pracovního postupu nejvyšší úrovně:

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

Chyby vzniklé při přeposílání podřízených událostí se také převádějí na nadřazené hodnoty workflow.ErrorEvent s připojeným ID dílčího pracovního postupu.

Vytváření kontrolních bodů

Dílčí pracovní postupy podporují vytváření kontrolních bodů. Když nadřazený pracovní postup vytvoří kontrolní bod, vazba podřízeného pracovního postupu uloží správce kontrolních bodů podřízeného pracovního postupu a všechna čekající kvalifikovaná mapování portů pro odpovědi do stavu nadřazeného executoru. Při obnovení může podřízený pracovní postup pokračovat se stavem vnořeného spuštění beze změny, včetně čekajících požadavků.

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)

Pokud je kontrolní bod obnoven v okamžiku, kdy má podřízený pracovní postup nevyřízený požadavek, obnovené spuštění nadřazeného pracovního postupu znovu publikuje událost s informacemi o kvalifikovaném požadavku. Volající může na základě tohoto znovu publikovaného požadavku vytvořit odpověď a odeslat ji zpět přes handle nadřazeného běhu.