Sub-Workflows

Een subwerkstroom is een volledige werkstroom die wordt uitgevoerd als een uitvoerder binnen een bovenliggende werkstroom. Hierdoor kunt u complexe systemen samenstellen op basis van kleinere, herbruikbare werkstroombouwstenen, elk met een eigen geïsoleerde uitvoeringscontext, statusbeheer en berichtroutering.

Overview

Subwerkstromen zijn handig als u het volgende wilt doen:

• De complexiteit opsplitsen : breek een grote werkstroom in kleinere, onafhankelijk testbare eenheden.
• Werkstroomlogica opnieuw gebruiken : sluit dezelfde subwerkstroom in meerdere bovenliggende werkstromen in.
• Isolatiestatus : zorg ervoor dat de interne status van elke subwerkstroom gescheiden blijft van het bovenliggende item.
• Gegevensstroom beheren: berichten voeren en verlaten de subwerkstroom alleen via de randen, zonder dat ze op verschillende niveaus worden uitgezonden.

Wanneer een subwerkstroom wordt toegevoegd aan een bovenliggende werkstroom, gedraagt deze zich net als elke andere uitvoerder: het ontvangt invoerberichten, voert de interne grafiek uit tot voltooiing en produceert uitvoerberichten voor downstream-uitvoerders.

Een Sub-Workflow maken

In C# stelt u subwerkstromen op twee manieren op:

• Directe binding — gebruik BindAsExecutor() om een werkstroom rechtstreeks als uitvoerder in de bovenliggende werkstroom in te sluiten. Hierdoor blijven de systeemeigen invoer-/uitvoertypen van de subwerkstroom behouden.
• Agentwikkeling — hiermee AsAIAgent() converteert u een werkstroom naar een agent en voegt u de agent toe aan de hoofdwerkstroom. Dit is handig wanneer de bovenliggende werkstroom gebruikmaakt van agent-gebaseerde executors.

Directe binding met BindAsExecutor

Met de BindAsExecutor() extensiemethode wordt een werkstroom geconverteerd naar een ExecutorBinding werkstroom die rechtstreeks aan een bovenliggende werkstroom kan worden toegevoegd:

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

De BindAsExecutorgetypte invoer- en uitvoertypen van de subwerkstroom blijven behouden. De bovenliggende werkstroom routeert berichten op basis van de werkelijke typen die de subwerkstroom verwacht en produceert.

Agentomwikkeling met AsAIAgent

Wanneer de bovenliggende werkstroom gebruikmaakt van op agents gebaseerde uitvoerders, converteert u de interne werkstroom naar een agent met behulp van AsAIAgent(). De WorkflowBuilder agent wordt automatisch verpakt in een uitvoerprogramma:

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

De interne werkstroom wordt vanuit het perspectief van de hoofdwerkstroom als één stap uitgevoerd. De coördinator verzendt berichten naar de analysepijplijn, die intern wordt uitgevoerd specialist1 → specialist2en stuurt het resultaat vervolgens door naar de revisor.

Tip

Gebruik BindAsExecutor() bij het werken met getypte uitvoerders en AsAIAgent() bij het werken met werkstromen op basis van agents. Zie Werkstromen als agenten voor meer informatie over het configureren van de conversie van werkstroom naar agent.

Invoer- en uitvoertypen

Wanneer een werkstroom wordt gebruikt als subwerkstroom, behoudt deze het type contracten van de interne uitvoerders.

Met BindAsExecutor, accepteert de subwerkstroomuitvoerer dezelfde invoertypen als de startexecutor van de binnenwerkstroom en verzendt dezelfde uitvoertypen die de interne werkstroom produceert. De randen van de bovenliggende werkstroom moeten uitvoerders verbinden waarvan de uitvoertypen overeenkomen met de verwachte invoertypen van de subwerkstroom en de uitvoertypen van de subwerkstroom moeten overeenkomen met de verwachte invoer van downstreamexecutors.

Met AsAIAgent wordt de subwerkstroom verpakt als een agent en volgt het de invoer-/uitvoercontracten van de Agent Executor (stringChatMessageIEnumerable<ChatMessage>).

Uitvoergedrag

Wanneer een subwerkstroom uitvoer produceert (via YieldOutputAsync), worden deze uitvoer standaard doorgestuurd als berichten naar verbonden uitvoerders in de bovenliggende werkstroom. Hierdoor kunnen downstreamexecutors subwerkstroomresultaten verwerken.

De ExecutorOptions klasse bepaalt dit gedrag:

Option	Verstek	Description
AutoSendMessageHandlerResultObject	true	Stuur subwerkstroomuitvoeringen als berichten naar verbonden uitvoerders in de bovenliggende graaf.
AutoYieldOutputHandlerResultObject	false	Genereer uitvoer van subwerkstromen direct naar de uitvoergebeurtenisstroom van de bovenliggende werkstroom.

Wanneer AutoYieldOutputHandlerResultObject is ingeschakeld, wordt de uitvoer van de subwerkstroom de interne routering van de bovenliggende werkstroom overgeslagen en rechtstreeks geleverd aan de aanroeper van de bovenliggende werkstroom.

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

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

Aanvragen en antwoorden

Subwerkstromen ondersteunen het mechanisme voor aanvragen en antwoorden volledig. Wanneer een uitvoerder in de subwerkstroom een aanvraag verzendt (bijvoorbeeld om menselijke invoer aan te vragen), wordt de WorkflowHostExecutorRequestInfoEvent naar de bovenliggende werkstroom doorgestuurd met een gekwalificeerde poort-id — de id van de subwerkstroom-uitvoerder wordt toegevoegd aan de poort-id (bijvoorbeeld SubWorkflow.GuessNumber).

Deze kwalificatie zorgt ervoor dat wanneer de bovenliggende werkstroom een antwoord ontvangt, het antwoord naar het juiste subwerkstroomexemplaar kan worden teruggestuurd. De bovenliggende werkstroom verwerkt subwerkstroomaanvragen met hetzelfde antwoordmechanisme als elke andere aanvraag:

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

Opmerking

Vanuit het perspectief van de initiator van de bovenliggende werkstroom is er geen verschil tussen een aanvraag van een hoofd-uitvoerder en een aanvraag van een subwerkstroom. Het framework verwerkt de routering transparant.

Hoe het werkt

Wanneer de bovenliggende werkstroom een bericht doorstuurt naar de subwerkstroomuitvoerders:

1. Invoerbezorging - het bericht wordt doorgestuurd naar de startuitvoerder van de interne werkstroom. Hiermee BindAsExecutormoet het berichttype overeenkomen met de verwachte typen van de startexecutor. Met AsAIAgent worden berichten genormaliseerd naar ChatMessage formaat.
2. Interne uitvoering : de interne werkstroom voert een eigen supersteplus uit.
3. Uitvoerverzameling : de uitvoer gebeurtenissen van de interne werkstroom worden verzameld. Met BindAsExecutor behouden uitvoer hun oorspronkelijke typen. Met AsAIAgentworden uitvoer geconverteerd naar antwoordberichten van de agent.
4. Doorsturen aanvragen : als de binnenste werkstroom aanvragen in behandeling heeft, worden deze doorgestuurd naar de bovenliggende werkstroom voor verwerking (zie Aanvragen en antwoorden).
5. Downstream-verzending : de resulterende berichten worden verzonden naar de volgende uitvoerder in de bovenliggende werkstroom.

Omdat de interne werkstroom een eigen uitvoeringscontext onderhoudt, is de status onafhankelijk van de bovenliggende werkstroom.

Tip

Zie Werkstromen als agents voor meer informatie over het configureren van de conversie van werkstroom naar agent, inclusief streaminggedrag en afhandeling van uitzonderingen.

Nesten op meerdere niveaus

Subwerkstromen kunnen worden genest naar willekeurige diepte. Elk niveau onderhoudt een eigen uitvoeringscontext:

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

Opmerking

Elk nestniveau voegt verwerkingsoverhead toe omdat de interne werkstroom zijn eigen superstep-loop uitvoert. Houd de nestdiepte redelijk voor prestatiegevoelige scenario's.

Foutafhandeling

