Subfluxos de Trabalho

Criando um Sub-Workflow

No C#, você redigi os sub-fluxos de trabalho de duas maneiras:

• Associação direta – use BindAsExecutor() para inserir um fluxo de trabalho diretamente como executor no fluxo de trabalho pai. Isso preserva os tipos nativos de entrada/saída do sub-fluxo de trabalho.
• Encapsulamento do agente – use AsAIAgent() para converter um fluxo de trabalho em um agente e, em seguida, adicione o agente ao fluxo de trabalho pai. Isso é útil quando o fluxo de trabalho pai usa executores baseados em agente.

Associação direta com BindAsExecutor

O BindAsExecutor() método de extensão converte um fluxo de trabalho em um 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();

Com BindAsExecutoros tipos de entrada e saída tipados do sub-fluxo de trabalho são preservados, o fluxo de trabalho pai roteia mensagens com base nos tipos reais que o sub-fluxo de trabalho espera e produz.

Encapsulamento de agente com AsAIAgent

Quando o fluxo de trabalho pai usa executores baseados em agente, converta o fluxo de trabalho interno em um agente usando AsAIAgent(). O WorkflowBuilder encapsula automaticamente o agente em um 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 opera como uma etapa única 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, em seguida, encaminha o resultado para o revisor.

Tipos de entrada e saída

Quando um fluxo de trabalho é usado como um sub-fluxo de trabalho, ele preserva os contratos de tipo de seus executores internos.

Com BindAsExecutoro executor de sub-fluxo 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 bordas do fluxo de trabalho pai devem conectar executores cujos tipos de saída correspondem aos tipos de entrada esperados do subfluxo de trabalho, e os tipos de saída do subfluxo de trabalho devem corresponder às entradas esperadas dos executores posteriores.

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

Comportamento de saída

Por padrão, quando um sub-fluxo de trabalho produz saídas (via YieldOutputAsync), essas saídas são encaminhadas como mensagens para executores conectados no fluxo de trabalho pai. Isso permite que executores downstream processem resultados de sub-fluxo de trabalho.

A ExecutorOptions classe controla esse comportamento:

Quando AutoYieldOutputHandlerResultObject estiver habilitado, as saídas de sub-fluxo de trabalho ignoram o roteamento interno do pai e são entregues diretamente ao chamador do fluxo de trabalho pai.

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

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

Solicitações e respostas

Os sub-fluxos de trabalho dão suporte total ao mecanismo de solicitação e resposta . Quando um executor dentro do subfluxo de trabalho envia uma solicitação (por exemplo, para solicitar entrada humana), o WorkflowHostExecutor encaminha ao RequestInfoEvent workflow principal com uma ID de porta qualificada — o ID do executor do subfluxo de trabalho é anexado ao ID da porta (por exemplo, SubWorkflow.GuessNumber).

Essa qualificação garante que, quando o fluxo de trabalho pai receber uma resposta, ele possa rotear a resposta de volta para a instância de subfluxo de trabalho correta. O fluxo de trabalho pai lida com solicitações de sub-fluxo de trabalho usando o mesmo mecanismo de resposta que qualquer outra solicitação:

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 roteia uma mensagem para o executor de subfluxo:

1. Entrega de entrada – a mensagem é encaminhada para o executor inicial do fluxo de trabalho interno. Com BindAsExecutor, o tipo de mensagem deve corresponder aos tipos esperados do executor inicial. Com AsAIAgent, as mensagens são normalizadas para ChatMessage formato.
2. Execução interna – o fluxo de trabalho interno executa seu próprio loop superstep.
3. Coleção de saída – os eventos de saída do fluxo de trabalho interno são coletados. Com BindAsExecutor, as saídas mantêm seus tipos originais. Com AsAIAgent, as saídas são convertidas em mensagens de resposta do agente.
4. Encaminhamento de solicitação – se o fluxo de trabalho interno tiver solicitações pendentes, elas serão encaminhadas para o fluxo de trabalho pai para tratamento (consulte Solicitações e Respostas).
5. Envio downstream – as mensagens resultantes são enviadas no fluxo de trabalho pai para o próximo executor.

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

