Konstruktor przepływu pracy i wdrażanie

Przepływ pracy łączy funkcje wykonawcze i krawędzie ze sobą w skierowany graf i zarządza wykonywaniem. Koordynuje wywołanie egzekutora, routing komunikatów i streaming zdarzeń.

Tworzenie przepływów pracy

Przepływy pracy są tworzone przy użyciu WorkflowBuilder klasy , która zapewnia płynny interfejs API do definiowania struktury przepływu pracy:

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

Przepływy pracy są tworzone przy użyciu WorkflowBuilder klasy :

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

Podczas walidacji w czasie kompilacji program obsługi może używać samodzielnej adnotacji kontenera, takiej jak list, dict, tuple lub Sequence. Może zaakceptować dane wyjściowe korzystające z odpowiedniego kontenera sparametryzowanego, takiego jak list[str].

Pakiet workflow udostępnia model wykonywania oparty na grafach, w którym funkcje wykonawcze są połączone za pomocą krawędzi.

Funkcja wykonawcza — jednostka przetwarzania, która odbiera dane wejściowe i generuje dane wyjściowe
Edge — łączy dane wyjściowe jednej funkcji wykonawczej z danymi wejściowymi innego
Builder — tworzy przepływy pracy przez definiowanie funkcji wykonawczych i krawędzi
Run — wykonuje przepływ pracy z podanymi danymi wejściowymi

Ta funkcja tworzy przepływ pracy przetwarzania tekstu bez konieczności używania modelu lub poświadczeń:

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) Uwidacznia wynik odwrotnego wykonawcy jako dane wyjściowe przepływu pracy. Krawędź nadal dostarcza wynik zapisany wielkimi literami do odwrotnego egzekutora niezależnie od oznaczenia wyjściowego.

Przekaż przepływ pracy zwrócony przez buildWorkflow do jednej z poniższych funkcji wykonywania. Dla danych wejściowych "Hello, World!", oba drukują Workflow output: !DLROW ,OLLEH.

Wykonywanie przepływu pracy

Przepływy pracy obsługują zarówno tryby przesyłania strumieniowego, jak i nieprzesyłania strumieniowego.

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

Użyj polecenia RunStreaming, jeśli chcesz widzieć zdarzenia, gdy się pojawiają. Przekaż kontekst wywołującego, aby anulowanie dotarło do workflow, sprawdź jego błąd po zakończeniu iteracji i zamknij uruchomienie po zakończeniu:

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 czeka, gdy zewnętrzne żądania są w toku. W przypadku przepływów pracy, które wymagają danych wejściowych od człowieka, obsłuż zdarzenia żądań lub użyj WatchUntilHalt, aby wrócić po wstrzymaniu. Zobacz Zdarzenia przepływu pracy.

Użyj polecenia Run , aby przejść do następnego zatrzymania, a następnie sprawdzić zebrane zdarzenia. Przebieg może zostać wstrzymany podczas oczekiwania na dane wejściowe zewnętrzne, a nie tylko po zakończeniu pracy:

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() przesuwa swoją zakładkę za każdym razem, gdy ją wywołasz. Druga pętla używa OutgoingEvents() do ponownego odtwarzania skumulowanych zdarzeń wykonawcy zamiast próby ponownego ich przetwarzania za pomocą polecenia NewEvents().

Te funkcje kończą swoje działanie po zaobserwowaniu wyniku. Pozostaw przebieg otwarty, jeśli aplikacja musi wysyłać więcej komunikatów lub wznowić go po przerwie.

Zobacz Zdarzenia przepływu pracy, aby uzyskać pełne informacje o konsumentach, klasyfikacji danych wyjściowych i semantyce wstrzymania.

Tip

Zobacz przykłady przepływów pracy, aby zapoznać się z kompletnymi, gotowymi do uruchomienia przykładami.

Walidacja przepływu pracy

Konstruktory przepływu pracy weryfikują graf przed wykonaniem. Dokładne kontrole zależą od zestawu SDK i dostępnych metadanych typu:

Zgodność typów: sprawdza zadeklarowane protokoły komunikatów między połączonymi wykonawcami, w których są dostępne metadane.
Spójność grafu: sprawdza, czy zarejestrowane executory są osiągalne z executora startowego.
Powiązanie wykonawcy: sprawdza, czy są obecne wymagane rejestracje wykonawcy.
Weryfikacja krawędzi: sprawdza nieprawidłowe połączenia i zduplikowane krawędzie zgodnie z regułami konstruktora.

