Compartilhar via

Criar um fluxo de trabalho sequencial simples

Este tutorial demonstra como criar um fluxo de trabalho sequencial simples usando fluxos de trabalho do Agent Framework.

Fluxos de trabalho sequenciais são a base da criação de sistemas complexos de agente de IA. Este tutorial mostra como criar um fluxo de trabalho simples de duas etapas em que cada etapa processa dados e os passa para a próxima etapa.

Visão geral

Neste tutorial, você criará um fluxo de trabalho com dois executores:

  1. Executor de Maiúsculas – Converte o texto de entrada para maiúsculas
  2. Executor de Texto Inverso – Inverte o texto e gera o resultado final

O fluxo de trabalho demonstra os principais conceitos como:

  • Criando um executor personalizado com um manipulador
  • Criando um executor personalizado a partir de uma função
  • Usar WorkflowBuilder para conectar executores com bordas
  • Processamento de dados por meio de etapas sequenciais
  • Observando a execução do fluxo de trabalho por meio de eventos

Conceitos abordados

Pré-requisitos

Implementação passo a passo

As seções a seguir mostram como criar o fluxo de trabalho sequencial passo a passo.

Etapa 1: Instalar pacotes NuGet

Primeiro, instale os pacotes necessários para seu projeto do .NET:

dotnet add package Microsoft.Agents.AI.Workflows --prerelease

Etapa 2: Definir o Executor em Letras Maiúsculas

Defina um executor que converte o texto em letras maiúsculas:

using System;
using System.Linq;
using System.Threading.Tasks;
using Microsoft.Agents.AI.Workflows;

/// <summary>
/// First executor: converts input text to uppercase.
/// </summary>
Func<string, string> uppercaseFunc = s => s.ToUpperInvariant();
var uppercase = uppercaseFunc.BindExecutor("UppercaseExecutor");

Pontos-chave:

  • Criar uma função que usa uma cadeia de caracteres e retorna a versão maiúscula
  • Usar BindExecutor() para criar um executor com base na função

Etapa 3: Definir o Executor de Texto Inverso

Defina um executor que inverte o texto:

/// <summary>
/// Second executor: reverses the input text and completes the workflow.
/// </summary>
internal sealed class ReverseTextExecutor() : Executor<string, string>("ReverseTextExecutor")
{
    public override ValueTask<string> HandleAsync(string input, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        // Reverse the input text
        return ValueTask.FromResult(new string(input.Reverse().ToArray()));
    }
}

ReverseTextExecutor reverse = new();

Pontos-chave:

  • Criar uma classe que herda de Executor<TInput, TOutput>
  • Implementar HandleAsync() para processar a entrada e retornar a saída

Etapa 4: Criar e conectar o fluxo de trabalho

Conecte os executores usando WorkflowBuilder:

// Build the workflow by connecting executors sequentially
WorkflowBuilder builder = new(uppercase);
builder.AddEdge(uppercase, reverse).WithOutputFrom(reverse);
var workflow = builder.Build();

Pontos-chave:

  • WorkflowBuilder construtor usa o executor inicial
  • AddEdge() cria uma conexão direcionada de maiúsculas para reverso
  • WithOutputFrom() especifica quais executores produzem saídas de fluxo de trabalho
  • Build() cria o fluxo de trabalho imutável

Etapa 5: Executar o fluxo de trabalho

Execute o fluxo de trabalho e observe os resultados:

// Execute the workflow with input data
await using Run run = await InProcessExecution.RunAsync(workflow, "Hello, World!");
foreach (WorkflowEvent evt in run.NewEvents)
{
    switch (evt)
    {
        case ExecutorCompletedEvent executorComplete:
            Console.WriteLine($"{executorComplete.ExecutorId}: {executorComplete.Data}");
            break;
    }
}

Etapa 6: Noções básicas sobre a saída do fluxo de trabalho

Ao executar o fluxo de trabalho, você verá uma saída como:

UppercaseExecutor: HELLO, WORLD!
ReverseTextExecutor: !DLROW ,OLLEH

A entrada "Olá, Mundo!" é convertida primeiro em maiúscula ("OLÁ, MUNDO!"), depois revertida ("! DLROW ,OLLEH").

Principais conceitos explicados

Interface de Executor

Executores de funções:

  • Usar BindExecutor() para criar um executor a partir de uma função

Implementação de Executor<TInput, TOutput>executores:

  • TInput: o tipo de dados que este executor aceita
  • TOutput: o tipo de dados que este executor produz
  • HandleAsync: o método que processa a entrada e retorna a saída

Padrão do Construtor de Fluxo de Trabalho do .NET

O WorkflowBuilder fornece uma API fluente para a construção de fluxos de trabalho:

  • Construtor: usa o executor inicial
  • AddEdge(): cria conexões direcionadas entre executores
  • WithOutputFrom(): especifica quais executores produzem saídas de fluxo de trabalho
  • Build(): cria o fluxo de trabalho imutável final

Tipos de evento .NET

Durante a execução, você pode observar esses tipos de evento:

  • ExecutorCompletedEvent - Quando um executor conclui o processamento

Exemplo completo do .NET

Para obter a implementação completa e pronta para execução, consulte o exemplo de 01_ExecutorsAndEdges no repositório do Agent Framework.

Este exemplo inclui:

  • Implementação completa com todas as instruções de uso e estrutura de classe
  • Comentários adicionais explicando os conceitos de fluxo de trabalho
  • Concluir a preparação e configuração do projeto

Visão geral

Neste tutorial, você criará um fluxo de trabalho com dois executores:

  1. Executor de Maiúsculas - Converte texto de entrada em maiúsculas
  2. Executor de Texto Inverso – Inverte o texto e gera o resultado final

