Generatore di flussi di lavoro ed esecuzione

Un Workflow collega gli executor e gli edge in un grafo diretto e ne gestisce l'esecuzione. Coordina la chiamata dell'executor, il routing dei messaggi e lo streaming di eventi.

Creazione di flussi di lavoro

I flussi di lavoro vengono costruiti usando la WorkflowBuilder classe , che fornisce un'API Fluent per definire la struttura del flusso di lavoro:

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

I flussi di lavoro vengono costruiti usando la WorkflowBuilder classe :

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

Durante la convalida in fase di compilazione, un gestore può usare un'annotazione del contenitore bare, ad esempio list, dict, tuple o Sequence. Può accettare l'output che usa il contenitore con parametri corrispondente, ad esempio list[str].

workflow fornisce un modello di esecuzione basato su grafi in cui gli esecutori sono collegati da archi.

  • Executor : unità di elaborazione che riceve l'input e produce l'output
  • Edge : connette l'output di un executor all'input di un altro
  • Builder - Costruisce flussi di lavoro definendo esecutori e archi
  • Esecuzione : esegue un flusso di lavoro con l'input specificato

Questa funzione crea un flusso di lavoro di elaborazione del testo senza richiedere un modello o credenziali:

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) espone il risultato dell'executor inverso come output del flusso di lavoro. L'edge restituisce comunque il risultato in maiuscolo all'executor inverso indipendentemente dalla designazione dell'output.

Passare il flusso di lavoro restituito da buildWorkflow a una delle funzioni di esecuzione seguenti. Per l'input "Hello, World!", entrambi stampano Workflow output: !DLROW ,OLLEH.

Esecuzione del flusso di lavoro

I flussi di lavoro supportano le modalità di esecuzione streaming e non-streaming:

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

Usare RunStreaming quando si desiderano eventi man mano che si verificano. Passare il contesto del chiamante in modo che l'annullamento raggiunga il flusso di lavoro, controllarne l'errore al termine dell'iterazione e chiudere l'esecuzione al termine:

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 attende mentre le richieste esterne sono in sospeso. Per i flussi di lavoro che necessitano di input umano, gestire gli eventi di richiesta o usare WatchUntilHalt per tornare a una pausa. Vedi Eventi del flusso di lavoro.

Usare Run per passare all'arresto successivo e quindi esaminare gli eventi raccolti. Un'esecuzione può bloccarsi durante l'attesa dell'input esterno, non solo quando il lavoro è stato completato:

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() avanza il segnalibro ogni volta che lo chiami. Il secondo ciclo usa OutgoingEvents() per riprodurre gli eventi executor accumulati invece di tentare di consumarli nuovamente con NewEvents().

Queste funzioni chiudono le esecuzioni dopo aver osservato il risultato. Mantenere invece aperto il run se l'applicazione deve inviare più messaggi o riprenderlo dopo una pausa.

Vedere Eventi del flusso di lavoro per consumer completi, classificazione dell'output e comportamento della pausa.

Tip

Vedere gli esempi del flusso di lavoro per esempi eseguibili completi.

Convalida del flusso di lavoro

I generatori di flussi di lavoro convalidano il grafo prima dell'esecuzione. I controlli esatti dipendono dall'SDK e dai metadati del tipo disponibili:

  • Compatibilità dei tipi: controlla i protocolli di messaggio dichiarati tra gli executor connessi in cui sono disponibili i metadati.
  • Connettività a grafo: verifica che gli executor registrati siano raggiungibili dall'executor di avvio.
  • Associazione dell'executor: verifica che siano presenti registrazioni dell'executor necessarie.
  • Convalida dell'edge: verifica la presenza di connessioni non valide e edge duplicati in base alle regole del generatore.

Go Build() rifiuta associazioni non risolte, esecutori registrati non raggiungibili, archi incondizionati duplicati e tipi di messaggio dichiarati incompatibili. Le route condizionali e i protocolli dinamici richiedono comunque test di runtime.

La convalida dei tipi può creare istanze temporanee dell'executor per controllare i protocolli. Se non è possibile controllare una factory o un protocollo, tale controllo di compatibilità viene ignorato. Una compilazione riuscita non garantisce che la creazione dell'executor o ogni recapito dei messaggi abbia esito positivo in fase di esecuzione. Gestire gli errori derivanti dalla creazione dell'esecuzione, dall'iterazione degli eventi e dagli eventi di errore del flusso di lavoro.

BuildWithOptions(workflow.BuildOptions{DisableOrphanValidation: true}) disabilita solo il controllo orfano-executor. Altre convalide continuano a essere eseguite.

Modello di esecuzione: passaggi sovrapposti

Il framework usa un modello di esecuzione Pregel modificato, ovvero un approccio BSP (Bulk Synchronous Parallel) con elaborazione basata su superstep.

Funzionamento di Supersteps

L'esecuzione del flusso di lavoro è organizzata in superstep distinti. Ogni passaggio superiore:

  1. Raccoglie tutti i messaggi in sospeso dal superstep precedente
  2. Instrada i messaggi agli executor di destinazione in base alle definizioni degli edge
  3. Esegue simultaneamente tutti gli executor di destinazione all'interno del superstep
  4. Attende il completamento di tutti gli esecutori prima di avanzare (barriera di sincronizzazione)
  5. Accoda eventuali nuovi messaggi emessi dagli executor per il superstep successivo
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   │
└─────────────────┘    └─────────────────┘

Barriera di sincronizzazione

La caratteristica più importante è la barriera di sincronizzazione tra i superstep. Gli executor attivati possono essere eseguiti contemporaneamente all'interno di un superstep, ma il flusso di lavoro non passa a superstep successivo fino al completamento.

Ciò influisce sui modelli fan-out: se si esegue il fan-out su più percorsi, uno con una catena di executor e un altro con un singolo executor a esecuzione prolungata, il percorso concatenato non può avanzare finché l'executor a esecuzione prolungata non viene completato.

Perché le superstep?

Il modello BSP offre garanzie importanti:

  • Limiti dei passaggi definiti: i messaggi generati durante un superstep vengono accodati per un'elaborazione successiva.
  • Confini del checkpoint: i flussi di lavoro abilitati per il checkpoint salvano lo stato di esecuzione ai confini dei superstep.
  • Aggiornamenti dello stato memorizzati nel buffer: viene eseguito il commit delle modifiche dello stato del flusso di lavoro in coda tra i superstep.

Queste garanzie non correggono l'esecuzione o l'ordinamento degli eventi tra esecutori concorrenti all'interno di un superstep. Sincronizzare l'accesso allo stato condiviso di proprietà dell'applicazione anziché presupporre che la barriera superstep impedisca tutte le race condition.

Uso del modello superstep

Se sono necessari percorsi paralleli veramente indipendenti che non si bloccano tra loro, consolidare i passaggi sequenziali in un singolo executor. Invece di concatenare step1 → step2 → step3, combina tale logica in un unico executor. Entrambi i percorsi paralleli vengono quindi eseguiti all'interno di un unico superstep.

Passaggi successivi

Agenti nei flussi di lavoro

Argomenti correlati:

  • Ultimo aggiornamento: