Podprzepływy pracy

Tworzenie Podprocesu

W języku C# utworzysz podprzepływy pracy na dwa sposoby:

• Powiązanie bezpośrednie — służy BindAsExecutor() do osadzania przepływu pracy bezpośrednio jako funkcji wykonawczej w nadrzędnym przepływie pracy. Spowoduje to zachowanie natywnych typów danych wejściowych/wyjściowych podrzędnego przepływu pracy.
• Zawijanie agentów — służy AsAIAgent() do konwertowania przepływu pracy na agenta, a następnie dodawania agenta do nadrzędnego przepływu pracy. Jest to przydatne, gdy nadrzędny przepływ pracy używa funkcji wykonawczych opartych na agencie.

Wiązanie bezpośrednie za pomocą bindAsExecutor

Metoda BindAsExecutor() rozszerzenia konwertuje przepływ pracy na element ExecutorBinding , który można dodać bezpośrednio do nadrzędnego przepływu pracy:

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

W BindAsExecutor typy wejściowe i wyjściowe podrzędnego przepływu pracy są zachowane — nadrzędny przepływ pracy kieruje komunikaty na podstawie rzeczywistych typów, które podprzepływ pracy oczekuje i generuje.

Zawijanie agentów przy użyciu AsAIAgent

Gdy nadrzędny przepływ pracy używa funkcji wykonawczych opartych na agencie, przekonwertuj wewnętrzny przepływ pracy na agenta przy użyciu polecenia AsAIAgent(). Automatycznie WorkflowBuilder opakowuje agenta w funkcji wykonawczej:

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

Wewnętrzny przepływ pracy jest uruchamiany jako jeden krok z punktu widzenia nadrzędnego przepływu pracy. Koordynator wysyła wiadomości do potoku analizy, który wewnętrznie uruchamia specialist1 → specialist2, a następnie przekazuje wynik do recenzenta.

Typy danych wejściowych i wyjściowych

Gdy przepływ pracy jest używany jako podrzędny przepływ pracy, zachowuje kontrakty typu swoich wewnętrznych funkcji wykonawczych.

W programie BindAsExecutorfunkcja wykonawcza podrzędnego przepływu pracy akceptuje te same typy danych wejściowych co funkcja wykonawcza uruchamiania wewnętrznego przepływu pracy i wysyła te same typy danych wyjściowych, które generuje wewnętrzny przepływ pracy. Krawędzie nadrzędnego przepływu pracy muszą łączyć funkcje wykonawcze, których typy wyjściowe są zgodne z oczekiwanymi typami wejściowymi podrzędnego przepływu pracy, a typy danych wyjściowych podrzędnego przepływu pracy muszą być zgodne z oczekiwanymi danymi wejściowymi podrzędnych funkcji wykonawczych.

W AsAIAgent podrzędny przepływ pracy jest opakowany jako agent i przestrzega umów wejściowych/wyjściowych Agent Executor (string, ChatMessage, IEnumerable<ChatMessage>).

Zachowanie wyjścia

Domyślnie, gdy podrzędny przepływ pracy generuje dane wyjściowe (za pośrednictwem YieldOutputAsync), te dane wyjściowe są przekazywane jako komunikaty do połączonych funkcji wykonawczych w nadrzędnym przepływie pracy. Dzięki temu podrzędne funkcje wykonawcze mogą przetwarzać wyniki podrzędnego przepływu pracy.

Klasa ExecutorOptions kontroluje to zachowanie:

Gdy AutoYieldOutputHandlerResultObject zostanie włączony, wyniki podrzędnego przepływu pracy pomijają wewnętrzny routing elementu nadrzędnego i są dostarczane bezpośrednio do obiektu wywołującego nadrzędnego przepływu pracy.

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

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

Żądania i odpowiedzi

Podprocesy workflow w pełni obsługują mechanizm żądania i odpowiedzi. Gdy element wykonawczy wewnątrz podrzędnego przepływu pracy wysyła żądanie (na przykład w celu uzyskania danych wejściowych od człowieka), WorkflowHostExecutor przekazuje RequestInfoEvent do nadrzędnego przepływu pracy z kwalifikowanym identyfikatorem portu — identyfikator elementu wykonawczego podrzędnego przepływu pracy jest dodawany do identyfikatora portu (na przykład SubWorkflow.GuessNumber).

To uprawnienie gwarantuje, że gdy nadrzędny przepływ pracy otrzyma odpowiedź, może skierować odpowiedź z powrotem do właściwego wystąpienia przepływu podrzędnego. Nadrzędny przepływ pracy obsługuje żądania podrzędnego przepływu pracy przy użyciu tego samego mechanizmu odpowiedzi co każde inne żądanie:

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

