Subfluxos de Trabalho

Criar um Subfluxo

Em C#, compões sub-fluxos de trabalho de duas formas:

• Ligação direta — usar BindAsExecutor() para incorporar um fluxo de trabalho diretamente como executor no fluxo de trabalho pai. Isto preserva os tipos nativos de entrada/saída do subfluxo de trabalho.
• Envolvimento de agentes — usar AsAIAgent() para converter um fluxo de trabalho num agente, depois adicionar o agente ao fluxo de trabalho principal. Isto é útil quando o fluxo de trabalho pai utiliza executores baseados em agentes.

Vinculação Direta Utilizando BindAsExecutor

O BindAsExecutor() método de extensão converte um fluxo de trabalho num ExecutorBinding que pode ser adicionado diretamente a um fluxo de trabalho pai:

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 com ele, os tipos de entrada e saída do sub-workflow são preservados — o workflow pai encaminha mensagens com base nos tipos reais que o sub-workflow espera e produz.

Encapsulamento do Agente com o AsAIAgent

Quando o fluxo de trabalho pai utiliza executores baseados em agentes, converta o fluxo interno para um agente usando AsAIAgent(). O WorkflowBuilder automaticamente envolve o agente num executor:

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

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

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

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

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

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

O fluxo de trabalho interno funciona como uma única etapa do ponto de vista do fluxo de trabalho pai. O coordenador envia mensagens para o pipeline de análise, que executa specialist1 → specialist2 internamente, e depois encaminha o resultado para o revisor.

Tipos de Entrada e Saída

Quando um fluxo de trabalho é usado como subfluxo de trabalho, preserva os contratos de tipo dos seus executores internos.

Com BindAsExecutor, o executor do subfluxo de trabalho aceita os mesmos tipos de entrada que o executor inicial do fluxo de trabalho interno e envia os mesmos tipos de saída que o fluxo de trabalho interno produz. As arestas do fluxo de trabalho pai devem ligar executores cujos tipos de saída correspondam aos tipos de entrada esperados do sub-fluxo de trabalho, e os tipos de saída do sub-fluxo de trabalho devem corresponder aos tipos de entrada esperados pelos executores a jusante.

Com AsAIAgent, o sub-fluxo de trabalho é encapsulado como um agente e segue os contratos de entrada e saída do Executor do Agente (string, ChatMessage, IEnumerable<ChatMessage>).

Comportamento de saída

Por defeito, quando um subfluxo de trabalho produz saídas (via YieldOutputAsync), essas saídas são transmitidas como mensagens para executores ligados no fluxo de trabalho pai. Isto permite que executores posteriores processem resultados de subfluxos de trabalho.

A ExecutorOptions classe controla este comportamento:

Quando AutoYieldOutputHandlerResultObject está ativado, as saídas do subfluxo bypassam o encaminhamento interno do fluxo de trabalho principal e são entregues diretamente ao chamador desse fluxo de trabalho.

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

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

Pedidos e Respostas

Os subfluxos de trabalho suportam totalmente o mecanismo de pedido e resposta . Quando um executor dentro do sub-workflow envia um pedido (por exemplo, para pedir input humano), o WorkflowHostExecutor encaminha RequestInfoEvent ao workflow pai com um ID de porta qualificado — o ID do executor do sub-workflow é colocado antes do ID da porta (por exemplo, SubWorkflow.GuessNumber).

Esta qualificação garante que, quando o fluxo de trabalho pai recebe uma resposta, pode encaminhar a resposta de volta para a instância correta do sub-workflow. O fluxo de trabalho pai gere subfluxos de trabalho usando o mesmo mecanismo de resposta de qualquer outro pedido:

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

Como funciona

Quando o fluxo de trabalho pai encaminha uma mensagem para o executor do sub-workflow:

1. Entrega de entrada — a mensagem é encaminhada para o executor de início do fluxo de trabalho interno. Com BindAsExecutor, o tipo de mensagem deve corresponder aos tipos esperados pelo executor inicial. Com AsAIAgent, as mensagens são normalizadas para o formato ChatMessage.
2. Execução interna — o fluxo de trabalho interno executa o seu próprio ciclo superstep.
3. Recolha de saída — os eventos de saída do fluxo de trabalho interno são recolhidos. Com BindAsExecutor, as saídas mantêm os seus tipos originais. Com AsAIAgent, as saídas são convertidas em mensagens de resposta do agente.
4. Encaminhamento de pedidos — se o fluxo de trabalho interno tiver pedidos pendentes, estes são encaminhados para o fluxo de trabalho pai para tratamento (ver Pedidos e Respostas).
5. Distribuição descendente — as mensagens resultantes são enviadas para o próximo executante no fluxo de trabalho pai.

Como o fluxo de trabalho interno mantém o seu próprio contexto de execução, o seu estado é independente do fluxo de trabalho pai.

Aninhamento Multi-Nível

Os sub-fluxos de trabalho podem ser aninhados a uma profundidade arbitrária. Cada nível mantém o seu próprio contexto de execução:

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

Tratamento de erros

Quando um subfluxo de trabalho falha, o erro é propagado para o fluxo de trabalho pai como um SubworkflowErrorEvent. O fluxo de trabalho pai pode observar estes erros através do seu fluxo de eventos:

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

Se o sub-workflow encontrar uma exceção não tratada, a execução do workflow pai continua, mas o executor do sub-workflow para de processar mensagens adicionais.

Pontos de verificação

Quando um checkpoint é feito no fluxo de trabalho pai, o estado da sessão do sub-agente de workflow é serializado como parte dos dados do checkpoint do executor pai. Durante a restauração, o estado da sessão é desserializado, de forma a permitir que o fluxo de trabalho principal retome com o estado do subfluxo de trabalho intacto.

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

Criar um Subfluxo

Em Python, cria-se um sub-fluxo de trabalho ao envolver um Workflow em um WorkflowExecutor e adicioná-lo a um fluxo de trabalho pai.

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

O fluxo de trabalho interno funciona como uma única etapa do ponto de vista do fluxo de trabalho pai. O coordenador envia mensagens para o pipeline de análise, que executa specialist1 → specialist2 internamente, e depois encaminha o resultado para o revisor.

Parâmetros do WorkflowExecutor

Encapsular subfluxos de trabalho

Envolva explicitamente as instâncias Workflow num WorkflowExecutor antes de as adicionar a um fluxo de trabalho principal. Os agentes podem ser passados diretamente para WorkflowBuilder, mas as instâncias brutas Workflow requerem este wrapper.

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

O envolvimento explícito permite-lhe:

• Atribui um ID de executor específico para referência em múltiplas arestas.
• Reutilize a mesma WorkflowExecutor instância ao longo do gráfico.

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

Tipos de Entrada e Saída

O WorkflowExecutor herda a sua assinatura de tipo do workflow encapsulado:

• Os tipos de entrada correspondem aos tipos de entrada do executor inicial do fluxo de trabalho embalado (mais SubWorkflowResponseMessage para tratar respostas a pedidos encaminhados).
• Os tipos de saída correspondem aos tipos de saída do fluxo de trabalho encapsulado. Se algum executor no sub-workflow for capaz de solicitar-responder, SubWorkflowRequestMessage também é incluído como tipo de saída.

Isto significa que as arestas do fluxo de trabalho pai devem ligar executores cujos tipos de saída correspondem aos tipos de entrada esperados do subfluxo de trabalho. De forma semelhante, os executores a jusante devem aceitar os tipos que o subfluxo de trabalho produz:

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

Comportamento de saída

Por defeito (allow_direct_output=False), quando um sub-fluxo de trabalho produz saídas via yield_output, essas saídas são encaminhadas como mensagens para executores ligados no fluxo de trabalho pai usando send_message. Isto permite que executores posteriores processem resultados de subworkflows como parte do grafo pai.

Quando allow_direct_output=True, as saídas do subfluxo de trabalho são enviadas diretamente para o fluxo de eventos do fluxo de trabalho pai. As saídas do sub-fluxo de trabalho passam a ser saídas do workflow pai, contornando o encaminhamento interno do executor parental:

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

Emissões intermédias dos fluxos de trabalho infantis

"intermediate" Os eventos produzidos num fluxo de trabalho filho propagam-se automaticamente para o fluxo de eventos do fluxo de trabalho pai. São atribuídos ao próprio WorkflowExecutor de id (e não ao executor interno que os emitiu originalmente), o que preserva o encapsulamento. É crucial que estes eventos mantenham a etiqueta "intermediate", independentemente de como o pai designa o WorkflowExecutor nas suas próprias listas output_from ou 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}")

Pedidos e Respostas

Os subfluxos de trabalho suportam totalmente o mecanismo de pedido e resposta . Quando um executor dentro de um subfluxo de trabalho chama ctx.request_info(), WorkflowExecutor interceta o pedido e trata com base na configuração propagate_request.

Interceção de Requisições no Fluxo de Trabalho Pai (Padrão)

