Opbouwfunctie voor werkstromen en uitvoering

Een werkstroom koppelt uitvoerders en randen aan elkaar in een gerichte grafiek en beheert de uitvoering. Het coördineert aanroep van uitvoerders, berichtroutering en gebeurtenisstreaming.

Werkstromen bouwen

Werkstromen worden samengesteld met behulp van de WorkflowBuilder klasse, die een fluent API biedt voor het definiëren van de werkstroomstructuur:

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

Werkstromen worden samengesteld met behulp van de WorkflowBuilder klasse:

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

Tijdens de build-tijdvalidatie kan een handler een kale containerannotatie gebruiken, zoals list, dict, tuple of Sequence. Het kan uitvoer accepteren die gebruikmaakt van de bijbehorende geparameteriseerde container, zoals list[str].

Het workflow pakket biedt een op grafieken gebaseerd uitvoeringsmodel waarbij uitvoerders zijn verbonden door randen.

  • Uitvoerder : een verwerkingseenheid die invoer ontvangt en uitvoer produceert
  • Edge : de uitvoer van één uitvoerder verbinden met de invoer van een andere
  • Opbouwfunctie : werkstromen samenstellen door uitvoerders en randen te definiëren
  • Uitvoeren - Voert een werkstroom uit met opgegeven invoer

Met deze functie wordt een werkstroom voor tekstverwerking gebouwd zonder dat hiervoor een model of inloggegevens nodig zijn:

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) geeft het resultaat van de reverse-executor weer als werkstroomuitvoer. De edge levert nog steeds het hoofdletterresultaat aan de reverse executor, ongeacht de uitvoeraanduiding.

Geef de door buildWorkflow geretourneerde werkstroom door aan een van de onderstaande uitvoeringsfuncties. Voor de invoer "Hello, World!" geven beide Workflow output: !DLROW ,OLLEH weer.

Werkstroomuitvoering

Werkstromen ondersteunen zowel streaming- als niet-streaminguitvoeringsmodi:

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

Gebruik RunStreaming wanneer u gebeurtenissen wilt zodra ze plaatsvinden. Geef de context van de aanroeper door, zodat het annuleren de workflow bereikt, controleer de fout nadat de iteratie is beëindigd en sluit de run wanneer u klaar bent:

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 wacht terwijl externe verzoeken nog openstaan. Voor werkstromen die menselijke invoer nodig hebben, verwerkt u aanvraagevenementen of gebruikt u WatchUntilHalt om na een pauze terug te keren. Zie Workflowgebeurtenissen.

Gebruik Run dit om naar de volgende stop te gaan en vervolgens de verzamelde gebeurtenissen te inspecteren. Een run kan stoppen tijdens het wachten op externe invoer, niet alleen wanneer het werk is voltooid:

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() schuift de bladwijzer op telkens wanneer u deze aanroept. De tweede lus gebruikt OutgoingEvents() om samengevoegde executorgebeurtenissen opnieuw af te spelen in plaats van ze opnieuw te verwerken met NewEvents().

Deze functies sluiten hun runs na het observeren van het resultaat. Houd de run in plaats daarvan open als uw toepassing meer berichten moet verzenden of deze na een pauze moet hervatten.

Zie Workflowgebeurtenissen voor volledige consumenten, uitvoerclassificatie en pauzesemantiek.

Tip

Zie de werkstroomvoorbeelden voor volledige runnable voorbeelden.

Werkstroomvalidatie

Workflowbouwers valideren de graaf voordat deze wordt uitgevoerd. De exacte controles zijn afhankelijk van de SDK en de beschikbare metagegevens van het type:

  • Typecompatibiliteit: Controleert gedeclareerde berichtprotocollen tussen verbonden uitvoerders waar metagegevens beschikbaar zijn.
  • Graph-connectiviteit: controleert of geregistreerde executors bereikbaar zijn vanaf de startexecutor.
  • Executorbinding: Controleert of de vereiste executorregistraties aanwezig zijn.
  • Edge-validatie: controleert op ongeldige verbindingen en dubbele edges volgens de regels van de builder.