Jak to działa

Gdy nadrzędny przepływ pracy przesyła wiadomość do wykonawcy podrzędnego przepływu pracy:

1. Dostarczanie danych wejściowych — komunikat jest przekazywany do funkcji wykonawczej uruchamiania wewnętrznego przepływu pracy. W przypadku BindAsExecutor typ komunikatu musi być zgodny z oczekiwanymi typami startowego wykonawcy. W programie AsAIAgentkomunikaty są znormalizowane do formatowania ChatMessage .
2. Wykonywanie wewnętrzne — wewnętrzny przepływ pracy uruchamia własną pętlę superstep.
3. Kolekcja danych wyjściowych — zbierane są zdarzenia wyjściowe wewnętrznego przepływu pracy. W przypadku BindAsExecutorparametrów dane wyjściowe zachowują swoje oryginalne typy. Za pomocą AsAIAgent dane wyjściowe są konwertowane na wiadomości odpowiedzi agenta.
4. Przekazywanie żądań — jeśli wewnętrzny przepływ pracy ma oczekujące żądania, są one przekazywane do nadrzędnego przepływu pracy w celu obsługi (zobacz Żądania i odpowiedzi).
5. Wysyłanie podrzędne — komunikaty wynikowe są wysyłane do następnego wykonawcy w nadrzędnym przepływie pracy.

Ponieważ wewnętrzny przepływ pracy utrzymuje własny kontekst wykonywania, jego stan jest niezależny od nadrzędnego przepływu pracy.

Wielopoziomowe zagnieżdżanie

Podrzędne przepływy pracy można zagnieżdżać do dowolnej głębokości. Każdy poziom zachowuje własny kontekst wykonywania:

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

Obsługa błędów

W przypadku niepowodzenia podrzędnego przepływu pracy błąd jest propagowany do nadrzędnego przepływu pracy jako SubworkflowErrorEvent. Nadrzędny przepływ pracy może obserwować te błędy za pośrednictwem strumienia zdarzeń:

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

Jeśli podrzędny przepływ pracy napotka nieobsługiwany wyjątek, wykonanie nadrzędnego przepływu pracy będzie kontynuowane, ale funkcja wykonawcza podrzędnego przepływu pracy przestanie przetwarzać dalsze komunikaty.

Tworzenie punktów kontrolnych

Gdy punkt kontrolny jest wykonywany w przepływie pracy nadrzędnym, stan sesji agenta podrzędnego przepływu pracy jest serializowany jako część danych punktu kontrolnego wykonawcy nadrzędnego. Po przywróceniu stan sesji jest deserializowany, co umożliwia kontynuowanie nadrzędnego przepływu roboczego ze stanem podrzędnego przepływu pracy zachowanym.

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

Tworzenie Podprocesu

W języku Python tworzysz podprzepływ, opakowując Workflow w WorkflowExecutor i dodając go do nadrzędnego przepływu.

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

Wewnętrzny przepływ pracy jest uruchamiany jako jeden krok z punktu widzenia nadrzędnego przepływu pracy. Koordynator wysyła wiadomości do potoku analizy, który wewnętrznie uruchamia specialist1 → specialist2, a następnie przekazuje wynik do recenzenta.

Parametry elementu WorkflowExecutor

Zawijanie podrzędnych przepływów pracy

Jawnie opakuj instancje Workflow elementem WorkflowExecutor przed dodaniem ich do przepływu nadrzędnego. Agenty mogą być przekazywane bezpośrednio do WorkflowBuilder, ale surowe instancje Workflow wymagają tego opakowania.

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

Jawne zawijanie umożliwia:

• Przypisz określony identyfikator wykonawcy do użycia w wielu krawędziach.
• Ponownie użyj tego samego WorkflowExecutor wystąpienia na grafie.

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

Typy danych wejściowych i wyjściowych

Element WorkflowExecutor dziedziczy sygnaturę typu z opakowanego procesu:

• Typy danych wejściowych odpowiadają typom danych wejściowych startowego wykonawcy w zawiniętych przepływach pracy (plus SubWorkflowResponseMessage do obsługi odpowiedzi na przekazane żądania).
• Typy danych wyjściowych są zgodne z typami wyjściowymi opakowanych przepływów pracy. Jeśli jakikolwiek wykonawca w podprocesie jest zdolny do obsługi żądania-odpowiedzi, SubWorkflowRequestMessage jest również dołączony jako typ wyjściowy.

