Criador e Execução de Fluxos de Trabalho

Um fluxo de trabalho liga executores e arestas num grafo direcionado e gere a execução. Coordena a invocação do executor, o encaminhamento 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 do 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();

Os 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 construção, um processador pode usar uma anotação de contentor nua, como list, dict, tuple ou Sequence. Pode aceitar saídas que utilizam o contentor parametrizado correspondente, como list[str].

O workflow pacote fornece um modelo de execução baseado em grafos onde os executores estão ligados por arestas.

  • Executor - Uma unidade de processamento que recebe entrada e produz saída
  • Edge - Liga a saída de um executor à entrada de outro
  • Builder - Constrói fluxos de trabalho definindo executores e arestas
  • Run - Executa um fluxo de trabalho com a entrada dada

Esta função constrói um fluxo de trabalho de processamento de texto sem precisar de 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 do fluxo de trabalho. O edge continua a entregar o resultado em maiúsculas ao executor de reversão, independentemente da designação de saída.

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

Execução do fluxo de trabalho

Os fluxos de trabalho suportam tanto os modos de execução em fluxo contínuo como os modos não em fluxo contínuo.

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

Usa RunStreaming quando quiseres acompanhar os eventos à medida que acontecem. Passe o contexto do chamador para que o cancelamento chegue ao fluxo de trabalho, verifique o erro após a iteração terminar e feche a execução 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 fica em espera enquanto pedidos externos estão pendentes. Para fluxos de trabalho que requerem intervenção humana, trate eventos de solicitação ou use WatchUntilHalt para retornar em pausa. Consulte Eventos de fluxo de trabalho.

Utilize Run para avançar até à próxima paragem e depois inspecionar os eventos recolhidos. Uma execução pode parar enquanto espera por entrada externa, não apenas quando o seu trabalho está 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 o marcador cada vez que o chamas. O segundo ciclo usa OutgoingEvents() para reproduzir eventos do executor acumulados em vez de tentar consumi-los novamente com NewEvents().

Estas funções encerram as suas execuções após observarem o resultado. Mantém a execução aberta caso a tua aplicação precise de enviar mais mensagens ou retome-a após uma pausa.

Veja Eventos de fluxo de trabalho para consumidores de conclusão, classificação de saída e semântica de pausa.

Tip

Consulte os exemplos de workflow para amostras completas executáveis.

Validação do fluxo de trabalho

Os construtores de workflow validam o gráfico antes da execução. As verificações exatas dependem do SDK e dos metadados de tipo disponíveis:

  • Compatibilidade de tipo: Verifica protocolos de mensagens declarados entre executores ligados onde os metadados estão disponíveis.
  • Conectividade de grafos: Verifica se os executores registados são alcançáveis a partir do executor inicial.
  • Vinculação do executor: Verifica se os registos necessários do executor estão presentes.
  • Validação de arestas: Verifica ligações inválidas e arestas duplicadas de acordo com as regras do construtor.

O Go Build() rejeita associações não resolvidas, executores registados inalcançáveis, arestas duplicadas incondicionais e tipos de mensagens declaradas incompatíveis. Rotas condicionais e protocolos dinâmicos ainda requerem testes em tempo de execução.

A validação de tipos pode criar instâncias executoras temporárias para inspecionar os seus protocolos. Se uma fábrica ou protocolo não puder ser inspecionado, essa verificação de compatibilidade é ignorada. Uma compilação bem-sucedida não garante que a criação do executor ou todas as entregas de mensagens tenham sucesso em tempo de execução. Lidar com erros de criação de uma execução, iteração de eventos e eventos de falha do workflow.

BuildWithOptions(workflow.BuildOptions{DisableOrphanValidation: true}) desativa apenas a verificação orphan-executor. Outras validações continuam a funcionar.

Modelo de Execução: Supersteps

O framework utiliza um modelo de execução Pregel modificado — uma abordagem Bulk Synchronous Parallel (BSP) com processamento baseado em superpassos.

Como Funcionam os Supersteps

A execução do fluxo de trabalho está organizada em superpassos discretos. Cada superpasso:

  1. Recolhe todas as mensagens pendentes do superstep anterior
  2. Encaminha mensagens para executores alvo com base em definições de arestas
  3. Executa todos os executores alvo de forma concorrente dentro do superpasso
  4. Espera que todos os executores completem antes de avançar (barreira de sincronização)
  5. Coloca em fila quaisquer novas mensagens emitidas pelos executores para o próximo 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   │
└─────────────────┘    └─────────────────┘

Barreira de Sincronização

A característica mais importante é a barreira de sincronização entre superetapas. Executores desencadeados podem correr em simultâneo dentro de um superpasso, mas o fluxo de trabalho não avança para o próximo superpasso até terminarem.

Isto afeta os padrões de dispersão: se se espalhar por múltiplos caminhos — um com uma cadeia de executores e outro com um único executor de longa duração — o caminho encadeado não pode avançar até que o executor de longa duração esteja concluído.

Porquê Supersteps?

O modelo BSP fornece garantias importantes:

  • Limites definidos dos passos: As mensagens emitidas durante um superpasso são colocadas na fila para processamento posterior.
  • Fronteiras de checkpoint: Fluxos de trabalho com checkpoints ativados guardam o estado de execução nas fronteiras de checkpoint de cada superstep.
  • Atualizações de estado em buffer: As alterações de estado do fluxo de trabalho em fila são aplicadas entre superpassos.

Estas garantias não corrigem a execução ou a ordenação de eventos entre executores concorrentes num único superpasso. Sincronizar o acesso ao estado partilhado propriedade da aplicação em vez de assumir que a barreira superstep impede todas as corridas.

Trabalhar com o Modelo Superstep

Se precisares de caminhos paralelos verdadeiramente independentes que não se bloqueiem mutuamente, consolida passos sequenciais num único executor. Em vez de encadear step1 → step2 → step3, combine essa lógica num único executor. Ambos os caminhos paralelos executam-se então dentro de um único superpasso.

Passos seguintes

Agentes em Fluxos de Trabalho

Tópicos relacionados:

  • Última atualização a