Sottoflussi di lavoro

Creazione di un Sub-Workflow

In C# i flussi di lavoro secondari vengono creati in due modi:

• Associazione diretta : usare BindAsExecutor() per incorporare un flusso di lavoro direttamente come executor nel flusso di lavoro padre. In questo modo vengono mantenuti i tipi di input/output nativi del flusso di lavoro secondario.
• Involucro agente — usare AsAIAgent() per convertire un flusso di lavoro in un agente, quindi aggiungere l'agente al flusso di lavoro padre. Ciò è utile quando il flusso di lavoro padre utilizza esecutori basati su agenti.

Associazione diretta con BindAsExecutor

Il BindAsExecutor() metodo di estensione converte un flusso di lavoro in un oggetto ExecutorBinding che può essere aggiunto direttamente a un flusso di lavoro padre:

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, i tipi di input e output tipizzato del flusso di lavoro secondario vengono mantenuti. Il flusso di lavoro padre instrada i messaggi in base ai tipi effettivi previsti e prodotti dal flusso di lavoro secondario.

Packaging dell'agente con AsAIAgent

Quando il flusso di lavoro padre utilizza esecutori basati su agenti, convertire il flusso di lavoro interno in un agente utilizzando AsAIAgent(). WorkflowBuilder avvolge automaticamente l'agente in un 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();

Il flusso di lavoro interno viene eseguito come singolo passaggio dal punto di vista del flusso di lavoro padre. Il coordinatore invia messaggi alla pipeline di analisi, che esegue specialist1 → specialist2internamente e quindi inoltra il risultato al revisore.

Tipi di input e output

Quando un flusso di lavoro viene usato come flusso di lavoro secondario, mantiene i contratti di tipo dei relativi executor interni.

Con BindAsExecutor, l'executor del flusso di lavoro secondario accetta gli stessi tipi di input dell'executor di avvio del flusso di lavoro interno e invia gli stessi tipi di output prodotti dal flusso di lavoro interno. I bordi del flusso di lavoro padre devono connettere executor i cui tipi di output corrispondono ai tipi di input previsti del flusso di lavoro secondario e i tipi di output del flusso di lavoro secondario devono corrispondere agli input previsti degli executor downstream.

Con AsAIAgent, il flusso di lavoro secondario viene incapsulato come agente e segue i contratti di input/output dell'Agent Executor (string, ChatMessage, IEnumerable<ChatMessage>).

Comportamento di output

Per impostazione predefinita, quando un flusso di lavoro secondario produce output (tramite YieldOutputAsync), tali output vengono inoltrati come messaggi agli executor connessi nel flusso di lavoro padre. Ciò consente agli executor downstream di elaborare i risultati del flusso di lavoro secondario.

La ExecutorOptions classe controlla questo comportamento:

Quando AutoYieldOutputHandlerResultObject è abilitato, gli output del flusso di lavoro secondario ignorano il routing interno dell'elemento padre e vengono recapitati direttamente al chiamante del flusso di lavoro padre.

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

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

Richieste e risposte

I flussi di lavoro secondari supportano completamente il meccanismo di richiesta e risposta . Quando un executor all'interno del flusso di lavoro secondario invia una richiesta (ad esempio, per richiedere l'input umano), l'WorkflowHostExecutorRequestInfoEvent viene inoltrato al flusso di lavoro padre con un ID di porta qualificato, ovvero l'ID dell'executor del flusso di lavoro secondario viene anteposto all'ID porta (ad esempio, SubWorkflow.GuessNumber).

Questa qualificazione garantisce che quando il flusso di lavoro padre riceve una risposta, può instradare la risposta all'istanza corretta del flusso di lavoro secondario. Il flusso di lavoro padre gestisce le richieste del flusso di lavoro secondario usando lo stesso meccanismo di risposta di qualsiasi altra richiesta:

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

Funzionamento

Quando il flusso di lavoro padre instrada un messaggio all'executor del flusso di lavoro secondario:

1. Recapito di input: il messaggio viene inoltrato all'esecutore di avvio del flusso di lavoro interno. Con BindAsExecutor, il tipo di messaggio deve corrispondere ai tipi previsti dell'executor di avvio. Con AsAIAgent, i messaggi vengono normalizzati al formato ChatMessage.
2. Esecuzione interna — il flusso di lavoro interno esegue il proprio ciclo di superstep.
3. Raccolta di output : vengono raccolti gli eventi di output del flusso di lavoro interno. Con BindAsExecutor, gli output mantengono i tipi originali. Con AsAIAgent, gli output vengono convertiti in messaggi di risposta dell'agente.
4. Inoltro delle richieste : se il flusso di lavoro interno ha richieste in sospeso, vengono inoltrate al flusso di lavoro padre per la gestione (vedere Richieste e risposte).
5. Invio a valle: i messaggi risultanti vengono inviati all'executor successivo del flusso di lavoro padre.