Oznacza to, że krawędzie nadrzędnego przepływu pracy muszą łączyć funkcje wykonawcze, których typy danych wyjściowych są zgodne z oczekiwanymi typami wejściowymi podrzędnego przepływu pracy. Podobnie podrzędne funkcje wykonawcze muszą akceptować typy generowane przez podrzędny przepływ pracy:

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

Zachowanie wyjścia

Domyślnie (allow_direct_output=False), gdy podrzędny przepływ pracy generuje dane wyjściowe za pośrednictwem yield_output, te dane wyjściowe są przekazywane jako komunikaty do połączonych wykonawców w nadrzędnym przepływie pracy przy użyciu send_message. Dzięki temu podrzędne funkcje wykonawcze przetwarzają wyniki podrzędnego przepływu pracy w ramach nadrzędnego grafu.

Gdy allow_direct_output=True wyjściowe dane podrzędnego przepływu pracy są zwracane bezpośrednio do strumienia zdarzeń nadrzędnego przepływu pracy. Dane wyjściowe podrzędnego przepływu pracy stają się danymi wyjściowymi nadrzędnego przepływu pracy, pomijając wewnętrzny routing funkcji wykonawczej elementu nadrzędnego:

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

Emisje pośrednie z podrzędnych przepływów pracy

"intermediate" zdarzenia generowane w obrębie podrzędnego przepływu pracy są automatycznie propagowane przez strumień zdarzeń nadrzędnego przepływu pracy. Są one przypisywane do WorkflowExecutorwłasnych id (nie do wewnętrznego modułu wykonawczego, który pierwotnie je emitował), co zachowuje hermetyzację. Co istotne, te zdarzenia zachowują etykietę "intermediate" niezależnie od tego, w jaki sposób element nadrzędny oznacza element WorkflowExecutor na swoich listach output_from lub 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}")

Żądania i odpowiedzi

Podprocesy workflow w pełni obsługują mechanizm żądania i odpowiedzi. Gdy wykonawca wewnątrz pod-procesu roboczego wywołuje ctx.request_info(), WorkflowExecutor przechwytuje żądanie i obsługuje je na podstawie ustawienia propagate_request.

Przechwytywanie żądań w nadrzędnym workflowu (ustawienie domyślne)

Z propagate_request=False (ustawieniem domyślnym) żądania z podrzędnego przepływu są opakowane w SubWorkflowRequestMessage i wysyłane do połączonych wykonawców w nadrzędnym przepływie. Dzięki temu nadrzędne funkcje wykonawcze mogą obsługiwać żądanie lokalnie:

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)

Metoda create_response() sprawdza, czy typ danych odpowiedzi jest zgodny z oczekiwanym typem z oryginalnego żądania. Jeśli typy nie są zgodne, zostaje zgłoszony TypeError.

Propagowanie żądań do zewnętrznych podmiotów wywołujących

Za pomocą propagate_request=True żądania z podrzędnego przepływu pracy są przekazywane do strumienia zdarzeń nadrzędnego przepływu pracy przy użyciu standardowego mechanizmu request_info. Obiekt wywołujący nadrzędnego przepływu pracy obsługuje te żądania w taki sam sposób, jak każde inne żądanie pętli typu 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)

Jak to działa

Gdy nadrzędny przepływ pracy kieruje komunikat do elementu WorkflowExecutor:

1. Dostarczanie danych wejściowych — komunikat jest przekazywany do funkcji wykonawczej uruchamiania wewnętrznego przepływu pracy. Typ komunikatu musi być zgodny z oczekiwanymi typami danych wejściowych rozruchowego wykonawcy.
2. Wykonywanie wewnętrzne — wewnętrzny przepływ pracy uruchamia własną pętlę superstep do ukończenia lub dopóki nie potrzebuje zewnętrznych danych wejściowych.
3. Kolekcja danych wyjściowych — zdarzenia wyjściowe wewnętrznego przepływu pracy są zbierane i przekazywane na podstawie ustawień allow_direct_output.
4. Przekazywanie żądań — jeśli wewnętrzny przepływ pracy ma oczekujące żądania, są one przekazywane na propagate_request podstawie ustawienia (zobacz Żądania i odpowiedzi).
5. Akumulacja odpowiedzi — WorkflowExecutor zbiera odpowiedzi i wznawia działanie podrzędnego przepływu pracy tylko wtedy, gdy odebrano wszystkie oczekiwane odpowiedzi dla danego wykonania.
6. Wysyłanie podrzędne — dane wyjściowe są wysyłane do następnego wykonawcy w nadrzędnym przepływie pracy.