Aninhamento de vários níveis

Os sub-fluxos de trabalho podem ser aninhados em profundidade arbitrária. Cada nível mantém 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 sub-fluxo de trabalho falha, o erro é propagado para o fluxo de trabalho pai como um SubworkflowErrorEvent. O fluxo de trabalho pai pode observar esses erros por meio de 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 subfluxo de trabalho encontrar uma exceção sem tratamento, a execução do fluxo de trabalho pai continuará, mas o executor do subfluxo de trabalho parará de processar novas mensagens.

Definindo o ponto de verificação

Quando um ponto de verificação é feito no fluxo de trabalho pai, o estado da sessão do agente do subfluxo de trabalho é serializado como parte dos dados de ponto de verificação do executor pai. Na restauração, o estado da sessão é desserializado, permitindo que o fluxo de trabalho pai seja retomado 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);

Criando um Sub-Workflow

No Python, você cria um sub-fluxo de trabalho encapsulando um Workflow em um WorkflowExecutor e adicionando-o 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 opera como uma etapa única 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, em seguida, encaminha o resultado para o revisor.

Os parâmetros do WorkflowExecutor

Encapsulando sub-fluxos de trabalho

Encapsule explicitamente as instâncias Workflow em um WorkflowExecutor antes de adicioná-las a um fluxo de trabalho pai. Os agentes podem ser passados diretamente para WorkflowBuilder, mas as instâncias brutas Workflow exigem esse 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 encapsulamento explícito permite que você:

• Atribua uma ID de executor específica para referência em várias bordas.
• Reutilize a mesma WorkflowExecutor instância no grafo.

# 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 sua assinatura de tipo do fluxo de trabalho encapsulado:

• Os tipos de entrada correspondem aos tipos de entrada do executor inicial do fluxo de trabalho encapsulado (além de SubWorkflowResponseMessage lidar com respostas a solicitações encaminhadas).
• Os tipos de saída correspondem aos tipos de saída do fluxo de trabalho encapsulado. Se qualquer executor no sub-fluxo de trabalho for capaz de solicitação-resposta, SubWorkflowRequestMessage também será incluído como um tipo de saída.

Isso significa que as bordas do fluxo de trabalho pai devem conectar executores cujos tipos de saída correspondem aos tipos de entrada esperados do sub-fluxo de trabalho. Da mesma forma, os executores downstream devem aceitar os tipos que o sub-fluxo 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 padrão (allow_direct_output=False), quando um sub-fluxo de trabalho produz saídas por meio yield_output, essas saídas são encaminhadas como mensagens para executores conectados no fluxo de trabalho pai usando send_message. Isso permite que executores a jusante processem resultados de subfluxo de trabalho como parte do grafo pai.

Quando allow_direct_output=True as saídas de subfluxo de trabalho são enviadas diretamente ao fluxo de eventos do fluxo de trabalho pai. As saídas do sub-fluxo de trabalho tornam-se saídas do fluxo de trabalho pai, ignorando o roteamento interno do executor pai:

# 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 intermediárias de fluxos de trabalho filho

"intermediate" eventos produzidos dentro de um fluxo de trabalho filho se propagam automaticamente pelo fluxo de eventos do fluxo de trabalho pai. Eles são atribuídos ao próprio WorkflowExecutorid (e não ao executor interno que os emitiu originalmente), o que preserva o encapsulamento. Fundamentalmente, esses eventos mantêm o "intermediate" rótulo independentemente de como o pai designa o WorkflowExecutor em 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}")

Solicitações e respostas

Os sub-fluxos de trabalho dão suporte total ao mecanismo de solicitação e resposta . Quando um executor dentro de um sub-fluxo de trabalho chama ctx.request_info(), o WorkflowExecutor intercepta a solicitação e a manipula com base na configuração de propagate_request.

Intercepção de solicitações no fluxo de trabalho pai (padrão)