Com propagate_request=False (o predefinido), os pedidos do sub-workflow são encapsulados num SubWorkflowRequestMessage e enviados para executores ligados no workflow pai. Isto permite que os executores pais tratem do pedido localmente:

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)

O create_response() método valida que o tipo de dados de resposta corresponde ao tipo esperado do pedido original. Se os tipos não coincidirem, TypeError é lançado.

Propagação de Solicitações para Destinatários Externos

Com propagate_request=True, os pedidos do subfluxo de trabalho são propagados para o fluxo de eventos do fluxo de trabalho pai usando o mecanismo padrão request_info . O chamador do fluxo de trabalho pai trata estas solicitações da mesma forma que qualquer outra solicitação de intervenção humana:

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)

Como funciona

Quando o fluxo de trabalho pai encaminha uma mensagem para o WorkflowExecutor:

1. Entrega de entrada — a mensagem é encaminhada para o executor de início do fluxo de trabalho interno. O tipo de mensagem deve corresponder aos tipos de entrada esperados pelo executor inicial.
2. Execução interna — o fluxo de trabalho interno executa o seu próprio ciclo superstep até à sua conclusão, ou até precisar de input externo.
3. Recolha de saída — os eventos de saída do fluxo de trabalho interno são recolhidos e encaminhados com base na allow_direct_output definição.
4. Encaminhamento de pedidos — se o fluxo de trabalho interno tiver pedidos pendentes, estes são encaminhados com base na propagate_request definição (ver Pedidos e Respostas).
5. Acumulação de respostas — recolhe WorkflowExecutor respostas e retoma o subfluxo de trabalho apenas quando todas as respostas esperadas para uma determinada execução já foram recebidas.
6. Despacho downstream — as saídas são enviadas para o próximo executor no fluxo de trabalho pai.

O sub-workflow mantém o seu próprio estado interno de forma independente do fluxo de trabalho principal. As mensagens são encaminhadas apenas através das arestas que ligam o WorkflowExecutor ao resto do grafo pai — não há nenhuma mensagem a ser transmitida entre os níveis de aninhamento.

Aninhamento Multi-Nível

Os sub-fluxos de trabalho podem ser aninhados a uma profundidade arbitrária. Cada nível mantém o seu próprio contexto de execução:

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

Tratamento de erros

Quando um sub-fluxo de trabalho falha, o erro é propagado para o fluxo de trabalho pai. O WorkflowExecutor captura o evento falhado do subfluxo de trabalho e converte-o num evento de erro no contexto pai.

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)

Se o sub-workflow encontrar uma exceção não tratada, o workflow pai recebe um evento de erro com os detalhes das exceções, incluindo o ID do sub-workflow.

Pontos de verificação

Os subfluxos de trabalho suportam pontos de verificação. Quando é feito um checkpoint no fluxo de trabalho pai, o WorkflowExecutor serializa o seu estado interno, incluindo o progresso de execução do fluxo interno e quaisquer mensagens em cache. Na restauração, este estado é deserializado, permitindo que o fluxo de trabalho principal retome com o sub-fluxo de trabalho intacto.

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)

Criar um Subfluxo

No Go, cria-se um sub-fluxo de trabalho construindo um *workflow.Workflow e associando-o ao fluxo de trabalho pai com 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
}

O fluxo de trabalho filho vinculado corre como um executor único do ponto de vista do fluxo de trabalho pai. As mensagens entram através da ligação, o subfluxo de trabalho executa o seu próprio grafo interno e as suas saídas são redirecionadas de volta para o grafo principal.

Tipos de Entrada e Saída

A ligação herda o protocolo do fluxo de trabalho encapsulado. O fluxo de trabalho pai pode enviar mensagens cujos tipos de execução correspondem aos tipos de entrada aceites pelo fluxo de trabalho filho, e a ligação expõe os tipos de saída obtidos pelo fluxo de trabalho filho como tipos de mensagem e de saída.

Isto significa que as arestas do workflow pai devem ligar executores cujos tipos de saída correspondam às entradas aceites pelo sub-workflow, e os executores a jusante devem processar os tipos produzidos pelo workflow filho:

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

Comportamento de saída

Quando um fluxo de trabalho filho gera uma saída, a associação do subfluxo envia essa saída como uma mensagem da associação para os executores conectados no fluxo de trabalho principal. Se o workflow principal também marcar a ligação do sub-workflow com WithOutputFrom, o mesmo valor é emitido como um elemento workflow.OutputEvent principal, cujo ExecutorID é o ID da ligação do sub-workflow.

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

