Unter-Workflows

Erstellen eines Sub-Workflow

In C# verfassen Sie Unterworkflows auf zwei Arten:

• Direkte Bindung – Verwenden Sie diese Option BindAsExecutor() , um einen Workflow direkt als Executor in den übergeordneten Workflow einzubetten. Dadurch bleiben die systemeigenen Eingabe-/Ausgabetypen des Unterworkflows erhalten.
• Agent-Wrapping – Verwenden Sie AsAIAgent(), um einen Workflow in einen Agenten zu konvertieren und den Agenten dem übergeordneten Workflow hinzuzufügen. Dies ist nützlich, wenn der übergeordnete Workflow agentbasierte Executoren verwendet.

Direkte Bindung mit BindAsExecutor

Die BindAsExecutor() Erweiterungsmethode konvertiert einen Workflow in einen ExecutorBinding Workflow, der direkt zu einem übergeordneten Workflow hinzugefügt werden kann:

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

Bei diesem Vorgang BindAsExecutorwerden die eingegebenen Eingabe- und Ausgabetypen des Unterworkflows beibehalten – der übergeordnete Workflow leitet Nachrichten basierend auf den tatsächlichen Typen weiter, die der Unterworkflow erwartet und erzeugt.

Agent Wrapping mit AsAIAgent

Wenn der übergeordnete Workflow agentenbasierte Executors verwendet, konvertieren Sie den inneren Workflow mithilfe von AsAIAgent() in einen Agenten. Der WorkflowBuilder-Agent wird automatisch in einen Executor eingebettet.

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

Der innere Workflow wird als einzelner Schritt aus Sicht des übergeordneten Workflows betrachtet und ausgeführt. Der Koordinator sendet Nachrichten an die Analysepipeline, die intern ausgeführt wird specialist1 → specialist2 und leitet das Ergebnis dann an den Prüfer weiter.

Eingabe- und Ausgabetypen

Wenn ein Workflow als Unterworkflow verwendet wird, behält er die Typverträge seiner internen Executoren bei.

Mit BindAsExecutor akzeptiert der Unterworkflow-Executor dieselben Eingabetypen wie der Start-Executor des inneren Workflows und sendet die gleichen Ausgabetypen, die der innere Workflow erzeugt. Die Edges des übergeordneten Workflows müssen Ausführende verbinden, deren Ausgabetypen den erwarteten Eingabetypen des Unterworkflows entsprechen, und die Ausgabetypen des Unterworkflows müssen mit den erwarteten Eingaben der nachgeschalteten Executoren übereinstimmen.

Mit AsAIAgent wird der Unterworkflow als Agent umschlossen und folgt den Agent-Executor-Eingabe-/Ausgabeverträgen (string, ChatMessage, IEnumerable<ChatMessage>).

Ausgabeverhalten

Wenn ein Unterworkflow Ausgaben (via YieldOutputAsync) erzeugt, werden diese Ausgaben standardmäßig als Nachrichten an verbundene Executoren im übergeordneten Workflow weitergeleitet. Dies ermöglicht nachgeschalteten Executoren das Verarbeiten von Unterworkflowergebnissen.

Die ExecutorOptions Klasse steuert dieses Verhalten:

Wenn AutoYieldOutputHandlerResultObject aktiviert ist, umgehen Unterworkflowausgaben das interne Routing des übergeordneten Workflows und werden direkt an den Anrufer des übergeordneten Workflows übermittelt.

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

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

Anforderungen und Antworten

Teilworkflows unterstützen den Anforderungs- und Antwortmechanismus vollständig. Wenn ein Executor innerhalb des Unterworkflows eine Anforderung sendet (z. B. um menschliche Eingaben anzufordern), wird die WorkflowHostExecutorRequestInfoEvent an den übergeordneten Workflow mit einer qualifizierten Port-ID weitergeleitet – die ID des Subworkflow-Executors wird der Port-ID vorangestellt (z. B. SubWorkflow.GuessNumber).

Mit dieser Qualifizierung wird sichergestellt, dass die Antwort, wenn der übergeordnete Workflow eine Antwort empfängt, an die richtige Unterworkflowinstanz zurückgeleitet werden kann. Der übergeordnete Workflow verarbeitet Unterworkflowanforderungen mithilfe desselben Antwortmechanismus wie jede andere Anforderung:

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

Funktionsweise

