Partilhar 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.

Os fluxos de trabalho sequenciais são a base da construção de sistemas complexos de agentes de IA. Este tutorial mostra como criar um fluxo de trabalho simples de duas etapas, onde 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 texto de entrada em maiúsculas
  2. Reverse Text Executor - Inverte o texto e produz o resultado final

O fluxo de trabalho demonstra conceitos fundamentais como:

  • Criar um executor personalizado com um único handler
  • Criação de um executor personalizado a partir de uma função
  • Usando WorkflowBuilder para conectar executores com bordas
  • Processamento de dados através de etapas sequenciais
  • Observando a execução do fluxo de trabalho através 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 .NET:

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

Etapa 2: Definir o Executor de Maiúsculas

Defina um executor que converta texto em 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 principais:

  • Criar uma função que recebe uma string e devolve a versão maiúscula.
  • Use BindExecutor() para criar um executor a partir da função

Etapa 3: Definir o executor de texto reverso

Defina um executor que inverta 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 principais:

  • Crie uma classe que herde de Executor<TInput, TOutput>
  • Implementar HandleAsync() para processar a entrada e devolver 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 principais:

  • WorkflowBuilder construtor aceita o executor inicial
  • AddEdge() cria uma conexão direcionada de maiúsculas para inversas
  • 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á saídas como:

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

A entrada "Hello, World!" é primeiro convertida em maiúsculas ("HELLO, WORLD!"), depois invertida ("!DLROW ,OLLEH").

Conceitos-chave explicados

Interface do Executor

Executores das funções:

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

Os executores implementam Executor<TInput, TOutput>:

  • 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 .NET

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

  • Construtor: Recebe 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 final imutável

Tipos de evento .NET

Durante a execução, você pode observar estes tipos de eventos:

  • ExecutorCompletedEvent - Quando um executor termina 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 todos usando instruções e estrutura de classe
  • Comentários adicionais explicando os conceitos de fluxo de trabalho
  • Conclua a instalação e configuração do projeto

Visão geral

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

  1. Conversor para maiúsculas - Converte o texto fornecido em letras maiúsculas
  2. Reverse Text Executor - Inverte o texto e produz o resultado final

O fluxo de trabalho demonstra conceitos fundamentais como:

  • Usando o decorador @executor para criar nós de fluxo de trabalho
  • Conectando executores com WorkflowBuilder
  • Passando dados entre etapas com ctx.send_message()
  • Resultado final com ctx.yield_output()
  • Transmissão de eventos para observação 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
  • Não são necessários serviços de IA externos 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 converta texto em maiúsculas implementando um executor com um método handler:

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 principais:

  • O @executor decorador registra essa função como um nó de fluxo de trabalho
  • WorkflowContext[str] indica que este executor envia uma string a jusante 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 inverta o texto e produza 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 principais:

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

  • 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: Entendendo 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

Conceitos-chave explicados

Tipos de contexto de fluxo de trabalho

O WorkflowContext tipo genérico define que dados fluem entre executores.

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

Tipos de Eventos

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

  • ExecutorInvokedEvent - Quando um executor começa a processar
  • ExecutorCompletedEvent - Quando um executor termina o processamento
  • WorkflowOutputEvent - Contém o resultado final do fluxo de trabalho

Padrão Python Construtor de Fluxo de Trabalho

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 a implementação completa e pronta a executar, consulte o exemplo no repositório 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 amostra mostrando os resultados esperados

Próximas Etapas

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

  • Last updated on