Os eventos personalizados do fluxo de trabalho emitidos dentro do fluxo de trabalho filho são reencaminhados para o fluxo de eventos do fluxo de trabalho principal. Os próprios eventos do ciclo de vida inicial e superpasso do sub-workflow mantêm-se internos para que o fluxo principal se mantenha focado em eventos externamente significativos.

Pedidos e Respostas

Os subfluxos de trabalho suportam o mecanismo de pedido e resposta . Quando um executor dentro do subfluxo de trabalho emite um pedido externo, a ligação do subfluxo de trabalho qualifica o ID da porta de pedido antepondo o ID da ligação. Por exemplo, uma porta de pedido subordinado com o nome ApprovalPort passa a ser ApprovalSubWorkflow.ApprovalPort no fluxo de trabalho principal.

Para expor o pedido subordinado através do fluxo de trabalho principal, adicione uma porta principal RequestPort com o ID qualificado e encaminhe pedidos e respostas entre a vinculação do subfluxo de trabalho e essa porta:

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

O chamador trata do pedido do fluxo de trabalho pai e envia a resposta de volta através do mesmo handle de execução. A ligação do subfluxo de trabalho remove o prefixo qualificado antes de entregar a resposta ao fluxo de trabalho filho.

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

Use predicados para manter limitadas as extremidades do pedido e da resposta:

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

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

Como funciona

Quando o fluxo de trabalho pai encaminha uma mensagem para a ligação do subfluxo de trabalho:

1. Entrega de entrada — a ligação aceita mensagens que correspondem aos tipos de entrada aceites pelo fluxo de trabalho filho e coloca-as em fila no executor inicial do fluxo de trabalho filho.
2. Execução interna — o fluxo de trabalho filho é executado no mesmo ambiente de execução no processo e mantém o seu próprio ciclo de superetapas.
3. Reencaminhamento de saída — os valores subordinados workflow.OutputEvent são enviados como mensagens da vinculação para os executores do elemento superior a jusante, e também são produzidos pelo elemento superior se a vinculação estiver listada em WithOutputFrom.
4. Reencaminhamento de pedidos — os pedidos subordinados workflow.RequestInfoEvent são reemitidos com IDs de porta qualificados e podem ser encaminhados pelas ligações do elemento pai RequestPort.
5. Encaminhamento de eventos — os eventos personalizados do fluxo de trabalho subordinado são adicionados ao fluxo principal. Os erros são apresentados como valores principais workflow.ErrorEvent, com o ID do subfluxo de trabalho registado.
6. Encaminhamento a jusante — as mensagens resultantes seguem pelas arestas do fluxo de trabalho principal.

O fluxo de trabalho filho mantém o seu estado e o encaminhamento das mensagens separados dos do fluxo de trabalho pai. As mensagens atravessam a fronteira apenas através das arestas ligadas à ligação do subfluxo de trabalho.

Aninhamento Multi-Nível

Os sub-fluxos de trabalho podem ser aninhados a uma profundidade arbitrária. Cada subfluxo de trabalho é associado antes de ser adicionado ao fluxo de trabalho que o contém:

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

Tratamento de erros

Quando um fluxo de trabalho filho emite um erro, a associação do subfluxo de trabalho encaminha-o para o fluxo de trabalho principal como workflow.ErrorEvent e define SubWorkflowID com o ID da associação. O fluxo de trabalho pai pode observar estes erros através do mesmo fluxo de eventos que utiliza para erros de fluxo de trabalho de topo:

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

Os erros gerados durante o encaminhamento de eventos subordinados também são convertidos em valores do fluxo de trabalho principal workflow.ErrorEvent, com o ID do subfluxo de trabalho anexado.

Pontos de verificação

Os subfluxos de trabalho suportam pontos de verificação. Quando o fluxo de trabalho principal cria um ponto de verificação, a associação do subfluxo de trabalho armazena o gestor de pontos de verificação do fluxo de trabalho subordinado e quaisquer mapeamentos pendentes de portas de resposta qualificadas no estado do executor principal. Na restauração, o fluxo de trabalho filho pode retomar com o seu estado de execução aninhado intacto, incluindo pedidos pendentes.

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)

Se um checkpoint for restaurado enquanto um fluxo de trabalho filho tiver um pedido pendente, a execução principal restaurada republica o evento de informação do pedido qualificado. O chamador pode criar uma resposta a partir desse pedido republicado e enviá-la de volta através do identificador da execução principal.