Wenn der übergeordnete Workflow eine Nachricht an den Unterworkflow-Executor weiter leitet:

1. Eingabeübermittlung – die Nachricht wird an den Startausführer des inneren Workflows weitergeleitet. Bei BindAsExecutor muss der Nachrichtentyp mit den vom Start-Executor erwarteten Typen übereinstimmen. Mit AsAIAgent werden Nachrichten auf das ChatMessage Format normalisiert.
2. Innere Ausführung – der innere Workflow durchläuft seine eigene Superstep-Schleife.
3. Ausgabeauflistung – die Ausgabeereignisse des inneren Workflows werden erfasst. Mit BindAsExecutor behalten die Ausgaben ihre ursprünglichen Typen. Mit AsAIAgent werden Ausgaben in Nachrichten für Agentenantworten konvertiert.
4. Anforderungsweiterleitung – wenn der innere Workflow ausstehende Anforderungen hat, werden sie zur Behandlung an den übergeordneten Workflow weitergeleitet (siehe Anforderungen und Antworten).
5. Nachgeschalteter Verteiler – die resultierenden Nachrichten werden an den nächsten Executor im übergeordneten Workflow gesendet.

Da der innere Workflow seinen eigenen Ausführungskontext verwaltet, ist der Zustand unabhängig vom übergeordneten Workflow.

Mehrstufige Verschachtelung

Unterworkflows können in beliebige Tiefe geschachtelt werden. Jede Ebene behält ihren eigenen Ausführungskontext bei:

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

Fehlerbehandlung

Wenn ein Unterworkflow fehlschlägt, wird der Fehler als ein SubworkflowErrorEventFehler an den übergeordneten Workflow weitergegeben. Der übergeordnete Workflow kann diese Fehler über seinen Ereignisdatenstrom beobachten:

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

Wenn der Unterworkflow auf eine unbehandelte Ausnahme stößt, wird die Ausführung des übergeordneten Workflows fortgesetzt, der Unterworkflow-Executor beendet jedoch die Verarbeitung weiterer Nachrichten.

Erstellen von Prüfpunkten

Wenn ein Prüfpunkt für den übergeordneten Workflow übernommen wird, wird der Sitzungszustand des Sub-Workflow-Agents als Teil der Prüfpunktdaten des übergeordneten Executors serialisiert. Bei der Wiederherstellung wird der Sitzungszustand deserialisiert, wodurch der übergeordnete Workflow mit dem aktuellen Zustand des Unterworkflows fortgesetzt werden kann.

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

Erstellen eines Sub-Workflow

In Python erstellen Sie einen Unterworkflow, indem Sie ein Workflow in ein WorkflowExecutor einschließen und es einem übergeordneten Workflow hinzufügen.

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

Der innere Workflow wird als einzelner Schritt aus Sicht des übergeordneten Workflows betrachtet und ausgeführt. Der Koordinator sendet Nachrichten an die Analysepipeline, die intern ausgeführt wird specialist1 → specialist2 und leitet das Ergebnis dann an den Prüfer weiter.

WorkflowExecutor Parameter

Unter-Workflows kapseln

Umschließen Sie Workflow-Instanzen vor dem Hinzufügen zu einem übergeordneten Workflow explizit mit einem WorkflowExecutor. Agents können direkt an WorkflowBuilder übergeben werden, aber rohe Workflow-Instanzen erfordern diesen 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()
)

Explizites Umschließen ermöglicht Ihnen Folgendes:

• Weisen Sie eine bestimmte Executor-ID zur Referenzierung in mehreren Kanten zu.
• Verwenden Sie dieselbe WorkflowExecutor Instanz im gesamten Diagramm wieder.

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

Eingabe- und Ausgabetypen

Die WorkflowExecutor erbt seine Typsignatur vom eingebetteten Workflow.

• Eingabetypen entsprechen den Startausführereingabetypen des Workflows (sowie SubWorkflowResponseMessage für die Behandlung von Antworten auf weitergeleitete Anforderungen).
• Ausgabetypen entsprechen den Ausgabetypen des umschlossenen Workflows. Wenn ein Executor im Unterworkflow anforderungsantwortfähig ist, wird SubWorkflowRequestMessage auch als Ausgabetyp einbezogen.