Wanneer een subwerkstroom mislukt, wordt de fout als een SubworkflowErrorEvent naar de ouderwerkstroom doorgegeven. De bovenliggende werkstroom kan deze fouten waarnemen via zijn gebeurtenisstroom.

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

Als de subwerkstroom een niet-verwerkte uitzondering tegenkomt, wordt de uitvoering van de bovenliggende werkstroom voortgezet, maar de uitvoerfunctie van de subwerkstroom stopt met het verwerken van verdere berichten.

Controlepunten maken

Wanneer een controlepunt wordt uitgevoerd op de bovenliggende werkstroom, wordt de sessiestatus van de subwerkstroomagent geserialiseerd als onderdeel van de controlepuntgegevens van de bovenliggende uitvoerder. Bij herstel wordt de sessiestatus gedeserialiseerd, zodat de bovenliggende werkstroom kan worden hervat met de status van de subwerkstroom intact.

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

Een Sub-Workflow maken

In Python maakt u een subwerkstroom door een Workflow in een WorkflowExecutor te verpakken en deze toe te voegen aan een ouderworkflow.

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

De interne werkstroom wordt vanuit het perspectief van de hoofdwerkstroom als één stap uitgevoerd. De coördinator verzendt berichten naar de analysepijplijn, die intern wordt uitgevoerd specialist1 → specialist2en stuurt het resultaat vervolgens door naar de revisor.

WorkflowExecutor Parameterinstellingen

Parameter	Type	Verstek	Description
workflow	Workflow	—	Het werkstroomexemplaar om te verpakken als een uitvoerder.
id	str	—	Unieke id voor deze uitvoerder.
allow_direct_output	bool	False	Wanneer True subwerkstroomuitvoer rechtstreeks naar de gebeurtenisstroom van de bovenliggende werkstroom wordt geretourneerd, worden ze niet als berichten naar de aangesloten uitvoerders verzonden.
propagate_request	bool	False	Wanneer True aanvragen van de subwerkstroom worden doorgegeven aan de gebeurtenisstroom van de bovenliggende werkstroom, worden ze behandeld als reguliere aanvraaggegevens en gebeurtenissen. Wanneer False, aanvragen worden verpakt in SubWorkflowRequestMessage voor onderschepping door ouder uitvoerders.

Subworkflows inpakken

Verpak Workflow-instanties expliciet in een WorkflowExecutor voordat u ze toevoegt aan een bovenliggende workflow. Agents kunnen rechtstreeks worden doorgegeven aan WorkflowBuilder, maar voor ruwe Workflow-instanties is deze wrapper vereist.

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

Met expliciete terugloop kunt u het volgende doen:

• Wijs een specifieke uitvoerder-ID toe voor verwijzing in meerdere verbindingen.
• Gebruik hetzelfde WorkflowExecutor exemplaar opnieuw in de grafiek.

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

Invoer- en uitvoertypen

De WorkflowExecutor typesignatuur wordt overgenomen uit de ingekapselde werkstroom.

• Invoertypen komen overeen met de begininvoertypen van de werkstroom (plus SubWorkflowResponseMessage voor het verwerken van antwoorden op doorgestuurde aanvragen).
• Uitvoertypen komen overeen met de uitvoertypen van de verpakte werkstroom. Als een uitvoerder in de subwerkstroom geschikt is voor aanvraagrespons, SubWorkflowRequestMessage wordt deze ook opgenomen als uitvoertype.

Dit betekent dat de randen van de bovenliggende werkstroom uitvoerders moeten verbinden waarvan de uitvoertypen overeenkomen met de verwachte invoertypen van de subwerkstroom. Op dezelfde manier moeten downstreamexecutors de typen accepteren die de subwerkstroom produceert:

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

Uitvoergedrag

allow_direct_output=FalseWanneer een subwerkstroom uitvoer produceert via yield_output, worden deze uitvoer standaard doorgestuurd als berichten naar verbonden uitvoerders in de bovenliggende werkstroom met behulp van send_message. Hierdoor kunnen downstreamexecutors subwerkstroomresultaten verwerken als onderdeel van de bovenliggende grafiek.

