Tworzenie prostego sekwencyjnego przepływu pracy

Przegląd

W tym samouczku utworzysz przepływ pracy z dwoma funkcjami wykonawczych:

1. Moduł wielkich liter — konwertuje tekst wejściowy na wielkie litery
2. Funkcja wykonawcza odwrotnego tekstu — odwraca tekst i zwraca wynik końcowy

Przepływ pracy demonstruje podstawowe pojęcia, takie jak:

• Tworzenie niestandardowej funkcji wykonawczej z jedną procedurą obsługi
• Tworzenie niestandardowej funkcji wykonawczej na podstawie funkcji
• Używanie WorkflowBuilder funkcji do łączenia funkcji wykonawczych z krawędziami
• Przetwarzanie danych za pomocą kroków sekwencyjnych
• Obserwowanie wykonywania przepływu pracy za pośrednictwem zdarzeń

Omówione pojęcia

• Executors
• Bezpośrednie krawędzie
• Konstruktor przepływu pracy
• Events

Wymagania wstępne

• Pakiet .NET 8.0 SDK lub nowszy
• W tym przykładzie podstawowym nie są wymagane żadne zewnętrzne usługi sztucznej inteligencji
• Nowa aplikacja konsolowa

Implementacja krok po kroku

W poniższych sekcjach pokazano, jak utworzyć sekwencyjny przepływ pracy krok po kroku.

Krok 1. Instalowanie pakietów NuGet

Najpierw zainstaluj wymagane pakiety dla projektu .NET:

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

Krok 2. Definiowanie funkcji wykonawczej wielkiej litery

Zdefiniuj funkcję wykonawcza, która konwertuje tekst na wielkie litery:

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

Kluczowe punkty:

• Utwórz funkcję, która przyjmuje ciąg i zwraca wersję wielkimi literami.
• Użyj BindExecutor() do utworzenia wykonawcy na podstawie funkcji

Krok 3. Definiowanie egzekutora odwrotnego tekstu

Zdefiniuj funkcję wykonawcza, która odwraca tekst:

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

Kluczowe punkty:

• Utwórz klasę, która dziedziczy z Executor<TInput, TOutput>
• Zaimplementuj HandleAsync() , aby przetworzyć dane wejściowe i zwrócić dane wyjściowe

Krok 4. Kompilowanie i łączenie przepływu pracy

Połącz funkcje wykonawcze przy użyciu polecenia WorkflowBuilder:

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

Kluczowe punkty:

• WorkflowBuilder konstruktor przyjmuje początkowego egzekutora
• AddEdge() Tworzy przekierowane połączenie z wielkich liter na odwrotne
• WithOutputFrom() określa, które funkcje wykonawcze generują dane wyjściowe przepływu pracy
• Build() tworzy niezmienny przepływ pracy

Krok 5. Wykonywanie przepływu pracy

Uruchom przepływ pracy i obserwuj wyniki:

// 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;
    }
}

Krok 6. Opis danych wyjściowych przepływu pracy

Po uruchomieniu przepływu pracy zobaczysz dane wyjściowe, takie jak:

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

Dane wejściowe "Hello, World!" są najpierw konwertowane na wielkie litery ("HELLO, WORLD!"), a następnie odwrócone ("! DLROW ,OLLEH").

Objaśnione kluczowe pojęcia

Interfejs wykonawczy

Funkcje wykonawcze:

• Użyj BindExecutor(), aby utworzyć wykonawcę na podstawie funkcji

Egzekutory implementująExecutor<TInput, TOutput>

• TInput: typ danych akceptowanych przez funkcję wykonawcza
• TOutput: typ danych generowanych przez funkcję wykonawcza
• HandleAsync: metoda, która przetwarza dane wejściowe i zwraca dane wyjściowe

Wzorzec konstruktora przepływu pracy platformy .NET

Zapewnia WorkflowBuilder płynny interfejs API do tworzenia przepływów pracy:

• Konstruktor: przyjmuje początkowego wykonawcę
• AddEdge(): Tworzy połączenia kierowane między funkcjami wykonawczych
• WithOutputFrom(): określa, które funkcje wykonawcze generują dane wyjściowe przepływu pracy
• Build(): Tworzy ostatni niezmienny przepływ pracy

Typy zdarzeń platformy .NET

Podczas wykonywania można obserwować następujące typy zdarzeń:

• ExecutorCompletedEvent - Po zakończeniu przetwarzania przez funkcję wykonawcza

Kompletny przykład platformy .NET

Aby uzyskać pełną, gotową do uruchomienia implementację, zobacz przykład 01_ExecutorsAndEdges w repozytorium Platformy agentów.

Ten przykład obejmuje:

• Pełna implementacja ze wszystkimi instrukcjami using i strukturą klas
• Dodatkowe komentarze objaśniające pojęcia dotyczące przepływu pracy
• Ukończ instalację i konfigurację projektu

Przegląd

W tym samouczku utworzysz przepływ pracy z dwoma funkcjami wykonawczych:

1. Moduł wielkich liter — konwertuje tekst wejściowy na pisanie wielkimi literami
2. Funkcja wykonawcza odwrotnego tekstu — odwraca tekst i zwraca wynik końcowy

Przepływ pracy demonstruje podstawowe pojęcia, takie jak:

• Użycie dekoratora @executor do tworzenia węzłów przepływu pracy
• Łączenie funkcji wykonawczych za pomocą polecenia WorkflowBuilder
• Przekazywanie danych między krokami za pomocą polecenia ctx.send_message()
• Uzyskiwanie końcowych danych wyjściowych za pomocą polecenia ctx.yield_output()
• Zdarzenia strumieniowe dla monitorowania w czasie rzeczywistym

Omówione pojęcia

• Executors
• Bezpośrednie krawędzie
• Konstruktor przepływu pracy
• Events

Wymagania wstępne

• Środowisko Python w wersji 3.10 lub nowszej
• Zainstalowany pakiet Agent Framework Core w języku Python: pip install agent-framework-core --pre
• W tym przykładzie podstawowym nie są wymagane żadne zewnętrzne usługi sztucznej inteligencji

Implementacja krok po kroku

W poniższych sekcjach pokazano, jak utworzyć sekwencyjny przepływ pracy krok po kroku.

Krok 1. Importowanie wymaganych modułów

Najpierw zaimportuj niezbędne moduły z programu Agent Framework:

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

Krok 2. Tworzenie pierwszego wykonawcy

Utwórz funkcję wykonawczą, która konwertuje tekst na wielkie litery, implementując funkcję wykonawczą za pomocą metody obsługi:

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)

Kluczowe punkty:

• Dekorator @executor rejestruje tę funkcję jako węzeł przepływu pracy
• WorkflowContext[str] wskazuje, że funkcja wykonawcza wysyła ciąg podrzędny, określając pierwszy parametr typu
• ctx.send_message() przekazuje dane do następnego kroku

Krok 3. Tworzenie drugiego modułu wykonawczego

Utwórz egzekutor, który odwraca tekst i zwraca końcowe dane wyjściowe z metody przy użyciu @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)

Kluczowe punkty:

• WorkflowContext[Never, str] wskazuje, że jest to funkcja wykonawcza terminalu, która nie wysyła żadnych komunikatów, określając Never jako pierwszy parametr typu, ale generuje dane wyjściowe przepływu pracy, określając str jako drugi parametr
• ctx.yield_output() udostępnia końcowy wynik przepływu pracy
• Przepływ pracy kończy się, gdy przejdzie w stan bezczynności

Krok 4. Kompilowanie przepływu pracy

Połącz funkcje wykonawcze przy użyciu polecenia WorkflowBuilder:

upper_case = UpperCase(id="upper_case_executor")

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

Kluczowe punkty:

• add_edge() tworzy połączenia kierowane między wykonawcami
• set_start_executor() definiuje punkt wejścia
• build() finalizuje przepływ pracy

Krok 5. Uruchamianie przepływu pracy za pomocą przesyłania strumieniowego

Wykonaj przepływ pracy i obserwuj zdarzenia w czasie rzeczywistym:

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

Krok 6. Omówienie danych wyjściowych

Po uruchomieniu przepływu pracy zobaczysz zdarzenia takie jak:

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

Objaśnione kluczowe pojęcia

Typy kontekstowe przepływu pracy

Typ WorkflowContext ogólny definiuje przepływy danych między funkcjami wykonawczych:

• WorkflowContext[str] - Wysyła ciąg do następnego wykonawcy
• WorkflowContext[Never, str] - Funkcja wykonawcza terminalu, która zwraca dane wyjściowe przepływu pracy typu string

Typy zdarzeń

Podczas wykonywania przesyłania strumieniowego będziesz obserwować następujące typy zdarzeń:

• ExecutorInvokedEvent - Kiedy funkcja wykonawcza rozpoczyna przetwarzanie
• ExecutorCompletedEvent - Po zakończeniu przetwarzania przez funkcję wykonawcza
• WorkflowOutputEvent — Zawiera końcowy wynik przepływu pracy

Wzorzec konstruktora przepływów pracy języka Python

Zapewnia WorkflowBuilder płynny interfejs API do tworzenia przepływów pracy:

• add_edge(): Tworzy połączenia kierowane między egzekutorami
• set_start_executor(): Definiuje punkt wejścia przepływu pracy
• build(): Finalizuje i zwraca niezmienny obiekt przepływu pracy

Kompletny przykład

Aby uzyskać pełną, gotową do uruchomienia implementację, zobacz przykład w repozytorium Platformy agentów.

Ten przykład obejmuje:

• Pełna implementacja ze wszystkimi importami i dokumentacją
• Dodatkowe komentarze objaśniające pojęcia dotyczące przepływu pracy
• Przykładowe dane wyjściowe przedstawiające oczekiwane wyniki