Poiché il flusso di lavoro interno mantiene il proprio contesto di esecuzione, il relativo stato è indipendente dal flusso di lavoro padre.

Annidamento a più livelli

I flussi di lavoro secondari possono essere annidati in profondità arbitraria. Ogni livello mantiene il proprio contesto di esecuzione:

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

Gestione degli errori

Quando un flusso di lavoro secondario ha esito negativo, l'errore viene propagato al flusso di lavoro padre come .SubworkflowErrorEvent Il flusso di lavoro padre può osservare questi errori tramite il flusso di eventi:

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

Se il flusso di lavoro secondario rileva un'eccezione non gestita, l'esecuzione del flusso di lavoro padre continua ma l'executor del flusso di lavoro secondario interrompe l'elaborazione di altri messaggi.

Checkpoint

Quando viene eseguito un checkpoint nel flusso di lavoro padre, lo stato della sessione dell'agente del flusso di lavoro secondario viene serializzato come parte dei dati del checkpoint dell'executor padre. Al ripristino, lo stato della sessione viene deserializzato, consentendo al flusso di lavoro padre di riprendere con lo stato del flusso di lavoro secondario intatto.

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

Creazione di un Sub-Workflow

In Python si crea un flusso di lavoro secondario eseguendo il wrapping di un Workflow oggetto in un WorkflowExecutor oggetto e aggiungendolo a un flusso di lavoro 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()
)

Il flusso di lavoro interno viene eseguito come singolo passaggio dal punto di vista del flusso di lavoro padre. Il coordinatore invia messaggi alla pipeline di analisi, che esegue specialist1 → specialist2internamente e quindi inoltra il risultato al revisore.

Parametri WorkflowExecutor

Raggruppamento di sottoflussi di lavoro

Racchiudere esplicitamente le istanze Workflow in un elemento WorkflowExecutor prima di aggiungerle a un workflow padre. Gli agenti possono essere passati direttamente a WorkflowBuilder, ma le istanze Workflow grezze richiedono questo 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()
)

La racchiusura esplicita consente di:

• Assegnare un ID esecutore specifico da utilizzare come riferimento in più archi.
• Riutilizzare la stessa WorkflowExecutor istanza nel grafico.

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

Tipi di input e output

WorkflowExecutor eredita la firma del tipo dal flusso di lavoro incapsulato.

• I tipi di input corrispondono ai tipi di input dell'esecutore di avvio del flusso di lavoro incapsulato (più SubWorkflowResponseMessage per la gestione delle risposte alle richieste reindirizzate).
• I tipi di output corrispondono ai tipi di output del flusso di lavoro di cui è stato eseguito il wrapping. Se un executor nel flusso di lavoro secondario è in grado di rispondere alle richieste, SubWorkflowRequestMessage viene incluso anche come tipo di output.

Ciò significa che i bordi del flusso di lavoro padre devono connettere executor i cui tipi di output corrispondono ai tipi di input previsti del flusso di lavoro secondario. Analogamente, gli executor downstream devono accettare i tipi prodotti dal flusso di lavoro secondario:

# 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 di output

Per impostazione predefinita (allow_direct_output=False), quando un flusso di lavoro secondario produce output tramite yield_output, tali output vengono inoltrati come messaggi agli executor connessi nel flusso di lavoro padre usando send_message. Ciò consente agli executor downstream di elaborare i risultati del flusso di lavoro secondario come parte del grafico padre.

Quando allow_direct_output=True, gli output del flusso di lavoro secondario vengono restituiti direttamente al flusso di eventi del flusso di lavoro padre. Gli output del flusso di lavoro secondario diventano output del flusso di lavoro padre, ignorando il routing interno dell'executor 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)

Emissioni intermedie dai flussi di lavoro figlio

"intermediate" gli eventi generati all'interno di un workflow figlio risalgono automaticamente attraverso il flusso di eventi del workflow padre. Sono attribuiti al WorkflowExecutor stesso id (non all'esecutore interno che li ha originariamente emessi), il che preserva l'incapsulamento. È fondamentale che questi eventi mantengano l'etichetta "intermediate" indipendentemente da come l'elemento padre designa il WorkflowExecutor nei propri elenchi 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}")

Richieste e risposte

I flussi di lavoro secondari supportano completamente il meccanismo di richiesta e risposta . Quando un executor all'interno di un flusso di lavoro secondario chiama ctx.request_info(), WorkflowExecutor intercetta la richiesta e la gestisce in base all'impostazione propagate_request .

Intercettazione delle richieste nel flusso di lavoro padre (impostazione predefinita)

Con propagate_request=False (impostazione predefinita), le richieste del sottoflusso di lavoro vengono incluse in un SubWorkflowRequestMessage e inviate agli executor connessi nel flusso di lavoro padre. In questo modo gli executor padre possono gestire la richiesta in locale:

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)

Il create_response() metodo convalida che il tipo di dati della risposta corrisponda al tipo previsto dalla richiesta originale. Se i tipi non corrispondono, viene generato un oggetto TypeError .

Propagazione delle richieste ai chiamanti esterni

Con propagate_request=True, le richieste dal flusso di lavoro secondario vengono propagate al flusso di eventi del flusso di lavoro padre usando il meccanismo standard request_info . Il chiamante del flusso di lavoro padre gestisce queste richieste allo stesso modo di qualsiasi altra richiesta di tipo 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)

Funzionamento

Quando il flusso di lavoro padre instrada un messaggio a WorkflowExecutor:

1. Recapito di input: il messaggio viene inoltrato all'esecutore di avvio del flusso di lavoro interno. Il tipo di messaggio deve corrispondere ai tipi di input previsti dell'executor start.
2. Esecuzione interna: il flusso di lavoro interno esegue il proprio ciclo di superstep fino a quando non è completato o fino a quando necessita di input esterno.
3. Raccolta di output: gli eventi di output del flusso di lavoro interno vengono raccolti e inoltrati in base all'impostazione allow_direct_output .
4. Inoltro delle richieste : se il flusso di lavoro interno ha richieste in sospeso, vengono inoltrate in base all'impostazione propagate_request (vedere Richieste e risposte).
5. Accumulo di risposta : WorkflowExecutor raccoglie le risposte e riprende il flusso di lavoro secondario solo quando sono state ricevute tutte le risposte previste per una determinata esecuzione.
6. Dispatch downstream : gli output vengono inviati all'executor successivo nel flusso di lavoro padre.

Il flusso di lavoro secondario mantiene il proprio stato interno indipendentemente dall'elemento padre. I messaggi vengono instradati solo attraverso i bordi che collegano l'oggetto WorkflowExecutor al resto del grafico padre, senza trasmissione di messaggi tra i livelli di annidamento.

Annidamento a più livelli

I flussi di lavoro secondari possono essere annidati in profondità arbitraria. Ogni livello mantiene il proprio contesto di esecuzione:

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

Gestione degli errori

Quando un flusso di lavoro secondario ha esito negativo, l'errore viene propagato al flusso di lavoro padre. WorkflowExecutor Acquisisce l'evento non riuscito dal flusso di lavoro secondario e lo converte in un evento di errore nel contesto padre:

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 il flusso di lavoro secondario rileva un'eccezione non gestita, il flusso di lavoro padre riceve un evento di errore con i dettagli dell'eccezione, incluso l'ID del flusso di lavoro secondario.

Checkpoint

I flussi di lavoro secondari supportano il checkpoint. Quando viene eseguito un checkpoint nel flusso di lavoro padre, il WorkflowExecutor serializza lo stato interno, incluso lo stato di esecuzione del flusso di lavoro interno ed eventuali messaggi memorizzati nella cache. Al ripristino, questo stato viene deserializzato, consentendo al flusso di lavoro padre di riprendere con il flusso di lavoro secondario intatto.

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)

Creazione di un Sub-Workflow

In Go, si crea un sottoflusso di lavoro creando un *workflow.Workflow e collegandolo al flusso di lavoro padre 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
}

Il flusso di lavoro figlio collegato viene eseguito come un unico esecutore dal punto di vista del flusso di lavoro padre. I messaggi vengono immessi tramite l'associazione, il flusso di lavoro figlio esegue il proprio grafico interno e gli output del flusso di lavoro figlio vengono indirizzati di nuovo al grafico padre.

Tipi di input e output

Il binding eredita il protocollo del workflow incapsulato. Il flusso di lavoro padre può inviare messaggi i cui tipi in fase di esecuzione corrispondono ai tipi di input accettati dal flusso di lavoro figlio, e il binding espone i tipi di output prodotti dal flusso di lavoro figlio sia come tipi di messaggio sia come tipi di output.

Ciò significa che i bordi del flusso di lavoro padre devono connettere executor i cui tipi di output corrispondono agli input accettati del flusso di lavoro secondario e gli executor downstream devono gestire i tipi restituiti dal flusso di lavoro figlio:

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 di output