Wanneer allow_direct_output=True de uitvoer van de subwerkstroom rechtstreeks wordt doorgegeven naar de gebeurtenisstroom van de bovenliggende werkstroom. De uitvoer van de subwerkstroom wordt uitvoer van de bovenliggende werkstroom, waardoor de interne uitvoerroutering van de bovenliggende werkstroom wordt overgeslagen:

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

Tussenliggende emissies van onderliggende werkstromen

"intermediate" gebeurtenissen die binnen een onderliggende werkstroom ontstaan, worden automatisch doorgegeven via de gebeurtenisstroom van de bovenliggende werkstroom. Ze worden toegeschreven aan de eigen WorkflowExecutor van id (niet aan de interne uitvoerder die ze oorspronkelijk heeft gegenereerd), waardoor de inkapseling behouden blijft. Het is cruciaal dat deze gebeurtenissen het "intermediate" label behouden, ongeacht hoe het bovenliggende item het WorkflowExecutor label in zijn eigen output_from- of intermediate_output_from-lijsten aanduidt.

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

Aanvragen en antwoorden

Subwerkstromen ondersteunen het mechanisme voor aanvragen en antwoorden volledig. Wanneer een uitvoerder binnen een subwerkstroom ctx.request_info() aanroept, onderschept de WorkflowExecutor het verzoek en verwerkt het op basis van de propagate_request instelling.

Aanvragen onderscheppen in de Hoofdwerkstroom (Standaard)

Met propagate_request=False (de standaardinstelling) worden aanvragen van de subwerkstroom verpakt in een SubWorkflowRequestMessage en verzonden naar verbonden uitvoerders in de bovenliggende werkstroom. Hierdoor kunnen bovenliggende uitvoerders de aanvraag lokaal verwerken:

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)

De create_response() methode valideert dat het antwoordgegevenstype overeenkomt met het verwachte type van de oorspronkelijke aanvraag. Als de typen niet overeenkomen, wordt een TypeError opgeworpen.

Important

Gebruik bij het terugsturen van het antwoord target_id=request.executor_id om SubWorkflowResponseMessage naar de juiste WorkflowExecutor instantie te routeren.

Aanvragen doorgeven aan externe bellers

Hiermee propagate_request=Trueworden aanvragen van de subwerkstroom doorgegeven aan de gebeurtenisstroom van de bovenliggende werkstroom met behulp van het standaardmechanisme request_info . De aanroeper van de bovenliggende werkstroom verwerkt deze aanvragen op dezelfde manier als elke andere human-in-the-loop-aanvraag:

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)

Hoe het werkt

Wanneer de bovenliggende werkstroom een bericht naar WorkflowExecutor stuurt:

1. Invoerbezorging - het bericht wordt doorgestuurd naar de startuitvoerder van de interne werkstroom. Het berichttype moet overeenkomen met de verwachte invoertypen van de startexecutor.
2. Interne uitvoering : de interne werkstroom voert een eigen supersteplus uit tot voltooiing of totdat er externe invoer nodig is.
3. Uitvoerverzameling : de uitvoergebeurtenissen van de interne werkstroom worden verzameld en doorgestuurd op basis van de allow_direct_output instelling.
4. Doorsturen aanvragen : als de interne werkstroom aanvragen in behandeling heeft, worden ze doorgestuurd op basis van de propagate_request instelling (zie Aanvragen en antwoorden).
5. Responsaccumulatie: de WorkflowExecutor verzamelt reacties en hervat de subwerkstroom alleen wanneer alle verwachte reacties voor een bepaalde uitvoering zijn ontvangen.
6. Downstream-verzending : uitvoer wordt verzonden naar de volgende uitvoerder in de bovenliggende werkstroom.

De subwerkstroom behoudt zijn eigen interne status onafhankelijk van de ouder. Berichten worden alleen gerouteerd via de randen die de WorkflowExecutor verbinden met de rest van de bovenliggende grafiek. Er is geen berichtuitzending over geneste niveaus.

Nesten op meerdere niveaus

Subwerkstromen kunnen worden genest naar willekeurige diepte. Elk niveau onderhoudt een eigen uitvoeringscontext:

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

Opmerking

Elk nestniveau voegt verwerkingsoverhead toe omdat de interne werkstroom zijn eigen superstep-loop uitvoert. Houd de nestdiepte redelijk voor prestatiegevoelige scenario's.

