Subflujos de Trabajo

Creación de un Sub-Workflow

En C#, se componen subprocesos de dos maneras:

• Enlace directo : use BindAsExecutor() para insertar un flujo de trabajo directamente como ejecutor en el flujo de trabajo primario. Esto conserva los tipos nativos de entrada y salida del sub-flujo de trabajo.
• Ajuste del agente : use AsAIAgent() para convertir un flujo de trabajo en un agente y, a continuación, agregue el agente al flujo de trabajo primario. Esto resulta útil cuando el flujo de trabajo primario usa ejecutores basados en agentes.

Vinculación directa con BindAsExecutor

El BindAsExecutor() método de extensión convierte un flujo de trabajo en un ExecutorBinding que se puede agregar directamente a un flujo de trabajo primario:

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

Con BindAsExecutor, se conservan los tipos de entrada y salida tipados del sub-flujo de trabajo: el flujo de trabajo primario enruta los mensajes en función de los tipos reales que espera y genera el subproceso.

Ajuste del agente con AsAIAgent

Cuando el flujo de trabajo primario usa ejecutores basados en agente, convierta el flujo de trabajo interno en un agente mediante AsAIAgent(). WorkflowBuilder envuelve automáticamente al agente en un ejecutor.

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

El flujo de trabajo interno se ejecuta como un solo paso desde la perspectiva del flujo de trabajo primario. El coordinador envía mensajes a la canalización de análisis, que ejecuta specialist1 → specialist2internamente y, a continuación, reenvía el resultado al revisor.

Tipos de entrada y salida

Cuando se utiliza un flujo de trabajo como subflujo, mantiene los contratos de tipo de sus ejecutores internos.

Con BindAsExecutor, el ejecutor del sub-flujo de trabajo acepta los mismos tipos de entrada que el ejecutor de inicio del flujo de trabajo interno y envía los mismos tipos de salida que genera el flujo de trabajo interno. Los bordes del flujo de trabajo primario deben conectar ejecutores cuyos tipos de salida coincidan con los tipos de entrada esperados del sub-flujo de trabajo y los tipos de salida del subproceso deben coincidir con las entradas esperadas de los ejecutores de nivel inferior.

Con AsAIAgent, el subflujo de trabajo se ajusta como agente y sigue los contratos de entrada/salida del Ejecutor del Agente (string, ChatMessage, IEnumerable<ChatMessage>).

Comportamiento de salida

De forma predeterminada, cuando un sub-flujo de trabajo genera salidas (a través de YieldOutputAsync), esas salidas se reenvían como mensajes a ejecutores conectados en el flujo de trabajo principal. Esto permite que los ejecutores de nivel inferior procesen los resultados del sub-flujo de trabajo.

La ExecutorOptions clase controla este comportamiento:

Cuando AutoYieldOutputHandlerResultObject está habilitado, los subflujos de trabajo eluden el enrutamiento interno del flujo de trabajo principal y se entregan directamente al llamador del flujo de trabajo principal.

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

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

Solicitudes y respuestas

Los subprocesos admiten completamente el mecanismo de solicitud y respuesta . Cuando un ejecutor dentro del subflujo de trabajo envía una solicitud (por ejemplo, para solicitar la entrada humana), el WorkflowHostExecutor la reenvía al RequestInfoEvent flujo de trabajo principal con un identificador de puerto completo. El identificador del ejecutor del subflujo de trabajo se antepone al identificador de puerto (por ejemplo, SubWorkflow.GuessNumber).

Esta calificación garantiza que cuando el flujo de trabajo primario recibe una respuesta, puede enrutar la respuesta a la instancia correcta del sub-flujo de trabajo. El flujo de trabajo primario controla las solicitudes de sub-flujo de trabajo mediante el mismo mecanismo de respuesta que cualquier otra solicitud:

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

Funcionamiento

Cuando el flujo de trabajo primario enruta un mensaje al ejecutor del sub-flujo de trabajo:

1. Entrega de entrada : el mensaje se reenvía al ejecutor de inicio del flujo de trabajo interno. Con BindAsExecutor, el tipo de mensaje debe coincidir con los tipos esperados del ejecutor de inicio. Con AsAIAgent, los mensajes se normalizan para dar ChatMessage formato.
2. Ejecución interna — el flujo de trabajo interno ejecuta su propio bucle superstep.
3. Colección de salida : se recopilan los eventos de salida del flujo de trabajo interno. Con BindAsExecutor, las salidas conservan sus tipos originales. Con AsAIAgent, las salidas se convierten en mensajes de respuesta del agente.
4. Reenvío de solicitudes: si el flujo de trabajo interno tiene solicitudes pendientes, se reenvían al flujo de trabajo primario para controlar (consulte Solicitudes y respuestas).
5. Envío de bajada : los mensajes resultantes se envían al siguiente ejecutor del flujo de trabajo primario.