Go Build() weigert niet-opgeloste bindingen, onbereikbare geregistreerde executors, dubbele onvoorwaardelijke verbindingen en niet-compatibele gedeclareerde berichttypen. Voorwaardelijke routes en dynamische protocollen vereisen nog steeds runtimetests.

Typevalidatie kan tijdelijke executorinstanties maken om hun protocollen te inspecteren. Als een factory of protocol niet kan worden gecontroleerd, wordt deze compatibiliteitscontrole overgeslagen. Een geslaagde build garandeert niet dat het maken van een executor of elke berichtbezorging in runtime slaagt. Afhandelen van fouten bij het aanmaken van runs, gebeurtenisiteratie en workflowfoutgebeurtenissen.

BuildWithOptions(workflow.BuildOptions{DisableOrphanValidation: true}) schakelt alleen de orphan-executor-controle uit. Andere validatie wordt nog steeds uitgevoerd.

Uitvoeringsmodel: Supersteps

Het framework maakt gebruik van een gewijzigd Pregel-uitvoeringsmodel, een BSP-benadering (Bulk Synchronous Parallel) met verwerking op basis van superstep.

Hoe Supersteps werken

De uitvoering van de werkstroom is ingedeeld in discrete supersteps. Elke superstep:

  1. Verzamelt alle berichten die in behandeling zijn van de vorige superstap
  2. Hiermee worden berichten gerouteerd naar doelexecutors op basis van edge-definities
  3. Voert alle doelexecutors gelijktijdig uit binnen de superstep
  4. Wacht tot alle uitvoerders zijn voltooid voordat wordt doorgegaan (synchronisatiebarrière)
  5. Hiermee worden nieuwe berichten die door uitvoerders worden verzonden in de wachtrij geplaatst voor de volgende superstep
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   │
└─────────────────┘    └─────────────────┘

Synchronisatiebarrière

Het belangrijkste kenmerk is de synchronisatiebarrière tussen supersteps. Geactiveerde executors kunnen gelijktijdig worden uitgevoerd binnen een superstep, maar de workflow gaat niet verder naar de volgende superstep totdat ze zijn voltooid.

Dit is van invloed op uitwaaipatronen: als u uitwaaiert naar meerdere paden ( één met een keten van uitvoerders en een andere met één langlopende uitvoerder), kan het gekoppelde pad pas verdergaan als de langlopende uitvoerder is voltooid.

Waarom Supersteps?

Het BSP-model biedt belangrijke garanties:

  • Gedefinieerde stapgrenzen: berichten die tijdens een superstep worden verzonden, worden in de wachtrij geplaatst voor latere verwerking.
  • Checkpointgrenzen: Werkstromen met controlepunten slaan de uitvoeringsstatus op bij de grenzen van supersteps.
  • Gebufferde statusupdates: wijzigingen in de werkstroomstatus in de wachtrij worden doorgevoerd tussen supersteps.

Deze garanties bieden geen vaste uitvoerings- of gebeurtenisvolgorde tussen gelijktijdige uitvoerders binnen één superstap. Synchroniseer de toegang tot de gedeelde status van de toepassing in plaats van ervan uit te gaan dat de superstepbarrière alle races voorkomt.

Werken met het Superstep-model

Als u echt onafhankelijke parallelle paden nodig hebt die elkaar niet blokkeren, voegt u opeenvolgende stappen samen in één uitvoerprogramma. In plaats van de logica te koppelen step1 → step2 → step3, combineert u die logica in één uitvoerder. Beide parallelle paden worden vervolgens uitgevoerd binnen één superstep.

Volgende stappen 

Agents in werkprocessen

Verwante onderwerpen:

  • Laatst bijgewerkt om