Munkafolyamat-szerkesztő és -végrehajtás

A munkafolyamat összekapcsolja a végrehajtókat és éleket egy irányított gráfba, és kezeli a végrehajtást. Koordinálja a végrehajtói hívásokat, az üzenet-útválasztást és az eseménystreamelést.

Munkafolyamatok létrehozása

A munkafolyamatok az WorkflowBuilder osztály használatával jönnek létre, amely egy fluent API-t biztosít a munkafolyamat-struktúra meghatározásához:

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

A munkafolyamatok az WorkflowBuilder osztály használatával jönnek létre:

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

A workflow csomag egy gráfalapú végrehajtási modellt biztosít, amelyben a végrehajtókat élek kötik össze.

  • Végrehajtó – Bemenetet fogadó és kimenetet termelő feldolgozó egység
  • Edge – Az egyik végrehajtó kimenetének csatlakoztatása egy másik bemenetéhez
  • Builder – Munkafolyamatok létrehozása végrehajtók és élek definiálásával
  • Futtatás - Egy munkafolyamat végrehajtása a megadott bemenettel
import (
    "github.com/microsoft/agent-framework-go/workflow"
    "github.com/microsoft/agent-framework-go/workflow/inproc"
)

uppercase := workflow.NewExecutor("UppercaseExecutor", func(input string) string {
    return strings.ToUpper(input)
}).Bind()

reverse := workflow.NewExecutor("ReverseExecutor", func(input string) string {
    runes := []rune(input)
    slices.Reverse(runes)
    return string(runes)
}).Bind()

wf, err := workflow.NewBuilder(uppercase).
    AddEdge(uppercase, reverse).
    WithOutputFrom(reverse).
    Build()
if err != nil {
    return err
}

Munkafolyamat végrehajtása

A munkafolyamatok támogatják a streamelési és a nem streamelési végrehajtási módokat is:

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

Akkor használja RunStreaming , ha eseményeket szeretne, amikor azok történnek:

stream, err := inproc.Default.RunStreaming(context.Background(), wf, "Hello, World!")
if err != nil {
    return err
}
defer stream.Close(context.Background())

for evt, err := range stream.WatchStream(context.Background()) {
    if err != nil {
        return err
    }
    if output, ok := evt.(workflow.OutputEvent); ok {
        fmt.Printf("Workflow completed: %v\n", output.Output)
    }
}

Akkor használja Run , ha meg szeretné várni a munkafolyamat befejezését, majd vizsgálja meg az összegyűjtött eseményeket:

run, err := inproc.Default.Run(context.Background(), wf, "Hello, World!")
if err != nil {
    return err
}

for evt := range run.NewEvents() {
    if output, ok := evt.(workflow.OutputEvent); ok {
        fmt.Printf("Final result: %v\n", output.Output)
    }
}

Megvizsgálhatja a nem adatfolyamos futtatás során gyűjtött végrehajtói eseményeket is:

for evt := range run.NewEvents() {
    if evt, ok := evt.(workflow.ExecutorCompletedEvent); ok {
        fmt.Printf("%s: %v\n", evt.ExecutorID, evt.Result)
    }
}

Tip

Tekintse meg a munkafolyamat-példákat a teljes futtatható mintákhoz.

Munkafolyamat érvényesítése

A keretrendszer átfogó ellenőrzést végez munkafolyamatok létrehozásakor:

  • Típuskompatibilitás: Biztosítja, hogy az üzenettípusok kompatibilisek legyenek a csatlakoztatott végrehajtók között
  • Gráfkapcsolat: Ellenőrzi, hogy az összes végrehajtó elérhető-e a kezdő végrehajtótól
  • Végrehajtók kötése: Annak ellenőrzése, hogy minden végrehajtó megfelelően van-e kötve és példányosítva
  • Élérvényesítés: Ismétlődő élek és érvénytelen kapcsolatok ellenőrzése

Végrehajtási modell: Szuperszteps

A keretrendszer egy módosított Pregel végrehajtási modellt használ – egy tömeges szinkron párhuzamos (BSP) megközelítést superstep-alapú feldolgozással.

A Supersteps működése

A munkafolyamat-végrehajtás különálló szuperstepsekbe van rendezve. Minden szuperlépés:

  1. Összegyűjti az előző szupersztep összes függőben lévő üzenetét
  2. Üzenetek irányítása célvégrehajtókhoz éldefiníciók alapján
  3. A szuperlépésben az összes cél-végrehajtó egyidejű futtatása
  4. Várakozás az összes végrehajtó befejezésére a továbbhaladás előtt (szinkronizációs akadály)
  5. A végrehajtók által a következő szuperstepre küldött új üzenetek várólistája
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   │
└─────────────────┘    └─────────────────┘

Szinkronizálási akadály

A legfontosabb jellemző a szuperstepsek közötti szinkronizálási akadály. Egyetlen szuperstepben az összes aktivált végrehajtó párhuzamosan fut, de a munkafolyamat nem halad tovább a következő szuperstepre, amíg minden végrehajtó be nem fejeződik.

Ez hatással van a fan-out mintákra: ha több útvonalra ágazik ki — az egyik egy végrehajtói lánccal, a másik pedig egyetlen hosszú ideig futó végrehajtóval — a láncolt útvonal nem haladhat előre, amíg a hosszú ideig futó végrehajtó be nem fejeződik.

Miért szupersteps?

A BSP-modell fontos garanciákat biztosít:

  • Determinisztikus végrehajtás: Ugyanazt a bemenetet figyelembe véve a munkafolyamat mindig ugyanabban a sorrendben fut
  • Megbízható ellenőrzőpont-kezelés: Az állapot mentése lehetséges szuperlépés határokon a hibatűrés érdekében
  • Egyszerűbb érvelés: Nincsenek versenyfeltételek a szupersztepsek között; mindegyik az üzenetek konzisztens nézetét látja

A Superstep-modell használata

Ha valóban független párhuzamos útvonalakra van szüksége, amelyek nem blokkolják egymást, konszolidálja a szekvenciális lépéseket egyetlen végrehajtóba. A logikai láncolás helyett kombinálja a logikát egy végrehajtóval. Ezután mindkét párhuzamos útvonal egyetlen szupersztepjelen belül fut.

Következő lépések

Ügynökök a munkafolyamatokban

Kapcsolódó témakörök:

  • Last updated on