Generador de flujos de trabajo y ejecución

Los flujos de trabajo se construyen mediante la WorkflowBuilder clase , que proporciona una API fluida para definir la estructura del flujo de trabajo:

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

Los flujos de trabajo se construyen mediante la WorkflowBuilder clase :

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 validación en tiempo de compilación, un controlador puede usar una anotación de contenedor simple, como list, dict, tuple o Sequence. Puede aceptar la salida que usa el contenedor parametrizado correspondiente, como list[str].

El workflow paquete proporciona un modelo de ejecución basado en grafos donde los ejecutores están conectados por bordes.

• Ejecutor : una unidad de procesamiento que recibe la entrada y genera la salida.
• Edge : conecta la salida de un ejecutor a la entrada de otra.
• Generador : construye flujos de trabajo mediante la definición de ejecutores y bordes
• Ejecutar : ejecuta un flujo de trabajo con una entrada determinada.

Esta función compila un flujo de trabajo de procesamiento de texto sin necesidad de un modelo o credenciales:

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) expone el resultado del ejecutor inverso como salida del flujo de trabajo. El borde sigue entregando el resultado en mayúsculas al ejecutor inverso independientemente de la designación de la salida.

Pase el flujo de trabajo devuelto por buildWorkflow a cualquiera de las funciones de ejecución siguientes. Para la entrada "Hello, World!", ambos imprimen Workflow output: !DLROW ,OLLEH.

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

Utilice RunStreaming cuando quiera recibir los eventos a medida que se produzcan. Pase el contexto del llamante para que la cancelación llegue al flujo de trabajo, compruebe su error después de que finalice la iteración y cierre la ejecución cuando haya terminado:

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 espera mientras las solicitudes externas están pendientes. Para los flujos de trabajo que necesitan la entrada humana, gestione los eventos de solicitud o use WatchUntilHalt para devolver en una pausa. Consulte Eventos de flujo de trabajo.

Use Run para avanzar hasta la siguiente detención y, a continuación, examine los eventos recopilados. Una ejecución puede quedar en pausa mientras se espera la entrada externa, no solo cuando se completa su trabajo:

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 su marcador cada vez que lo llama. El segundo bucle usa OutgoingEvents() para reproducir eventos de ejecutor acumulados en lugar de intentar consumirlos de nuevo con NewEvents().

Estas funciones cierran sus ejecuciones después de observar el resultado. Mantenga la ejecución abierta en lugar si su aplicación necesita enviar más mensajes o reanudarla después de una pausa.

Vea Eventos de flujo de trabajo para los consumidores completos, la clasificación de la salida y la semántica de las pausas.

Go Build() rechaza enlaces no resueltos, ejecutores registrados inaccesibles, bordes incondicionales duplicados y tipos de mensajes declarados incompatibles. Las rutas condicionales y los protocolos dinámicos siguen necesitando pruebas en tiempo de ejecución.

La validación de tipos puede crear instancias temporales del ejecutor para inspeccionar sus protocolos. Si no se puede inspeccionar una fábrica o protocolo, se omite esa comprobación de compatibilidad. Una compilación correcta no garantiza que la creación del ejecutor o cada entrega de mensajes se realice correctamente en tiempo de ejecución. Controle los errores de la creación de la ejecución, la iteración de eventos y los eventos de error de flujo de trabajo.

BuildWithOptions(workflow.BuildOptions{DisableOrphanValidation: true}) deshabilita solo la comprobación de orphan-executor. Se siguen realizando otras validaciones.