Warning

Alle gelijktijdige uitvoeringen van een WorkflowExecutor delen hetzelfde onderliggende werkstroomexemplaar. Uitvoerders in de subwerkstroom moeten staatloos zijn om interferentie tussen gelijktijdige uitvoeringen te voorkomen.

Foutafhandeling

Wanneer een subwerkstroom mislukt, wordt de fout doorgegeven aan de bovenliggende werkstroom. Hiermee WorkflowExecutor wordt de mislukte gebeurtenis uit de subwerkstroom vastgelegd en geconverteerd naar een foutgebeurtenis in de bovenliggende context:

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)

Als de subwerkstroom een niet-verwerkte uitzondering tegenkomt, ontvangt de bovenliggende werkstroom een foutgebeurtenis met de uitzonderingsgegevens, inclusief de id van de subwerkstroom.

Controlepunten maken

Subwerkstromen ondersteunen controlepunten. Wanneer een controlepunt wordt uitgevoerd op de bovenliggende werkstroom, wordt de WorkflowExecutor interne status geserialiseerd, inclusief de uitvoeringsvoortgang van de interne werkstroom en mogelijke gecachete berichten. Bij herstel wordt deze status gedeserialiseerd, zodat de bovenliggende werkstroom kan worden hervat met de subwerkstroom intact.

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)

Een Sub-Workflow maken

In Go maakt u een subwerkstroom door een *workflow.Workflow werkstroom te maken en deze te binden aan de bovenliggende werkstroom met 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
}

De gekoppelde onderliggende workflow draait als één executor vanuit het perspectief van de bovenliggende workflow. Berichten komen binnen via de binding, de onderliggende workflow voert zijn eigen interne graaf uit en de uitvoerwaarden van de onderliggende workflow worden teruggeleid naar de bovenliggende graaf.

Invoer- en uitvoertypen

De binding neemt het protocol van de verpakte werkstroom over. De bovenliggende workflow kan berichten verzenden waarvan de runtime-typen overeenkomen met de invoertypen die door de onderliggende workflow worden geaccepteerd, en de binding stelt de door de onderliggende workflow opgeleverde uitvoertypen beschikbaar als zowel berichttypen als uitvoertypen.

Dit betekent dat de randen van de bovenliggende werkstroom uitvoerders moeten verbinden waarvan de uitvoertypen overeenkomen met de geaccepteerde invoer van de subwerkstroom en downstreamuitvoerders de typen moeten verwerken die door de onderliggende werkstroom worden gegenereerd:

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

Uitvoergedrag

Wanneer een onderliggende werkstroom een uitvoer oplevert, verzendt de subwerkstroombinding die uitvoer als een bericht van de binding naar verbonden uitvoerders in de bovenliggende werkstroom. Als de bovenliggende werkstroom de subwerkstroombinding ook van WithOutputFrom voorziet, wordt dezelfde waarde uitgegeven als een bovenliggende workflow.OutputEvent, waarbij ExecutorID de bindings-id van de subwerkstroom is.

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

Aangepaste workflowgebeurtenissen die in de onderliggende workflow worden gegenereerd, worden doorgestuurd naar de gebeurtenisstream van de bovenliggende workflow. De eigen start- en superstep-levenscyclusgebeurtenissen van de subwerkstroom worden intern bewaard, zodat de bovenliggende stroom zich blijft concentreren op externe zinvolle gebeurtenissen.

Aanvragen en antwoorden

Subwerkstromen ondersteunen het mechanisme voor aanvragen en antwoorden . Wanneer een executor binnen de subworkflow een externe aanvraag verstuurt, specificeert de subworkflowbinding de ID van de aanvraagpoort door de bindings-ID ervoor te plaatsen. Bijvoorbeeld: een aanvraagpoort van een onderliggende werkstroom met de naam ApprovalPort wordt ApprovalSubWorkflow.ApprovalPort in de bovenliggende werkstroom.

Als u de onderliggende aanvraag zichtbaar wilt maken via de bovenliggende werkstroom, voegt u een bovenliggend RequestPort toe met de gekwalificeerde ID en routeert u aanvragen en antwoorden tussen de binding van de subwerkstroom en die poort:

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