O fluxo de trabalho demonstra os principais conceitos como:

  • Usando o @executor decorador para criar nós de fluxo de trabalho
  • Conectando executores com WorkflowBuilder
  • Passando dados entre etapas com ctx.send_message()
  • Gerando saída final com ctx.yield_output()
  • Eventos de streaming para observabilidade em tempo real

Conceitos abordados

Pré-requisitos

  • Python 3.10 ou posterior
  • Pacote python do Agent Framework Core instalado: pip install agent-framework-core --pre
  • Nenhum serviço de IA externo necessário para este exemplo básico

Implementação passo a passo

As seções a seguir mostram como criar o fluxo de trabalho sequencial passo a passo.

Etapa 1: Importar módulos necessários

Primeiro, importe os módulos necessários do Agent Framework:

import asyncio
from typing_extensions import Never
from agent_framework import WorkflowBuilder, WorkflowContext, WorkflowOutputEvent, executor

Etapa 2: Criar o Primeiro Executor

Crie um executor que converte o texto em letras maiúsculas implementando um executor com um método de manipulador:

class UpperCase(Executor):
    def __init__(self, id: str):
        super().__init__(id=id)

    @handler
    async def to_upper_case(self, text: str, ctx: WorkflowContext[str]) -> None:
        """Convert the input to uppercase and forward it to the next node.

        Note: The WorkflowContext is parameterized with the type this handler will
        emit. Here WorkflowContext[str] means downstream nodes should expect str.
        """
        result = text.upper()

        # Send the result to the next executor in the workflow.
        await ctx.send_message(result)

Pontos-chave:

  • O @executor decorador registra essa função como um nó de fluxo de trabalho
  • WorkflowContext[str] indica que esse executor envia uma cadeia de caracteres downstream especificando o primeiro parâmetro de tipo
  • ctx.send_message() passa dados para a próxima etapa

Etapa 3: Criar o Segundo Executor

Crie um executor que inverte o texto e produz a saída final de um método decorado com @executor:

@executor(id="reverse_text_executor")
async def reverse_text(text: str, ctx: WorkflowContext[Never, str]) -> None:
    """Reverse the input and yield the workflow output."""
    result = text[::-1]

    # Yield the final output for this workflow run
    await ctx.yield_output(result)

Pontos-chave:

  • WorkflowContext[Never, str] indica que esse é um executor de terminal que não envia mensagens especificando Never como o primeiro parâmetro de tipo, mas produz saídas de fluxo de trabalho especificando str como o segundo parâmetro
  • ctx.yield_output() fornece o resultado final do fluxo de trabalho
  • O fluxo de trabalho é concluído quando torna-se ocioso

Etapa 4: Criar o fluxo de trabalho

Conecte os executores usando WorkflowBuilder:

upper_case = UpperCase(id="upper_case_executor")

workflow = (
    WorkflowBuilder()
    .add_edge(upper_case, reverse_text)
    .set_start_executor(upper_case)
    .build()
)

Pontos-chave:

  • add_edge() cria conexões direcionadas entre executores
  • set_start_executor() define o ponto de entrada
  • build() finaliza o fluxo de trabalho

Etapa 5: Executar o fluxo de trabalho com streaming

Execute o fluxo de trabalho e observe eventos em tempo real:

async def main():
    # Run the workflow and stream events
    async for event in workflow.run_stream("hello world"):
        print(f"Event: {event}")
        if isinstance(event, WorkflowOutputEvent):
            print(f"Workflow completed with result: {event.data}")

if __name__ == "__main__":
    asyncio.run(main())

Etapa 6: Noções básicas sobre a saída

Ao executar o fluxo de trabalho, você verá eventos como:

Event: ExecutorInvokedEvent(executor_id=upper_case_executor)
Event: ExecutorCompletedEvent(executor_id=upper_case_executor)
Event: ExecutorInvokedEvent(executor_id=reverse_text_executor)
Event: ExecutorCompletedEvent(executor_id=reverse_text_executor)
Event: WorkflowOutputEvent(data='DLROW OLLEH', source_executor_id=reverse_text_executor)
Workflow completed with result: DLROW OLLEH

Principais conceitos explicados

Tipos de contexto de fluxo de trabalho

O WorkflowContext tipo genérico define quais dados fluem entre executores:

  • WorkflowContext[str] - Envia uma cadeia de caracteres para o próximo executor
  • WorkflowContext[Never, str] - Executor de terminal que produz uma saída de fluxo de trabalho do tipo string

Tipos de evento

Durante a execução do streaming, você observará estes tipos de evento:

  • ExecutorInvokedEvent - Quando um executor inicia o processamento
  • ExecutorCompletedEvent - Quando um executor conclui o processamento
  • WorkflowOutputEvent - Contém o resultado final do fluxo de trabalho

Padrão do Construtor de Fluxo de Trabalho do Python

O WorkflowBuilder fornece uma API fluente para a construção de fluxos de trabalho:

  • add_edge(): cria conexões direcionadas entre executores
  • set_start_executor(): define o ponto de entrada do fluxo de trabalho
  • build(): finaliza e retorna um objeto de fluxo de trabalho imutável

Exemplo completo

Para obter a implementação completa e pronta para execução, consulte o exemplo no repositório do Agent Framework.

Este exemplo inclui:

  • Implementação completa com todas as importações e documentação
  • Comentários adicionais explicando os conceitos de fluxo de trabalho
  • Saída de exemplo mostrando os resultados esperados

Próximas etapas

Saiba mais sobre como criar um fluxo de trabalho simultâneo simples

  • Last updated on