Dies bedeutet, dass die Edges des übergeordneten Workflows Ausführende verbinden müssen, deren Ausgabetypen den erwarteten Eingabetypen des Unterworkflows entsprechen. Ebenso müssen nachgeschaltete Executoren die Typen akzeptieren, die der Unterworkflow erzeugt:

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

Ausgabeverhalten

Standardmäßig (allow_direct_output=False), wenn ein Unterworkflow Ausgaben über yield_output erzeugt, werden diese Ausgaben als Nachrichten an verbundene Executoren im übergeordneten Workflow unter Verwendung von send_message weitergeleitet. Dies ermöglicht nachgeschalteten Executoren das Verarbeiten von Unterworkflowergebnissen als Teil des übergeordneten Diagramms.

Wenn allow_direct_output=True ausgeführt wird, werden die Ausgaben des Unterworkflows direkt an den Ereignisstrom des übergeordneten Workflows übertragen. Die Ausgaben des Unterworkflows werden zu Ausgaben des übergeordneten Workflows, wobei das interne Executorrouting des übergeordneten Workflows umgangen wird.

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

Zwischenemissionen von untergeordneten Workflows

"intermediate"-Ereignisse, die in einem untergeordneten Workflow erstellt wurden, blasen sich automatisch über den Ereignisdatenstrom des übergeordneten Elements aus. Sie werden dem WorkflowExecutor eigenen id (nicht dem inneren Executor, der sie ursprünglich ausgegeben hat) zugeschrieben, was die Kapselung bewahrt. Entscheidend ist, dass diese Ereignisse die Kennzeichnung "intermediate" beibehalten, unabhängig davon, wie das übergeordnete Element die WorkflowExecutor in seinen eigenen output_from- oder intermediate_output_from-Listen kennzeichnet.

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

Anforderungen und Antworten

Teilworkflows unterstützen den Anforderungs- und Antwortmechanismus vollständig. Wenn ein Executor innerhalb eines Unterworkflows ctx.request_info() aufruft, wird die Anfrage von WorkflowExecutor abgefangen und basierend auf der Einstellung von propagate_request behandelt.

Abfangen von Requests im übergeordneten Workflow (Voreinstellung)

Mit propagate_request=False (die Standardeinstellung) werden Anforderungen aus dem Unterworkflow in einen SubWorkflowRequestMessage gewickelt und an die verbundenen Ausführende im übergeordneten Workflow gesendet. Auf diese Weise können übergeordnete Executoren die Anforderung lokal verarbeiten:

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)

Die create_response() Methode überprüft, ob der Antwortdatentyp dem erwarteten Typ aus der ursprünglichen Anforderung entspricht. Wenn die Typen nicht übereinstimmen, wird eine TypeError ausgelöst.

Weiterleiten von Anfragen an externe Anrufer

Mit propagate_request=True werden Anfragen aus dem Unterworkflow mithilfe des Standardmechanismus request_info an den Ereignisstream des übergeordneten Workflows weitergeleitet. Der Aufrufer des übergeordneten Workflows bearbeitet diese Anfragen auf dieselbe Weise wie jede andere mit menschlicher Beteiligung:

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)

Funktionsweise

Wenn der übergeordnete Workflow eine Nachricht an WorkflowExecutor leitet:

1. Eingabeübermittlung – die Nachricht wird an den Startausführer des inneren Workflows weitergeleitet. Der Nachrichtentyp muss mit den erwarteten Eingabetypen des Startausführers übereinstimmen.
2. Innere Ausführung – der innere Workflow führt eine eigene Superstepschleife bis zur Vollendung aus oder bis er externe Eingaben benötigt.
3. Ausgabeauflistung – die Ausgabeereignisse des inneren Workflows werden basierend auf der allow_direct_output Einstellung erfasst und weitergeleitet.
4. Anforderungsweiterleitung – wenn der innere Workflow ausstehende Anforderungen hat, werden sie basierend auf der propagate_request Einstellung weitergeleitet (siehe Anforderungen und Antworten).
5. Antwortakkumulation — das WorkflowExecutor sammelt Antworten und setzt den Unterworkflow erst dann fort, wenn alle erwarteten Antworten für eine bestimmte Ausführung empfangen sind.
6. Downstream-Verteiler – Ausgaben werden an den nächsten Executor im übergeordneten Workflow gesendet.

