Arbetsflödesbyggare och utförande

Ett arbetsflöde kopplar exekutorer och kanter till en riktad graf och hanterar flödet. Den samordnar anrop av exekutorer, meddelanderoutning och händelseströmning.

Skapa arbetsflöden

Arbetsflöden skapas med hjälp av WorkflowBuilder klassen, vilket ger ett flytande API för att definiera arbetsflödesstrukturen:

using Microsoft.Agents.AI.Workflows;

var processor = new DataProcessor();
var validator = new Validator();
var formatter = new Formatter();

// Build workflow
WorkflowBuilder builder = new(processor); // Set starting executor
builder.AddEdge(processor, validator);
builder.AddEdge(validator, formatter);
var workflow = builder.Build();

Arbetsflöden skapas med hjälp av WorkflowBuilder klassen:

from agent_framework import WorkflowBuilder

processor = DataProcessor()
validator = Validator()
formatter = Formatter()

# Build workflow
builder = WorkflowBuilder(start_executor=processor)
builder.add_edge(processor, validator)
builder.add_edge(validator, formatter)
workflow = builder.build()

Under validering av byggtid kan en hanterare använda en enbart containerannotering, till exempel list, dict, tuple eller Sequence. Den kan acceptera utdata som använder motsvarande parametriserade container, till exempel list[str].

Paketet workflow innehåller en grafbaserad körningsmodell där utförare är anslutna via kanter.

Executor – en bearbetningsenhet som tar emot indata och genererar utdata
Edge – Ansluter utdata från en köre till indata från en annan
Builder – Bygger arbetsflöden genom att definiera exekverare och kanter
Kör – kör ett arbetsflöde med angivna indata

Den här funktionen skapar ett arbetsflöde för textbearbetning utan att en modell eller autentiseringsuppgifter krävs:

import (
    "slices"
    "strings"

    "github.com/microsoft/agent-framework-go/workflow"
)

func buildWorkflow() (*workflow.Workflow, error) {
    uppercase := workflow.NewExecutor(
        "UppercaseExecutor", strings.ToUpper,
    ).Bind()

    reverse := workflow.NewExecutor(
        "ReverseExecutor", func(input string) string {
            runes := []rune(input)
            slices.Reverse(runes)
            return string(runes)
        },
    ).Bind()

    return workflow.NewBuilder(uppercase).
        AddEdge(uppercase, reverse).
        WithOutputFrom(reverse).
        Build()
}

WithOutputFrom(reverse) visar den omvända exekverarens resultat som arbetsflödesutdata. edge-noden levererar fortfarande versalresultatet till den omvända exekveraren oavsett utdatabeteckning.

Skicka arbetsflödet som returneras av buildWorkflow till någon av körningsfunktionerna nedan. För indata "Hello, World!"skriver båda ut Workflow output: !DLROW ,OLLEH.

Arbetsflödeskörning

Arbetsflöden stöder både körningslägen för direktuppspelning och icke-direktuppspelning:

using Microsoft.Agents.AI.Workflows;

// Streaming execution — get events as they happen
StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, inputMessage);
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    if (evt is ExecutorCompletedEvent executorComplete)
    {
        Console.WriteLine($"{executorComplete.ExecutorId}: {executorComplete.Data}");
    }

    if (evt is WorkflowOutputEvent outputEvt)
    {
        Console.WriteLine($"Workflow completed: {outputEvt.Data}");
    }
}

// Non-streaming execution — wait for completion
Run result = await InProcessExecution.RunAsync(workflow, inputMessage);
foreach (WorkflowEvent evt in result.NewEvents)
{
    if (evt is WorkflowOutputEvent outputEvt)
    {
        Console.WriteLine($"Final result: {outputEvt.Data}");
    }
}

# Streaming execution — get events as they happen
async for event in workflow.run(input_message, stream=True):
    if event.type == "output":
        print(f"Workflow completed: {event.data}")

# Non-streaming execution — wait for completion
events = await workflow.run(input_message)
print(f"Final result: {events.get_outputs()}")