Podrzędny przepływ pracy zachowuje własny stan wewnętrzny niezależnie od elementu nadrzędnego. Komunikaty są kierowane tylko przez krawędzie łączące WorkflowExecutor z resztą grafu nadrzędnego — nie ma rozgłaszania komunikatów na poziomach zagnieżdżania.

Wielopoziomowe zagnieżdżanie

Podrzędne przepływy pracy można zagnieżdżać do dowolnej głębokości. Każdy poziom zachowuje własny kontekst wykonywania:

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

Obsługa błędów

W przypadku niepowodzenia podrzędnego przepływu pracy błąd jest propagowany do nadrzędnego przepływu pracy. Element WorkflowExecutor przechwytuje zdarzenie zakończone niepowodzeniem z podrzędnego przepływu pracy i przekształca je na zdarzenie błędu w kontekście nadrzędnym.

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)

Jeśli podrzędny przepływ pracy napotka nieobsługiwany wyjątek, nadrzędny przepływ pracy odbiera zdarzenie błędu ze szczegółami wyjątku, w tym identyfikatorem podrzędnego przepływu pracy.

Tworzenie punktów kontrolnych

Podprzepływy pracy obsługują punktowanie kontrolne. Gdy punkt kontrolny jest pobierany w nadrzędnym przepływie pracy, WorkflowExecutor serializuje jego stan wewnętrzny, w tym postęp wykonywania wewnętrznego przepływu pracy i wszelkie buforowane komunikaty. Po przywróceniu ten stan jest deserializowany, co umożliwia wznowienie nadrzędnego przepływu pracy z zachowaniem integralności przepływu pracy podrzędnego.

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)

Tworzenie Podprocesu

W języku Go utworzysz pod przepływ pracy, tworząc element *workflow.Workflow i łącząc go z nadrzędnym przepływem pracy za pomocą polecenia 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
}

Powiązany podrzędny przepływ pracy jest uruchamiany jako jedna funkcja wykonawcza z perspektywy nadrzędnego przepływu pracy. Komunikaty wchodzą przez wiązanie, podrzędny przepływ pracy uruchamia własny graf wewnętrzny, a wyjścia podrzędnego przepływu pracy są kierowane z powrotem do grafu nadrzędnego.

Typy danych wejściowych i wyjściowych

Powiązanie dziedziczy protokół opakowanego przepływu pracy. Nadrzędny przepływ pracy może wysyłać komunikaty, których typy środowiska uruchomieniowego są zgodne z akceptowanymi typami wejściowymi podrzędnego przepływu pracy, a powiązanie uwidacznia zwracane typy danych wyjściowych przepływu pracy podrzędnego jako typy komunikatów i typy danych wyjściowych.

Oznacza to, że krawędzie nadrzędnego przepływu pracy muszą łączyć funkcje wykonawcze, których typy danych wyjściowych są zgodne z akceptowanymi danymi wejściowymi podrzędnego przepływu pracy, a podrzędne funkcje wykonawcze muszą obsługiwać typy zwracane przez podrzędny przepływ pracy:

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

Zachowanie wyjścia

Gdy podrzędny przepływ pracy zwraca dane wyjściowe, powiązanie podrzędnego przepływu pracy wysyła te dane wyjściowe jako komunikat z powiązania do połączonych funkcji wykonawczych w nadrzędnym przepływie pracy. Jeśli nadrzędny przepływ pracy również oznacza powiązanie podprzepływu pracy znacznikiem WithOutputFrom, ta sama wartość jest generowana jako nadrzędny element workflow.OutputEvent, którego element ExecutorID jest identyfikatorem powiązania podprzepływu pracy.

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

Niestandardowe zdarzenia przepływu pracy emitowane wewnątrz podrzędnego przepływu pracy są przekazywane do nadrzędnego strumienia zdarzeń. Zdarzenia cyklu życia początkowego i nadrzędnego przepływu pracy podrzędnego są przechowywane wewnętrznych, dzięki czemu strumień nadrzędny pozostaje skoncentrowany na zdarzeniach istotnych dla zewnątrz.

Żądania i odpowiedzi

Podprzepływy pracy obsługują mechanizm żądanie–odpowiedź. Gdy egzekutor wewnątrz podprzepływu pracy wysyła żądanie zewnętrzne, powiązanie podprzepływu pracy kwalifikuje identyfikator portu żądania, dodając na jego początku identyfikator powiązania. Na przykład port żądania podrzędnego o nazwie ApprovalPort staje się ApprovalSubWorkflow.ApprovalPort w nadrzędnym przepływie pracy.