Der Unterworkflow verwaltet unabhängig vom übergeordneten Workflow seinen eigenen internen Zustand. Nachrichten werden nur über die Kanten geleitet, die WorkflowExecutor mit dem Rest des übergeordneten Diagramms verbinden. Es gibt kein Nachrichtenbroadcasting über die Ebenen hinweg.

Mehrstufige Verschachtelung

Unterworkflows können in beliebige Tiefe geschachtelt werden. Jede Ebene behält ihren eigenen Ausführungskontext bei:

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

Fehlerbehandlung

Wenn ein Unterworkflow fehlschlägt, wird der Fehler an den übergeordneten Workflow weitergegeben. Das Ereignis WorkflowExecutor erfasst das fehlgeschlagene Ereignis aus dem Unterworkflow und konvertiert es in ein Fehlerereignis im übergeordneten Kontext.

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)

Wenn der Unterworkflow auf eine unbehandelte Ausnahme stößt, empfängt der übergeordnete Workflow ein Fehlerereignis mit den Ausnahmedetails, einschließlich der ID des Unterworkflows.

Erstellen von Prüfpunkten

Teil-Workflows unterstützen Prüfpunkte. Wenn ein Prüfpunkt für den übergeordneten Workflow erstellt wird, serialisiert der WorkflowExecutor seinen internen Zustand, einschließlich des Ausführungsfortschritts des inneren Workflows und aller zwischengespeicherten Nachrichten. Bei der Wiederherstellung wird dieser Zustand deserialisiert, sodass der übergeordnete Workflow mit intakten Unterworkflow fortgesetzt werden kann.

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)

Erstellen eines Sub-Workflow

In Go erstellen Sie einen Unter-Workflow, indem Sie einen *workflow.Workflow erstellen und ihn mit inproc.BindSubworkflowAsExecutor in den übergeordneten Workflow einbinden.

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
}

Der gebundene untergeordnete Workflow wird als ein Ausführungsausführer aus der Perspektive des übergeordneten Workflows ausgeführt. Nachrichten durchlaufen die Bindung, der untergeordnete Workflow führt ein eigenes internes Diagramm aus, und die Ausgaben des untergeordneten Workflows werden an das übergeordnete Diagramm zurückgeleitet.

Eingabe- und Ausgabetypen

Die Bindung erbt das Protokoll des gekapselten Workflows. Der übergeordnete Workflow kann Nachrichten senden, deren Laufzeittypen den akzeptierten Eingabetypen des untergeordneten Workflows entsprechen, und die Bindung macht die zurückgegebenen Ausgabetypen des untergeordneten Workflows als Nachrichtentypen und Ausgabetypen verfügbar.

Dies bedeutet, dass die Kanten des übergeordneten Workflows Executoren verbinden müssen, deren Ausgabetypen den vom Unterworkflow akzeptierten Eingabetypen entsprechen, und dass nachgelagerte Executoren die vom untergeordneten Workflow erzeugten Typen verarbeiten müssen:

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

Ausgabeverhalten

Wenn ein untergeordneter Workflow eine Ausgabe zurückgibt, sendet die Unterworkflowbindung diese Ausgabe als Nachricht aus der Bindung an verbundene Executoren im übergeordneten Workflow. Wenn der übergeordnete Workflow auch die Unterworkflowbindung mit WithOutputFrommarkiert, wird derselbe Wert als übergeordnetes Element workflow.OutputEvent ausgegeben, dessen ExecutorID Unterworkflowbindungs-ID ist.

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

Benutzerdefinierte Workflow-Ereignisse, die innerhalb des untergeordneten Workflows ausgelöst werden, werden an den Ereignisstrom des übergeordneten Workflows weitergeleitet. Die eigenen Start- und Superstep-Lebenszyklusereignisse des Unterworkflows werden intern gehalten, sodass der übergeordnete Datenstrom auf extern sinnvolle Ereignisse konzentriert bleibt.

Anforderungen und Antworten

Unterworkflows unterstützen den Anforderungs- und Antwortmechanismus . Wenn ein Executor innerhalb des Sub-Workflows eine externe Anfrage stellt, qualifiziert die Sub-Workflow-Bindung die ID des Anfrageports, indem sie die Bindungs-ID voranstellt. Beispielsweise wird ein untergeordneter Anforderungsport mit dem Namen ApprovalPort im übergeordneten Workflow zu ApprovalSubWorkflow.ApprovalPort.