Dado que el flujo de trabajo interno mantiene su propio contexto de ejecución, su estado es independiente del flujo de trabajo primario.

Anidamiento de varios niveles

Los subflujos de trabajo pueden anidarse a una profundidad arbitraria. Cada nivel mantiene su propio contexto de ejecució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();

Tratamiento de errores

Cuando se produce un error en un sub-flujo de trabajo, el error se propaga al flujo de trabajo primario como .SubworkflowErrorEvent El flujo de trabajo primario puede observar estos errores a través de su flujo de eventos:

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

Si el sub-flujo de trabajo encuentra una excepción no controlada, la ejecución del flujo de trabajo primario continúa, pero el ejecutor del sub-flujo de trabajo deja de procesar más mensajes.

Creación de punto de comprobación

Cuando se toma un punto de control en el workflow principal, el estado de sesión del agente del sub-workflow se serializa como parte de los datos del punto de control del ejecutor principal. Al restaurar, el estado de sesión se deserializa, lo que permite que el flujo de trabajo primario se reanude con el estado del subproceso 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);

Creación de un Sub-Workflow

En Python, se crea un subflujo de trabajo encapsulando un Workflow dentro de un WorkflowExecutor y agregándolo a un flujo de trabajo padre.

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

El flujo de trabajo interno se ejecuta como un solo paso desde la perspectiva del flujo de trabajo primario. El coordinador envía mensajes a la canalización de análisis, que ejecuta specialist1 → specialist2internamente y, a continuación, reenvía el resultado al revisor.

Parámetros del Ejecutor de Flujo de Trabajo

Encapsulación de subflujos de trabajo

Encapsula explícitamente las instancias de Workflow en un WorkflowExecutor antes de agregarlas a un flujo de trabajo principal. Los agentes pueden pasarse directamente a WorkflowBuilder, pero las instancias directas de Workflow requieren este envoltorio.

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

El ajuste explícito le permite:

• Asigne un identificador de ejecutor específico como referencia en varios nodos.
• Vuelva a usar la misma WorkflowExecutor instancia en el 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 y salida

WorkflowExecutor hereda su firma de tipo del flujo de trabajo envolvente:

• Los tipos de entrada coinciden con los tipos de entrada iniciales del ejecutor del flujo de trabajo ajustado (además de SubWorkflowResponseMessage para gestionar las respuestas a las solicitudes reenviadas).
• Los tipos de salida coinciden con los tipos de salida del flujo de trabajo ajustado. Si algún ejecutor del sub-flujo de trabajo es compatible con solicitud-respuesta, SubWorkflowRequestMessage también se incluye como tipo de salida.

Esto significa que los bordes del flujo de trabajo primario deben conectar ejecutores cuyos tipos de salida coincidan con los tipos de entrada esperados del sub-flujo de trabajo. Del mismo modo, los ejecutores posteriores deben aceptar los tipos que produce el subproceso.

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

Comportamiento de salida

De forma predeterminada (allow_direct_output=False), cuando un sub-flujo de trabajo genera salidas a través de yield_output, esas salidas se reenvían como mensajes a ejecutores conectados en el flujo de trabajo primario mediante send_message. Esto permite que los ejecutores de nivel inferior procesen los resultados del sub-flujo de trabajo como parte del grafo primario.

Cuando allow_direct_output=True, las salidas del subflujo de trabajo se producen directamente en el flujo de eventos del flujo de trabajo principal. Las salidas del sub-flujo de trabajo se convierten en salidas del flujo de trabajo padre, omitiendo el enrutamiento interno del ejecutor del flujo de trabajo padre.

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

Emisiones intermedias de flujos de trabajo secundarios

Los eventos "intermediate" generados dentro de un flujo de trabajo secundario pasan automáticamente al flujo de eventos del flujo de trabajo principal. Se atribuyen a los WorkflowExecutor propios de id (no al ejecutor interno que los emitió originalmente), lo que preserva la encapsulación. Lo importante es que estos eventos mantienen la etiqueta "intermediate" independientemente de cómo el elemento principal designe el WorkflowExecutor en sus propias listas output_from o 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}")

Solicitudes y respuestas

Los subprocesos admiten completamente el mecanismo de solicitud y respuesta . Cuando un ejecutor dentro de un sub-flujo de trabajo llama a ctx.request_info(), WorkflowExecutor intercepta la solicitud y la controla en función de la configuración propagate_request.