Aby wyświetlić żądanie podrzędne za pośrednictwem nadrzędnego przepływu pracy, dodaj element nadrzędny RequestPort z kwalifikowanym identyfikatorem i żądaniami kierowania i odpowiedzi między powiązaniem podrzędnego przepływu pracy a tym portem:

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

Wywołujący obsługuje żądanie z nadrzędnego strumienia przepływu pracy i odsyła odpowiedź przez ten sam uchwyt uruchomienia. Powiązanie podprzepływu pracy usuwa prefiks kwalifikowany przed dostarczeniem odpowiedzi do podrzędnego przepływu pracy.

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

Użyj predykatów, aby zachować wąskie krawędzie żądania i odpowiedzi:

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

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

Jak to działa

Gdy nadrzędny przepływ pracy kieruje komunikat do powiązania podprzepływu:

1. Dostarczanie danych wejściowych — wiązanie akceptuje komunikaty odpowiadające typom danych wejściowych akceptowanym przez podrzędny przepływ pracy i umieszcza je w kolejce modułu wykonawczego uruchamiającego ten przepływ pracy.
2. Wewnętrzne wykonanie — podrzędny przepływ pracy jest uruchamiany w tym samym środowisku wykonawczym w obrębie procesu i zachowuje własną pętlę superstepów.
3. Przekazywanie danych wyjściowych — wartości elementu podrzędnego workflow.OutputEvent są wysyłane jako komunikaty z wiązania do wykonawców nadrzędnych znajdujących się niżej w przepływie, a także są również zwracane przez element nadrzędny, jeśli wiązanie jest wymienione w WithOutputFrom.
4. Przekazywanie żądań — żądania podrzędne workflow.RequestInfoEvent są wysyłane ponownie z użyciem kwalifikowanych identyfikatorów portów i mogą być kierowane przez powiązania nadrzędne RequestPort.
5. Przekazywanie zdarzeń — niestandardowe zdarzenia podrzędnego przepływu pracy trafiają do strumienia nadrzędnego. Błędy są prezentowane jako nadrzędne wartości workflow.ErrorEvent z zapisanym identyfikatorem podprzepływu pracy.
6. Wysyłanie podrzędne — komunikaty wynikowe są kontynuowane przez nadrzędne krawędzie przepływu pracy.

Podrzędny przepływ pracy utrzymuje stan i trasowanie komunikatów oddzielone od nadrzędnego przepływu pracy. Komunikaty przekraczają granicę tylko przez krawędzie połączone z powiązaniem podrzędnego przepływu pracy.

Wielopoziomowe zagnieżdżanie

Podrzędne przepływy pracy można zagnieżdżać do dowolnej głębokości. Każdy podrzędny przepływ pracy jest powiązany przed dodaniu go do przepływu pracy, który go zawiera:

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

Obsługa błędów

Gdy podrzędny przepływ pracy zgłasza błąd, powiązanie podprzepływu pracy przekazuje go do nadrzędnego przepływu pracy jako workflow.ErrorEvent i ustawia SubWorkflowID na identyfikator powiązania. Nadrzędny przepływ pracy może obserwować te błędy za pośrednictwem tego samego strumienia zdarzeń, który jest używany w przypadku błędów przepływu pracy najwyższego poziomu:

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

Błędy zgłaszane podczas przekazywania zdarzeń podrzędnych są również konwertowane na wartości nadrzędne workflow.ErrorEvent z dołączonym identyfikatorem podrzędnego przepływu pracy.

Tworzenie punktów kontrolnych

Podprzepływy pracy obsługują punktowanie kontrolne. Gdy nadrzędny przepływ pracy tworzy punkt kontrolny, powiązanie z podrzędnym przepływem pracy przechowuje menedżera punktów kontrolnych podrzędnego przepływu pracy oraz wszelkie oczekujące kwalifikowane mapowania portów odpowiedzi w stanie wykonawcy nadrzędnego przepływu pracy. Po przywróceniu podrzędny przepływ pracy może zostać wznowiony z zachowanym stanem zagnieżdżonego wykonania, w tym oczekujących żądań.

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)

Jeśli punkt kontrolny zostanie przywrócony, gdy podrzędny workflow ma oczekujące żądanie, przywrócony nadrzędny workflow ponownie publikuje zdarzenie z informacjami o kwalifikowanym żądaniu. Wywołujący może utworzyć odpowiedź na podstawie tego ponownie opublikowanego żądania i wysłać ją z powrotem za pośrednictwem uchwytu nadrzędnego uruchomienia.