Använd RunStreaming när du vill ha händelser när de inträffar. Skicka uppringarens kontext så att avbrottet når arbetsflödet, kontrollera dess fel när iterationen är slut och stäng körningen när du är klar:

import (
    "context"
    "errors"
    "fmt"

    "github.com/microsoft/agent-framework-go/workflow"
    "github.com/microsoft/agent-framework-go/workflow/inproc"
)

func runStreaming(
    ctx context.Context, wf *workflow.Workflow,
) (err error) {
    stream, err := inproc.Default.RunStreaming(ctx, wf, "Hello, World!")
    if err != nil {
        return err
    }
    defer func() {
        err = errors.Join(err, stream.Close(context.WithoutCancel(ctx)))
    }()

    for evt, streamErr := range stream.WatchStream(ctx) {
        if streamErr != nil {
            return streamErr
        }
        switch event := evt.(type) {
        case workflow.OutputEvent:
            fmt.Printf("Workflow output: %v\n", event.Output)
        case workflow.ErrorEvent:
            return event.Error
        case workflow.ExecutorFailedEvent:
            return fmt.Errorf("executor %q failed: %w",
                event.ExecutorID, event.Error)
        }
    }
    return ctx.Err()
}

WatchStream väntar medan externa begäranden är väntande. För arbetsflöden som behöver mänsklig indata, hantera begärandehändelser eller använd WatchUntilHalt för att återgå vid en paus. Se Arbetsflödeshändelser.

Använd Run för att gå vidare till nästa stopp och inspektera sedan de insamlade händelserna. En körning kan stanna i väntan på externa indata, inte bara när dess arbete är klart:

import (
    "context"
    "errors"
    "fmt"

    "github.com/microsoft/agent-framework-go/workflow"
    "github.com/microsoft/agent-framework-go/workflow/inproc"
)

func runToHalt(
    ctx context.Context, wf *workflow.Workflow,
) (err error) {
    run, err := inproc.Default.Run(ctx, wf, "Hello, World!")
    if err != nil {
        return err
    }
    defer func() {
        err = errors.Join(err, run.Close(context.WithoutCancel(ctx)))
    }()

    for evt := range run.NewEvents() {
        switch event := evt.(type) {
        case workflow.OutputEvent:
            fmt.Printf("Workflow output: %v\n", event.Output)
        case workflow.ErrorEvent:
            return event.Error
        case workflow.ExecutorFailedEvent:
            return fmt.Errorf("executor %q failed: %w",
                event.ExecutorID, event.Error)
        }
    }

    for evt := range run.OutgoingEvents() {
        if evt, ok := evt.(workflow.ExecutorCompletedEvent); ok {
            fmt.Printf("%s: %v\n", evt.ExecutorID, evt.Result)
        }
    }
    return ctx.Err()
}

NewEvents() flyttar fram bokmärket varje gång du anropar det. Den andra loopen använder OutgoingEvents() för att spela upp ackumulerade händelser från exekveraren i stället för att försöka konsumera dem igen med NewEvents().

Dessa funktioner avslutar sina körningar efter att ha kontrollerat resultatet. Håll körningen öppen i stället om programmet behöver skicka fler meddelanden eller återuppta den efter en paus.

Se Arbetsflödeshändelser för slutförande konsumenter, utdataklassificering och semantik för paus.

Tip

Se arbetsflödesexemplen för fullständiga körbara exempel.

Arbetsflödesverifiering

Arbetsflödesbyggare validerar grafen före körning. De exakta kontrollerna beror på SDK och tillgängliga typmetadata:

Typkompatibilitet: Kontrollerar deklarerade meddelandeprotokoll mellan anslutna exekverare där metadata är tillgängliga.
Grafanslutning: Kontrollerar att registrerade exekverare kan nås från startexekveraren.
Executorbindning: Kontrollerar att nödvändiga executorregistreringar finns.
Kantvalidering: Söker efter ogiltiga anslutningar och duplicerade kanter enligt byggverktygets regler.

