Tvůrce pracovních postupů a provádění

Pracovní postup spojuje exekutory a hrany do řízeného grafu a spravuje provádění. Koordinuje vyvolání exekutoru, směrování zpráv a streamování událostí.

Vytváření pracovních postupů

Pracovní postupy se vytvářejí pomocí WorkflowBuilder třídy, která poskytuje rozhraní API fluent pro definování struktury pracovního postupu:

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

Pracovní postupy se vytvářejí pomocí WorkflowBuilder třídy:

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

Během ověřování v době sestavení může obslužná rutina použít holou anotaci kontejneru, například list, dict, tuple nebo Sequence. Může přijmout výstup, který používá odpovídající parametrizovaný kontejner, například list[str].

Balíček workflow poskytuje grafový model provádění, ve kterém jsou vykonavatelé propojeni pomocí hran.

Exekutor – výpočetní jednotka, která přijímá vstup a vytváří výstup
Edge – připojí výstup jednoho exekutoru ke vstupu jiného exekutoru.
Tvůrce – Vytváření pracovních postupů definováním exekutorů a hran
Spustit - Spustí workflow se zadaným vstupem

Tato funkce vytváří pracovní postup zpracování textu, aniž by vyžadovala model nebo přihlašovací údaje:

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) zveřejňuje výsledek reverse executoru jako výstup pracovního postupu. Edge stále poskytuje výsledek převedený na velká písmena reverse executoru bez ohledu na označení výstupu.

Předejte pracovní postup vrácený funkcí buildWorkflow některé z níže uvedených funkcí spuštění. Pro vstup "Hello, World!" oba vytisknou Workflow output: !DLROW ,OLLEH.

Provádění pracovního postupu

Pracovní postupy podporují režimy streamování i spouštění bez streamování:

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

Použijte RunStreaming, když chcete, aby se události zobrazovaly v reálném čase. Předejte kontext volajícího, aby zrušení dosáhlo pracovního postupu, zkontrolujte jeho chybu po skončení iterace a po dokončení zavřete spuštění:

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 čeká na vyřízení externích požadavků. U pracovních postupů, které vyžadují lidský vstup, zpracujte události požadavků nebo použijte WatchUntilHalt pro návrat při pozastavení. Viz události pracovního postupu.

Pomocí Run přejděte na další zastavení a pak zkontrolujte shromážděné události. Běh se může zastavit při čekání na externí vstup, a to nejen při dokončení jeho práce:

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() při každém volání posune svou záložku dál. Druhá smyčka používá OutgoingEvents() k přehrání nahromaděných událostí exekutoru místo toho, aby se je znovu pokoušela zpracovat pomocí NewEvents().

Tyto funkce po obdržení výsledku ukončí svůj běh. Pokud vaše aplikace potřebuje posílat další zprávy nebo po pozastavení obnovit běh, nechte ho místo toho otevřený.

Podívejte se na Workflow events, kde najdete kompletní consumers, klasifikaci výstupu a sémantiku pozastavení.

Tip

Kompletní spustitelné ukázky najdete v příkladech pracovního postupu .

Ověření pracovního postupu

Nástroje pro vytváření pracovních postupů před spuštěním ověřují graf. Přesné kontroly závisí na sadě SDK a dostupných metadatech typu:

Kompatibilita typů: Kontroluje deklarované protokoly zpráv mezi připojenými exekutory, kde jsou k dispozici metadata.
Propojení grafu: Kontroluje, jestli jsou zaregistrované exekutory dostupné od spouštěcího exekutoru.
Svázání executoru: Kontroluje, zda jsou k dispozici požadované registrace executoru.
Ověření hran: Kontroluje neplatná připojení a duplicitní hrany podle pravidel tvůrce.

Go Build() odmítne nevyřešené vazby, nedostupné registrované exekutory, duplicitní nepodmíněné hrany a nekompatibilní deklarované typy zpráv. Podmíněné trasy a dynamické protokoly stále vyžadují testování za běhu.

Ověření typu může vytvořit dočasné instance exekutorů pro kontrolu jejich protokolů. Pokud není možné zkontrolovat továrnu nebo protokol, tato kontrola kompatibility se přeskočí. Úspěšné sestavení nezaručuje vytvoření exekutoru nebo úspěšné doručení každé zprávy za běhu. Zpracujte chyby při vytváření spuštění, iteraci událostí a událostech selhání pracovního postupu

BuildWithOptions(workflow.BuildOptions{DisableOrphanValidation: true}) vypne pouze kontrolu osamoceného exekutoru. Další ověřování se pořád spouští.

Model provádění: Supersteps

Architektura používá upravený model provádění Pregel – metodu BSP (Bulk Synchronous Parallel) se zpracováním založeným na superkrocích.

Jak fungují superkroky

Provádění pracovního postupu je uspořádané do samostatných superkroků. Každý superkrok:

Shromažďuje všechny čekající zprávy z předchozího superkroku.
Směruje zprávy do cílových vykonavatelů na základě definic hran.
Spustí všechny cílové vykonavatele souběžně v rámci superkroku.
Čeká na dokončení všech vykonavatelů než postoupí (synchronizační bariéra).
Zařadí všechny nové zprávy generované exekutory do fronty pro další superkrok.

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

Synchronizační bariéra

Nejdůležitější vlastností je synchronizační bariéra mezi superkroky. Aktivované exekutory se můžou spustit souběžně v rámci superkroku, ale pracovní postup nepřejde na další superkrok, dokud neskončí.

To má vliv na vzory rozvětvení: pokud rozvětvíte do více cest – jednu s řetězem vykonavatelů a druhou s jedním dlouhotrvajícím vykonavatelem – zřetězená cesta nemůže pokračovat, dokud dlouhotrvající vykonavatel nedokončí svou činnost.

Proč superkroky?

Model BSP poskytuje důležité záruky:

Definované hranice kroku: Zprávy generované během superkroku se zařadí do fronty pro pozdější zpracování.
Hranice kontrolních bodů: Pracovní postupy s podporou kontrolních bodů ukládají stav provádění na hranicích superkroků.
Bufferované aktualizace stavu: Změny stavu pracovního postupu ve frontě jsou zapsány mezi superkroky.

Tyto záruky nezajišťují provádění ani řazení událostí mezi souběžnými exekutory v rámci jednoho superkroku. Synchronizujte přístup ke sdílenému stavu aplikace a nepředpokládejte, že bariéra superkroku brání všem souběžným konfliktům.

Práce s modelem Superstep

Pokud potřebujete skutečně nezávislé paralelní cesty, které se navzájem nezablokují, konsolidujte sekvenční kroky do jednoho exekutoru. Místo řetězení step1 → step2 → step3zkombinujte tuto logiku do jednoho exekutoru. Obě paralelní cesty se pak provádějí v rámci jednoho superkroku.

Další kroky

Agenti v pracovních postupech

Související témata:

Exekutory – jednotky zpracování v pracovním postupu
Hrany – spojení mezi vykonavateli
Události – pozorovatelnost pracovního postupu
Správa stavu

Váš názor

Byla tato stránka užitečná?

Poslední aktualizace v 2026-10-29