Interceptar solicitudes en el flujo de trabajo primario (valor predeterminado)

Con propagate_request=False (valor predeterminado), las solicitudes del sub-flujo de trabajo se encapsulan en SubWorkflowRequestMessage y se envían a ejecutores conectados en el flujo de trabajo padre. Esto permite que los ejecutores primarios controle la solicitud 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)

El create_response() método valida que el tipo de datos de respuesta coincide con el tipo esperado de la solicitud original. Si los tipos no coinciden, se genera un TypeError.

Propagación de peticiones a llamadores externos

Con propagate_request=True, las solicitudes del sub-flujo de trabajo se propagan al flujo de eventos del flujo de trabajo primario mediante el mecanismo estándar request_info . El autor de la llamada del flujo de trabajo primario controla estas solicitudes de la misma manera que cualquier otra solicitud human-in-the-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)

Funcionamiento

Cuando el flujo de trabajo primario enruta un mensaje a WorkflowExecutor:

1. Entrega de entrada : el mensaje se reenvía al ejecutor de inicio del flujo de trabajo interno. El tipo de mensaje debe coincidir con los tipos de entrada esperados del ejecutor de inicio.
2. Ejecución interna : el flujo de trabajo interno ejecuta su propio bucle superstep hasta su finalización, o hasta que necesite una entrada externa.
3. Colección de salida : los eventos de salida del flujo de trabajo interno se recopilan y reenvían en función de la allow_direct_output configuración.
4. Reenvío de solicitudes: si el flujo de trabajo interno tiene solicitudes pendientes, se reenvían en función de la propagate_request configuración (consulte Solicitudes y respuestas).
5. AcumulaciónWorkflowExecutor de respuesta: recopila respuestas y reanuda el sub-flujo de trabajo solo cuando se han recibido todas las respuestas esperadas para una ejecución determinada.
6. Despacho aguas abajo — las salidas se envían al siguiente ejecutor en el flujo de trabajo principal.

El subflujo de trabajo mantiene su propio estado interno independientemente del elemento primario. Los mensajes se enrutan solo a través de los bordes que conectan el WorkflowExecutor con el resto del gráfico primario. No hay difusión de mensajes entre niveles de anidamiento.

Anidamiento de varios niveles

Los subflujos de trabajo pueden anidarse a una profundidad arbitraria. Cada nivel mantiene su propio contexto de ejecució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()
)

Tratamiento de errores

Cuando se produce un error en un sub-flujo de trabajo, el error se propaga al flujo de trabajo primario. WorkflowExecutor Captura el evento con errores del sub-flujo de trabajo y lo convierte en un evento de error en el contexto primario:

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)

Si el sub-flujo de trabajo encuentra una excepción no controlada, el flujo de trabajo primario recibe un evento de error con los detalles de la excepción, incluido el identificador del subproceso.

Creación de punto de comprobación

Los subprocesos admiten puntos de control. Cuando se toma un punto de control en el flujo de trabajo primario, WorkflowExecutor serializa su estado interno, incluido el progreso de ejecución del flujo de trabajo interno y los mensajes almacenados en caché. Al restaurar, este estado se deserializa, lo que permite que el flujo de trabajo primario se reanude con el sub-flujo de trabajo 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)

Creación de un Sub-Workflow

En Go, se crea un subflujo de trabajo creando un *workflow.Workflow y vinculándolo al flujo de trabajo principal con 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
}

El flujo de trabajo hijo vinculado se ejecuta como un único ejecutor desde la perspectiva del flujo de trabajo padre. Los mensajes entran a través de la vinculación, el flujo de trabajo secundario ejecuta su propio grafo interno y las salidas del flujo de trabajo secundario se redirigen de vuelta al grafo padre.

Tipos de entrada y salida

La vinculación hereda el protocolo del flujo de trabajo encapsulado. El flujo de trabajo primario puede enviar mensajes cuyos tipos en tiempo de ejecución coinciden con los tipos de entrada aceptados del flujo de trabajo secundario y el enlace expone los tipos de salida devueltos del flujo de trabajo secundario como tipos de mensaje y tipos de salida.

Esto significa que los bordes del flujo de trabajo primario deben conectar ejecutores cuyos tipos de salida coincidan con las entradas aceptadas del subproceso y los ejecutores descendentes deben controlar los tipos que produce el flujo de trabajo secundario:

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

Comportamiento de salida

Cuando un flujo de trabajo hijo genera un resultado, la vinculación del subflujo de trabajo envía ese resultado como un mensaje desde la vinculación a los ejecutores conectados en el flujo de trabajo principal. Si el flujo de trabajo primario también marca el enlace de sub-flujo de trabajo con WithOutputFrom, se emite el mismo valor que un elemento primario workflow.OutputEvent cuyo ExecutorID es el identificador de enlace del sub-flujo de trabajo.

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