Go Build() avvisar olösta bindningar, oåtkomliga registrerade exekverare, duplicerade ovillkorliga kanter och inkompatibla deklarerade meddelandetyper. Villkorsstyrda rutter och dynamiska protokoll kräver fortfarande testning under körning.

Typvalidering kan skapa tillfälliga exekverarinstanser för att inspektera deras gränssnittskontrakt. Om en fabrik eller ett protokoll inte kan inspekteras hoppas kompatibilitetskontrollen över. Ett lyckat bygge garanterar inte att exekutorskapandet eller varje meddelandeleverans lyckas vid körning. Hantera fel från skapande av körning, iterering av händelser och arbetsflödesfelhändelser.

BuildWithOptions(workflow.BuildOptions{DisableOrphanValidation: true}) inaktiverar endast kontrollen av orphan-executor. Övrig validering körs fortfarande.

Körningsmodell: Supersteg

Ramverket använder en modifierad Pregel-körningsmodell – en Bulk Synchronous Parallel (BSP)-ansats med bearbetning baserad på supersteg.

Så här fungerar Supersteps

Arbetsflödeskörning ordnas i diskreta supersteg. Varje supersteg:

Samlar in alla väntande meddelanden från föregående supersteg
Dirigerar meddelanden till målexekutorer baserat på gränsdefinitioner
Kör alla målexekutorer samtidigt i supersteget
Väntar på att alla exekutörer har slutfört sitt arbete innan de avancerar (synkroniseringsbarriär)
Lägger i kö alla nya meddelanden som genereras av exekutorer för nästa supersteg

Superstep N:
┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│  Collect All    │───▶│  Route Messages │───▶│  Execute All    │
│  Pending        │    │  Based on Type  │    │  Target         │
│  Messages       │    │  & Conditions   │    │  Executors      │
└─────────────────┘    └─────────────────┘    └─────────────────┘
                                                       │
                                                       │ (barrier: wait for all)
┌─────────────────┐    ┌─────────────────┐             │
│  Start Next     │◀───│  Emit Events &  │◀────────────┘
│  Superstep      │    │  New Messages   │
└─────────────────┘    └─────────────────┘

Synkroniseringsbarriär

Den viktigaste egenskapen är synkroniseringsbarriären mellan supersteg. Utlösta exekutorer kan köras samtidigt i ett supersteg, men arbetsflödet går inte vidare till nästa supersteg förrän de har slutförts.

Detta påverkar förgreningsmönster: om du delar upp till flera sökvägar – en med en kedja av exekutorer och en annan med en enda långvarig exekutor – kan den kedjade sökvägen inte avancera förrän den långvariga exekutorn har slutförts.

Varför Supersteps?

BSP-modellen ger viktiga garantier:

Definierade steggränser: Meddelanden som genereras under ett supersteg placeras i kö för senare bearbetning.
Kontrollpunktsgränser: Kontrollpunktsaktiverade arbetsflöden sparar körningstillstånd vid gränser för supersteg.
Uppdateringar av buffrat tillstånd: Ändringar i arbetsflödets tillstånd i kö verkställs mellan supersteg.

Dessa garantier åtgärdar inte exekvering eller händelseordning bland samtidiga exekutorer inom ett supersteg. Synkronisera åtkomst till programägt delat tillstånd i stället för att anta att superstegsbarriären förhindrar alla race conditions.

Arbeta med Superstep-modellen

Om du behöver verkligt oberoende parallella sökvägar som inte blockerar varandra, konsoliderar du sekventiella steg till en enda utförare. I stället för att länka step1 → step2 → step3 bör du kombinera den logiken i en enskild exekverare. Båda parallella vägarna körs sedan inom ett enda supersteg.

Nästa steg

Agenter i arbetsflöden

Relaterade ämnen:

Utförare – bearbetningsenheter i ett arbetsflöde
Kanter – anslutningar mellan exekutorer
Händelser – arbetsflödesobservabilitet
Tillståndshantering

Feedback

Var den här sidan till hjälp?

Uppdaterades senast 2026-10-29