De aanroeper verwerkt het verzoek uit de stream van de bovenliggende workflow en stuurt de reactie terug via dezelfde run-handle. De binding van de subworkflow verwijdert het gekwalificeerde voorvoegsel voordat het antwoord aan de onderliggende workflow wordt doorgegeven.

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

Gebruik predicaten om de aanvraag- en antwoordranden smal te houden:

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

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

Hoe het werkt

Wanneer de bovenliggende werkstroom een bericht doorstuurt naar de subwerkstroombinding:

1. Invoerlevering — de binding accepteert berichten die overeenkomen met de geaccepteerde invoertypen van de onderliggende workflow en plaatst deze in de wachtrij van de startexecutor van de onderliggende workflow.
2. Interne uitvoering — de subwerkstroom draait in dezelfde uitvoeringsomgeving binnen hetzelfde proces en heeft een eigen superstep-lus.
3. Uitvoer doorgeven — onderliggende workflow.OutputEvent-waarden worden als berichten van de binding naar stroomafwaartse bovenliggende uitvoerders verzonden en worden ook door de bovenliggende uitvoerder opgeleverd als de binding in WithOutputFrom is opgenomen.
4. Aanvraagdoorsturing — onderliggende workflow.RequestInfoEvent aanvragen worden opnieuw uitgezonden met gekwalificeerde poort-id's en kunnen via bovenliggende RequestPort bindingen worden gerouteerd.
5. Gebeurtenissen doorsturen — aangepaste gebeurtenissen van onderliggende workflows worden toegevoegd aan de bovenliggende stroom. Fouten worden getoond als bovenliggende waarden van workflow.ErrorEvent, waarbij de subworkflow-ID wordt vastgelegd.
6. Downstream-verzending : resulterende berichten worden voortgezet via de bovenliggende werkstroomranden.

De child-workflow houdt de status en de berichtroutering gescheiden van die van de parent-workflow. Berichten passeren de grens alleen via de verbindingen die gekoppeld zijn aan de binding van de subworkflow.

Nesten op meerdere niveaus

Subwerkstromen kunnen worden genest naar willekeurige diepte. Elke onderliggende workflow wordt gekoppeld voordat deze wordt toegevoegd aan de workflow waarin deze is opgenomen:

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

Opmerking

Elk nestingsniveau voegt overhead toe aan de uitvoering, omdat de onderliggende workflow een eigen superstep-lus doorloopt. Houd de nestdiepte redelijk voor prestatiegevoelige scenario's.

Foutafhandeling

Wanneer een onderliggende werkstroom een fout genereert, stuurt de subwerkstroombinding deze door naar de bovenliggende werkstroom als een workflow.ErrorEvent en stelt SubWorkflowID in op de bindings-id. De bovenliggende werkstroom kan deze fouten observeren via dezelfde gebeurtenisstroom die wordt gebruikt voor werkstroomfouten op het hoogste niveau:

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

Fouten die ontstaan tijdens het doorsturen van onderliggende gebeurtenissen, worden ook geconverteerd naar ouder-workflow.ErrorEvent-waarden met de bijgevoegde subworkflow-ID.

Controlepunten maken

Subwerkstromen ondersteunen controlepunten. Wanneer de bovenliggende werkstroom een controlepunt neemt, slaat de subwerkstroombinding het controlepuntbeheer van de onderliggende werkstroom en eventuele toewijzingen van gekwalificeerde antwoordpoorten op in de bovenliggende uitvoeringsstatus. Bij herstel kan de onderliggende workflow worden hervat met behoud van de geneste uitvoeringsstatus, inclusief openstaande aanvragen.

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)

Als een controlepunt wordt hersteld terwijl een onderliggende workflow een openstaand verzoek heeft, publiceert de herstelde bovenliggende uitvoering de gebeurtenis met gekwalificeerde informatie over het verzoek opnieuw. De beller kan een antwoord maken op basis van die opnieuw gepubliceerde aanvraag en deze terugsturen via de bovenliggende uitvoeringsgreep.

Volgende stappen 

Werkstromen als agents