Um die untergeordnete Anforderung über den übergeordneten Workflow anzuzeigen, fügen Sie ein übergeordnetes Element RequestPort mit der qualifizierten ID hinzu, und leiten Sie Anforderungen und Antworten zwischen der Unterworkflowbindung und diesem Port weiter:

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

Der Aufrufer verarbeitet die Anfrage aus dem Stream des übergeordneten Workflows und sendet die Antwort über dasselbe Run-Handle zurück. Die Unterworkflowbindung entfernt das qualifizierte Präfix, bevor die Antwort an den untergeordneten Workflow geliefert wird.

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

Verwenden Sie Prädikate, um die Anforderungs- und Antwortränder schmal zu halten:

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

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

Funktionsweise

Wenn der übergeordnete Workflow eine Nachricht an die Unterworkflowbindung weiter leitet:

1. Eingabeübermittlung – die Bindung akzeptiert Nachrichten, die den akzeptierten Eingabetypen des untergeordneten Workflows entsprechen, und übergibt sie in den Startausführer des untergeordneten Workflows.
2. Interne Ausführung — der untergeordnete Workflow wird in derselben Ausführungsumgebung innerhalb des Prozesses ausgeführt und behält seine eigene Superstep-Schleife bei.
3. Ausgabeweiterleitung – untergeordnete workflow.OutputEvent Werte werden als Nachrichten von der Bindung an nachgeschaltete übergeordnete Executoren gesendet und werden auch vom übergeordneten Objekt zurückgegeben, wenn die Bindung in WithOutputFromaufgeführt ist.
4. Anforderungsweiterleitung – untergeordnete workflow.RequestInfoEvent Anforderungen werden mit qualifizierten Port-IDs erneut ausgegeben und können über übergeordnete RequestPort Bindungen weitergeleitet werden.
5. Weiterleitung von Ereignissen – benutzerdefinierte Ereignisse aus untergeordneten Workflows werden dem übergeordneten Stream hinzugefügt. Fehler werden als übergeordnete Werte workflow.ErrorEvent angezeigt, wobei die ID des Unter-Workflows erfasst wird.
6. Nachgeschaltete Weiterleitung – resultierende Nachrichten werden über die Kanten des übergeordneten Workflows weitergeleitet.

Der untergeordnete Workflow behält den Status und das Nachrichtenrouting getrennt vom übergeordneten Element bei. Nachrichten überschreiten die Grenze nur durch die Ränder, die mit der Unterworkflowbindung verbunden sind.

Mehrstufige Verschachtelung

Unterworkflows können in beliebige Tiefe geschachtelt werden. Jeder untergeordnete Workflow ist gebunden, bevor er dem Workflow hinzugefügt wird, der ihn enthält:

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

Fehlerbehandlung

Wenn ein untergeordneter Workflow einen Fehler erzeugt, leitet die Sub-Workflow-Bindung diesen als workflow.ErrorEvent an den übergeordneten Workflow weiter und setzt SubWorkflowID auf die Bindungs-ID. Der übergeordnete Workflow kann diese Fehler über denselben Ereignisdatenstrom beobachten, der für Workflowfehler der obersten Ebene verwendet wird:

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

Fehler, die beim Weiterleiten untergeordneter Ereignisse ausgelöst wurden, werden auch in übergeordnete workflow.ErrorEvent Werte mit der angefügten Unterworkflow-ID konvertiert.

Erstellen von Prüfpunkten

Teil-Workflows unterstützen Prüfpunkte. Wenn der übergeordnete Workflow einen Prüfpunkt akzeptiert, speichert die Unterworkflowbindung den Prüfpunkt-Manager des untergeordneten Workflows und alle ausstehenden qualifizierten Antwortportzuordnungen im Status des übergeordneten Ausführungsmoduls. Bei der Wiederherstellung kann der untergeordnete Workflow mit intaktem verschachteltem Ausführungszustand fortgesetzt werden, einschließlich ausstehender Anfragen.

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)

Wenn ein Checkpoint wiederhergestellt wird, während ein Child-Workflow eine ausstehende Anfrage hat, veröffentlicht die wiederhergestellte übergeordnete Ausführung das Ereignis mit den Informationen zur qualifizierten Anfrage erneut. Der Aufrufer kann auf Grundlage dieser erneut veröffentlichten Anforderung eine Antwort erstellen und sie über das Handle der übergeordneten Ausführung zurücksenden.