Com propagate_request=False (o padrão), as solicitações do sub-fluxo de trabalho são encapsuladas em um SubWorkflowRequestMessage e enviadas para executores conectados no fluxo de trabalho pai. Isso permite que os executores pai lidem com a solicitação 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 da solicitação original. Se os tipos não corresponderem, um TypeError será gerado.

Propagação de solicitações para chamadores externos

Com propagate_request=True, as solicitações do sub-fluxo de trabalho são propagadas para o fluxo de eventos do fluxo de trabalho pai usando o mecanismo padrão request_info . O chamador do fluxo de trabalho pai lida com essas solicitações da mesma maneira que qualquer outra solicitação humana no loop:

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 roteia uma mensagem para:WorkflowExecutor

1. Entrega de entrada – a mensagem é encaminhada para o executor inicial do fluxo de trabalho interno. O tipo de mensagem deve corresponder aos tipos de entrada esperados do executor inicial.
2. Execução interna – o fluxo de trabalho interno executa seu próprio loop de superstep até a conclusão, ou até que precise de entrada externa.
3. Coleção de saída – os eventos de saída do fluxo de trabalho interno são coletados e encaminhados com base na allow_direct_output configuração.
4. Encaminhamento de solicitação – se o fluxo de trabalho interno tiver solicitações pendentes, elas serão encaminhadas com base na propagate_request configuração (consulte Solicitações e Respostas).
5. Acúmulo de resposta – o WorkflowExecutor coleta respostas e retoma o sub-fluxo de trabalho somente quando todas as respostas esperadas para uma determinada execução foram recebidas.
6. Expedição downstream – as saídas são enviadas para o próximo executor no fluxo de trabalho pai.

O subfluxo de trabalho mantém, independentemente do pai, seu próprio estado interno. As mensagens são roteadas apenas pelas bordas que conectam o WorkflowExecutor ao restante do grafo pai — não há transmissão de mensagens entre os níveis de aninhamento.

Aninhamento de vários níveis

Os sub-fluxos de trabalho podem ser aninhados em profundidade arbitrária. Cada nível mantém 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 com falha do sub-fluxo de trabalho e o converte em um 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-fluxo de trabalho encontrar uma exceção não tratada, o fluxo de trabalho pai receberá um evento de erro contendo os detalhes da exceção, incluindo a ID do sub-fluxo de trabalho.

Definindo o ponto de verificação

Os sub-fluxos de trabalho dão suporte ao ponto de verificação. Quando um ponto de verificação é feito no fluxo de trabalho pai, o WorkflowExecutor serializa seu estado interno, incluindo o progresso de execução do fluxo de trabalho interno e todas as mensagens armazenadas em cache. Na restauração, esse estado é desserializado, permitindo que o fluxo de trabalho pai seja retomado com o subfluxo 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)

Criando um Sub-Workflow

No Go, você cria um sub-fluxo de trabalho criando 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 é executado como um único executor do ponto de vista do fluxo de trabalho pai. As mensagens entram por meio da vinculação, o fluxo de trabalho filho executa seu próprio grafo interno, e as saídas do fluxo de trabalho filho são encaminhadas de volta ao grafo pai.

Tipos de entrada e saída

A associação herda o protocolo do fluxo de trabalho encapsulado. O fluxo de trabalho pai pode enviar mensagens cujos tipos em tempo de execução correspondem aos tipos de entrada aceitos pelo fluxo de trabalho filho, e a vinculação expõe os tipos de saída produzidos pelo fluxo de trabalho filho tanto como tipos de mensagem quanto como tipos de saída.

Isso significa que as bordas do fluxo de trabalho pai devem conectar executores cujos tipos de saída correspondem às entradas aceitas do sub-fluxo de trabalho e os executores downstream devem lidar com os tipos gerados pelo fluxo de trabalho 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, o vínculo do subfluxo de trabalho envia essa saída como uma mensagem do vínculo aos executores conectados no fluxo de trabalho pai. Se o fluxo de trabalho pai também marcar a associação do subfluxo de trabalho com WithOutputFrom, o mesmo valor será emitido como um workflow.OutputEvent pai, cujo ExecutorID é o ID da associação do subfluxo de trabalho.

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

Eventos de fluxo de trabalho personalizados emitidos dentro do fluxo de trabalho filho são encaminhados ao fluxo de eventos do fluxo de trabalho pai. Os eventos do ciclo de vida de início e de superstep do próprio subworkflow são mantidos internamente para que o fluxo pai permaneça focado em eventos significativos externamente.

Solicitações e respostas

Os sub-fluxos de trabalho dão suporte ao mecanismo de solicitação e resposta . Quando um executor dentro do subfluxo de trabalho publica uma solicitação externa, o vínculo do subfluxo de trabalho qualifica o ID da porta de solicitação, prefixando-o com o ID do vínculo. Por exemplo, uma porta de solicitação filha chamada ApprovalPort torna-se ApprovalSubWorkflow.ApprovalPort no fluxo de trabalho pai.

Para exibir a solicitação filho por meio do fluxo de trabalho pai, adicione um pai RequestPort com a ID qualificada e encaminhe solicitações e respostas entre a associação de sub-fluxo 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 manipula a solicitação do fluxo de trabalho pai e envia a resposta de volta pelo mesmo identificador de execução. A vinculação de 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 as bordas de solicitação e resposta estreitas:

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 vinculação do subfluxo de trabalho:

1. Entrega de entradas — a vinculação aceita mensagens que correspondem aos tipos de entrada aceitos pelo fluxo de trabalho filho e as enfileira no executor de inicialização do fluxo de trabalho filho.
2. Execução interna – o fluxo de trabalho filho é executado no mesmo ambiente de execução em processo e mantém seu próprio loop de superstep.
3. Encaminhamento de saída – os valores filho workflow.OutputEvent são enviados como mensagens da associação para executores pai downstream e também são gerados do pai se a associação estiver listada em WithOutputFrom.
4. Encaminhamento de solicitações – as solicitações filho workflow.RequestInfoEvent são remetidas com IDs de porta qualificadas e podem ser roteadas por meio de associações pai RequestPort .
5. Encaminhamento de eventos — eventos personalizados do workflow filho são adicionados ao fluxo pai. Os erros são exibidos como valores pai workflow.ErrorEvent com a ID do sub-fluxo de trabalho registrada.
6. Expedição downstream – as mensagens resultantes continuam por meio das bordas do fluxo de trabalho pai.

O fluxo de trabalho filho mantém seu estado e o roteamento de mensagens separados do fluxo de trabalho pai. As mensagens cruzam o limite apenas pelas extremidades conectadas ao vínculo do subfluxo de trabalho.

Aninhamento de vários níveis

Os sub-fluxos de trabalho podem ser aninhados em profundidade arbitrária. Cada fluxo de trabalho filho é vinculado 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 vinculação de subfluxo de trabalho o encaminha ao fluxo de trabalho pai na forma de workflow.ErrorEvent e define SubWorkflowID como o ID da vinculação. O fluxo de trabalho pai pode observar esses erros por meio do mesmo fluxo de eventos que usa para erros de fluxo de trabalho de nível superior:

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 ao encaminhar eventos secundários também são convertidos em valores do fluxo de trabalho pai, com a ID do subfluxo de trabalho associada ao valor workflow.ErrorEvent.

Definindo o ponto de verificação

Os sub-fluxos de trabalho dão suporte ao ponto de verificação. Quando o fluxo de trabalho pai faz um checkpoint, o vínculo do subfluxo de trabalho armazena o gerenciador de checkpoint do fluxo de trabalho filho e todos os mapeamentos qualificados de portas de resposta pendentes no estado do executor do fluxo de trabalho pai. Na restauração, o fluxo de trabalho filho pode ser retomado com seu estado de execução aninhado intacto, incluindo solicitações 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 ponto de verificação for restaurado enquanto um fluxo de trabalho filho tiver uma solicitação pendente, a execução pai restaurada republica o evento de informações de solicitação qualificada. O chamador pode criar uma resposta a partir dessa solicitação republicada e enviá-la de volta pelo identificador de execução pai.