Polecenie Go Build() odrzuca nierozwiązane powiązania, nieosiągalne zarejestrowane funkcje wykonawcze, zduplikowane bezwarunkowe krawędzie i niezgodne zadeklarowane typy komunikatów. Trasy warunkowe i protokoły dynamiczne nadal wymagają testów w czasie działania.

Walidacja typu może tworzyć tymczasowe instancje wykonawcy w celu sprawdzenia obsługiwanych protokołów. Jeśli nie można zbadać fabryki lub protokołu, to sprawdzenie zgodności zostanie pominięte. Pomyślna kompilacja nie gwarantuje, że tworzenie executora lub każde dostarczanie komunikatów zakończy się powodzeniem w czasie wykonywania. Obsługa błędów podczas tworzenia przebiegu, iteracji zdarzeń i zdarzeń niepowodzenia przepływu pracy.

BuildWithOptions(workflow.BuildOptions{DisableOrphanValidation: true}) wyłącza tylko sprawdzanie orphan-executor. Inne walidacje nadal są uruchamiane.

Model wykonywania: superkroki

Struktura używa zmodyfikowanego modelu wykonywania Pregel — podejścia równoległego zbiorczego synchronicznego (BSP) z przetwarzaniem opartym na superkrokach.

Jak działają superkroki

Organizacja przepływu zadań odbywa się w dyskretnych superkrokach. Każdy superkrok:

Zbiera wszystkie oczekujące komunikaty z poprzedniego superkroka
Kieruje komunikaty do docelowych wykonawców w oparciu o definicje krawędzi
Uruchamia wszystkie funkcje wykonawcze obiektów docelowych jednocześnie w ramach superkroka
Czeka na ukończenie wszystkich funkcji wykonawczych przed przejściem (bariera synchronizacji)
Kolejkuje wszystkie nowe komunikaty emitowane przez funkcje wykonawcze dla następnego kroku

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   │
└─────────────────┘    └─────────────────┘

Bariera synchronizacji

Najważniejszą cechą jest bariera synchronizacji między superkrokami. Wyzwalane egzekutory mogą działać współbieżnie w ramach superkroku, ale przepływ pracy nie przechodzi do następnego superkroku, dopóki nie zakończą działania.

Ma to wpływ na wzorce fan-out: jeśli rozgałęzisz na wiele ścieżek — jedną z łańcuchem wykonawców, a drugą z pojedynczym długotrwałym wykonawcą — ścieżka z łańcuchem wykonawców nie może przejść do momentu zakończenia działania długotrwałego wykonawcy.

Dlaczego superkroki?

Model BSP zapewnia ważne gwarancje:

Zdefiniowane granice kroków: komunikaty emitowane podczas superstepu są kolejkowane do późniejszego przetwarzania.
Granice punktów kontrolnych: Przepływy pracy z obsługą punktów kontrolnych zapisują stan wykonywania na granicach superkroku.
Aktualizacje stanu buforowanego: zmiany stanu przepływu pracy w kolejce są zatwierdzane między superkrokami.

Te gwarancje nie naprawiają wykonywania ani porządkowania zdarzeń między współbieżnymi wykonawcami w ramach jednego superkroka. Zsynchronizuj dostęp do stanu współużytkowanego należącego do aplikacji, zamiast zakładać, że bariera superkroku zapobiega wszystkim warunkom wyścigu.

Praca z modelem Superstep

Jeśli potrzebujesz naprawdę niezależnych ścieżek równoległych, które nie blokują się nawzajem, skonsoliduj sekwencyjne kroki do pojedynczego modułu wykonawczego. Zamiast łączyć łańcuch step1 → step2 → step3, połącz logikę w jedną funkcję wykonawcza. Obie ścieżki równoległe są następnie wykonywane w ramach jednego superkroka.

Następne kroki

Agenci w przepływach pracy

Powiązane tematy:

Funkcje wykonawcze — jednostki przetwarzania w przepływie pracy
Krawędzie — połączenia między wykonawcami
Zdarzenia — możliwość obserwowania przepływu pracy
Zarządzanie stanem

Opinia

Czy ta strona była pomocna?

Ostatnia aktualizacja o 2026-10-29