Quando un flusso di lavoro figlio restituisce un output, l'associazione del flusso di lavoro secondario invia tale output come messaggio dall'associazione agli executor connessi nel flusso di lavoro padre. Se il flusso di lavoro padre contrassegna anche l'associazione del flusso di lavoro secondario con WithOutputFrom, lo stesso valore viene generato come elemento padre workflow.OutputEvent il cui ExecutorID è l'ID di associazione del flusso di lavoro secondario.

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

Gli eventi del flusso di lavoro personalizzati generati all'interno del flusso di lavoro figlio vengono inoltrati al flusso di eventi padre. Gli eventi del ciclo di vita relativi all’avvio e ai superstep del workflow secondario vengono mantenuti interni, in modo che il flusso principale rimanga incentrato sugli eventi significativi per l’esterno.

Richieste e risposte

I flussi di lavoro secondari supportano il meccanismo di richiesta e risposta . Quando un executor all'interno del flusso di lavoro secondario invia una richiesta esterna, l'associazione del flusso di lavoro secondario qualifica l'ID porta della richiesta anteponendo l'ID di associazione. Ad esempio, una porta di richiesta figlia denominata ApprovalPort diventa ApprovalSubWorkflow.ApprovalPort nel workflow padre.

Per visualizzare la richiesta figlio tramite il flusso di lavoro padre, aggiungere un elemento padre RequestPort con l'ID completo e indirizzare le richieste e le risposte tra l'associazione del flusso di lavoro secondario e tale 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()

Il chiamante gestisce la richiesta dal flusso di lavoro padre e invia la risposta tramite lo stesso handle di esecuzione. L'associazione del flusso di lavoro secondario rimuove il prefisso qualificato prima di recapitare la risposta al flusso di lavoro figlio.

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

Usare i predicati per mantenere stretti i bordi della richiesta e della risposta:

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

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

Funzionamento

Quando il workflow principale instrada un messaggio al binding del sub-workflow:

1. Recapito dell'input — l'associazione accetta i messaggi che corrispondono ai tipi di input accettati dal flusso di lavoro figlio e li accoda nell'esecutore di avvio del flusso di lavoro figlio.
2. Esecuzione interna — il workflow figlio viene eseguito nello stesso ambiente di esecuzione nello stesso processo e mantiene il proprio ciclo di superstep.
3. Inoltro di output : i valori figlio workflow.OutputEvent vengono inviati come messaggi dall'associazione agli executor padre downstream e vengono restituiti anche dall'elemento padre se l'associazione è elencata in WithOutputFrom.
4. Inoltro delle richieste — le richieste figlie workflow.RequestInfoEvent vengono riemesse con ID di porta qualificati e possono essere instradate tramite binding padre RequestPort.
5. Inoltro eventi — gli eventi personalizzati del flusso di lavoro figlio vengono aggiunti al flusso principale. Gli errori vengono visualizzati come valori padre workflow.ErrorEvent con l'ID del flusso di lavoro secondario registrato.
6. Invio downstream : i messaggi risultanti continuano attraverso i bordi del flusso di lavoro padre.

Il workflow figlio mantiene il proprio stato e il routing dei messaggi separati da quelli del workflow padre. I messaggi superano il limite solo attraverso i bordi connessi all'associazione del flusso di lavoro secondario.

Annidamento a più livelli

I flussi di lavoro secondari possono essere annidati in profondità arbitraria. Ogni flusso di lavoro figlio viene vincolato prima di essere aggiunto al flusso di lavoro che 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()

Gestione degli errori

Quando un flusso di lavoro figlio genera un errore, l'associazione del flusso di lavoro secondario lo inoltra al flusso di lavoro padre come workflow.ErrorEvent e imposta SubWorkflowID sull'ID di associazione. Il flusso di lavoro padre può osservare questi errori tramite lo stesso flusso di eventi usato per gli errori del flusso di lavoro di primo livello:

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

Gli errori generati durante l'inoltro degli eventi figlio vengono convertiti anche in valori padre workflow.ErrorEvent con l'ID del flusso di lavoro secondario associato.

Checkpoint

I flussi di lavoro secondari supportano il checkpoint. Quando il workflow padre esegue un checkpoint, il binding del sottoflusso di lavoro archivia il gestore dei checkpoint del workflow figlio e tutte le eventuali mappature qualificate delle porte di risposta in sospeso nello stato dell'esecutore padre. Al ripristino, il flusso di lavoro figlio può riprendere con lo stato di esecuzione annidato intatto, incluse le richieste in sospeso.

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 un checkpoint viene ripristinato mentre un flusso di lavoro figlio ha una richiesta in sospeso, l'esecuzione padre ripristinata ripubblica l'evento con le informazioni sulla richiesta qualificata. Il chiamante può creare una risposta da tale richiesta ripubblicata e inviarla nuovamente tramite l'handle di esecuzione padre.