Construção e Execução de Fluxo de Trabalho

Um fluxo de trabalho vincula executores e bordas em um grafo direcionado e gerencia a execução. Ele coordena a invocação do executor, o roteamento de mensagens e o streaming de eventos.

Criando fluxos de trabalho

Os fluxos de trabalho são construídos usando a WorkflowBuilder classe, que fornece uma API fluente para definir a estrutura de fluxo de trabalho:

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

Fluxos de trabalho são construídos usando a 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 a validação em tempo de build, um manipulador pode usar uma anotação simples de contêiner, como list, tuple, dict ou Sequence. Ele pode aceitar a saída que usa o contêiner parametrizado correspondente, como list[str].

O workflow pacote fornece um modelo de execução baseado em grafo em que os executores são conectados por bordas.

  • Executor – Uma unidade de processamento que recebe entrada e produz saída
  • Borda – Conecta a saída de um executor à entrada de outro
  • Construtor – Constrói fluxos de trabalho definindo executores e bordas
  • Executar – Executa um fluxo de trabalho com determinada entrada

Essa função cria um fluxo de trabalho de processamento de texto sem exigir um modelo ou credenciais:

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) expõe o resultado do executor reverso como saída de fluxo de trabalho. A borda ainda fornece o resultado em letras maiúsculas para o executor reverso, independentemente da designação de saída.

Passe o fluxo de trabalho retornado por buildWorkflow para qualquer uma das funções de execução abaixo. Para a entrada "Hello, World!", ambos imprimem Workflow output: !DLROW ,OLLEH.

Execução de fluxo de trabalho

Os fluxos de trabalho dão suporte a modos de execução de streaming e não 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()}")

Use RunStreaming quando quiser que os eventos ocorram. Passe o contexto do chamador para que o cancelamento atinja o workflow, verifique o erro após o término da iteração e feche o run quando terminar:

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 aguarda enquanto as solicitações externas estão pendentes. Para fluxos de trabalho que precisam de entrada humana, manipule eventos de solicitação ou use WatchUntilHalt para retornar em um ponto de pausa. Consulte eventos de fluxo de trabalho.

Use Run para avançar para a próxima parada e, em seguida, inspecionar os eventos coletados. Uma execução pode parar enquanto aguarda a entrada externa, não apenas quando seu trabalho é concluído:

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() avança seu marcador cada vez que você o chama. O segundo loop usa OutgoingEvents() para reproduzir eventos acumulados do executor em vez de tentar consumi-los novamente com NewEvents().

Essas funções encerram suas execuções depois de observar o resultado. Mantenha a execução aberta se o aplicativo precisar enviar mais mensagens ou retomá-la após uma pausa.

Consulte eventos de fluxo de trabalho para consumidores completos, classificação de saída e semântica de pausa.

Tip

Consulte os exemplos de fluxo de trabalho para obter exemplos executáveis completos.

Validação de fluxo de trabalho

Os construtores de fluxo de trabalho validam o grafo antes da execução. As verificações exatas dependem do SDK e dos metadados de tipo disponíveis:

  • Compatibilidade de tipos: verifica os protocolos de mensagem declarados entre executores conectados em que os metadados estão disponíveis.
  • Conectividade do grafo: verifica se os executores registrados podem ser acessados no executor inicial.
  • Associação do executor: verifica se os registros de executor necessários estão presentes.
  • Validação de borda: verifica se há conexões inválidas e bordas duplicadas de acordo com as regras do construtor.

O Go Build() rejeita associações não resolvidas, executores registrados inacessíveis, arestas incondicionais duplicadas e tipos de mensagens declaradas incompatíveis. Rotas condicionais e protocolos dinâmicos ainda exigem testes de runtime.

A validação de tipo pode criar instâncias temporárias de executor para inspecionar seus protocolos. Se uma fábrica ou protocolo não puder ser inspecionado, essa verificação de compatibilidade é ignorada. Um build bem-sucedido não garante que a criação do executor ou cada entrega de mensagens seja bem-sucedida em tempo de execução. Trate erros de criação de execução, iteração de evento e eventos de falha de fluxo de trabalho.

BuildWithOptions(workflow.BuildOptions{DisableOrphanValidation: true}) desabilita apenas a verificação de executor órfão. Outra validação ainda é executada.

Modelo de execução: Supersteps

A estrutura usa um modelo de execução de Pregel modificado – uma abordagem BSP (Bulk Synchronous Parallel) com processamento baseado em superstep.

Como funcionam os Supersteps

A execução do fluxo de trabalho é organizada em superpassos discretos. Cada superstep:

  1. Coleta todas as mensagens pendentes do superpasso anterior
  2. Roteia mensagens para executores de destino com base em definições de borda
  3. Executa todos os executores de destino simultaneamente dentro do superstep
  4. Aguarda que todos os executores sejam concluídos antes de avançar (barreira de sincronização)
  5. Enfileira todas as novas mensagens emitidas pelos executores para a próxima superetapa
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   │
└─────────────────┘    └─────────────────┘

Barreira de sincronização

A característica mais importante é a barreira de sincronização entre superetapas. Os executores disparados podem ser executados simultaneamente em um superstep, mas o fluxo de trabalho não avança para o próximo superstep até que eles terminem.

Isso afeta os padrões de fan-out: se você usar vários caminhos, um com uma cadeia de executores e outro com um único executor de longa execução, o caminho encadeado não poderá avançar até que o executor de longa execução seja concluído.

Por que supersteps?

O modelo BSP fornece garantias importantes:

  • Limites de etapa definidos: as mensagens emitidas durante um superstep são enfileiradas para processamento posterior.
  • Limites de checkpoint: fluxos de trabalho habilitados para checkpoint salvam o estado de execução nos limites de superetapa.
  • Atualizações de estado em buffer: as alterações de estado do fluxo de trabalho enfileiradas são confirmadas entre superetapas.

Essas garantias não garantem a execução nem a ordenação de eventos entre executores simultâneos em uma superetapa. Sincronize o acesso ao estado compartilhado de propriedade do aplicativo em vez de assumir que a barreira de superetapa impede todas as condições de corrida.

Trabalhando com o Modelo Superstep

Se você precisar de caminhos paralelos verdadeiramente independentes que não se bloqueiem, consolide as etapas sequenciais em um único executor. Em vez de encadear step1 → step2 → step3, combine essa lógica em um executor. Os dois caminhos paralelos então são executados em uma única superetapa.

Próximas Etapas 

Agentes em fluxos de trabalho

Tópicos relacionados:

  • Última atualização em