Los eventos de flujo de trabajo personalizados emitidos en el flujo de trabajo secundario se reenvían al flujo de eventos del flujo de trabajo principal. Los propios eventos del ciclo de vida de inicio y de superpaso del subflujo de trabajo se mantienen internos, de modo que el flujo principal siga centrado en eventos con relevancia externa.

Solicitudes y respuestas

Los subprocesos admiten el mecanismo de solicitud y respuesta . Cuando un ejecutor dentro del subflujo de trabajo publica una solicitud externa, el enlace del subflujo de trabajo completa el identificador del puerto de solicitud anteponiendo el identificador del enlace. Por ejemplo, un puerto de solicitud secundario denominado ApprovalPort se convierte ApprovalSubWorkflow.ApprovalPort en el flujo de trabajo primario.

Para exponer la solicitud hija a través del flujo de trabajo principal, agregue un elemento principal RequestPort con el identificador cualificado y enrute las solicitudes y respuestas entre el enlace del subflujo de trabajo y ese puerto:

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

Quien llama gestiona la solicitud del flujo de trabajo principal y envía la respuesta de vuelta a través del mismo descriptor de ejecución. La vinculación del subflujo de trabajo elimina el prefijo calificado antes de entregar la respuesta al flujo de trabajo hijo.

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 mantener estrechos los bordes de solicitud y respuesta:

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

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

Funcionamiento

Cuando el flujo de trabajo principal dirige un mensaje a la vinculación del subflujo de trabajo:

1. Entrega de entradas — el enlace acepta mensajes que coinciden con los tipos de entrada que acepta el flujo de trabajo secundario y los encola en el ejecutor de inicio del flujo de trabajo secundario.
2. Ejecución interna — el flujo de trabajo secundario se ejecuta en el mismo entorno de ejecución dentro del proceso y mantiene su propio bucle de superpasos.
3. Reenvío de salida — los valores del elemento secundario workflow.OutputEvent se envían como mensajes desde el enlace a los ejecutores primarios posteriores, y también los devuelve el elemento primario si el enlace aparece en WithOutputFrom.
4. Reenvío de solicitudes: las solicitudes secundarias workflow.RequestInfoEvent se vuelven a emitir con identificadores de puerto calificados y se pueden enrutar a través de enlaces primarios RequestPort .
5. Reenvío de eventos — los eventos personalizados de flujos de trabajo secundarios se añaden al flujo principal. Los errores aparecen como valores principales workflow.ErrorEvent, con el identificador del subflujo de trabajo registrado.
6. Envío de bajada : los mensajes resultantes continúan a través de los bordes primarios del flujo de trabajo.

El flujo de trabajo hijo mantiene separados su estado y el enrutamiento de mensajes del flujo de trabajo padre. Los mensajes cruzan el límite solo a través de los bordes conectados al enlace de sub-flujo de trabajo.

Anidamiento de varios niveles

Los subflujos de trabajo pueden anidarse a una profundidad arbitraria. Cada flujo de trabajo secundario se enlaza antes de agregarlo al flujo de trabajo que lo contiene:

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

Tratamiento de errores

Cuando un flujo de trabajo secundario emite un error, el enlace de subproceso lo reenvía al flujo de trabajo primario como workflow.ErrorEvent y establece SubWorkflowID en el identificador de enlace. El flujo de trabajo primario puede observar estos errores a través de la misma secuencia de eventos que usa para los errores de flujo de trabajo de nivel 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)
    }
}

Los errores que se producen durante el reenvío de eventos hijo también se convierten en valores del elemento padre workflow.ErrorEvent con el ID del subflujo de trabajo adjunto.

Creación de punto de comprobación

Los subprocesos admiten puntos de control. Cuando el flujo de trabajo principal alcanza un punto de control, el enlace del subflujo de trabajo almacena el administrador de puntos de control del flujo de trabajo secundario y cualquier asignación cualificada pendiente de puertos de respuesta en el estado del ejecutor principal. Al restaurarse, el flujo de trabajo secundario puede reanudarse conservando intacto su estado de ejecución anidado, incluidas las solicitudes pendientes.

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)

Si se restaura un punto de control mientras un flujo de trabajo secundario tiene una solicitud pendiente, la ejecución primaria restaurada vuelve a publicar el evento de información de solicitud calificado. El autor de la llamada puede crear una respuesta a partir de esa solicitud republicada y devolverla a través del identificador de ejecución primario.