Pembuat & Eksekusi Alur Kerja

Alur Kerja mengikat eksekutor dan tepi bersama-sama ke dalam grafik yang diarahkan dan mengelola eksekusi. Ini mengoordinasikan pemanggilan pelaksana, perutean pesan, dan streaming peristiwa.

Membangun Alur Kerja

Alur kerja dibangun menggunakan WorkflowBuilder kelas , yang menyediakan API yang fasih untuk menentukan struktur alur kerja:

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

Alur kerja dibangun menggunakan WorkflowBuilder kelas :

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

Selama validasi build-time, handler dapat menggunakan anotasi kontainer polos, seperti list, dict, tuple, atau Sequence. Ini dapat menerima output yang menggunakan kontainer berparameter yang sesuai, seperti list[str].

Paket ini workflow menyediakan model eksekusi berbasis grafik di mana pelaksana dihubungkan oleh tepi.

  • Pelaksana - Unit pemrosesan yang menerima input dan menghasilkan output
  • Edge - Menyambungkan output dari satu eksekutor ke input pelaksana lain
  • Builder - Membangun alur kerja dengan menentukan pelaksana dan tepi
  • Jalankan - Menjalankan alur kerja dengan input yang diberikan

Fungsi ini membangun alur kerja pemrosesan teks tanpa memerlukan model atau kredensial:

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) mengekspos hasil eksekutor reverse sebagai output alur kerja. Edge masih memberikan hasil huruf besar ke eksekutor reverse terlepas dari penunjukan output.

Teruskan alur kerja yang dikembalikan oleh buildWorkflow ke salah satu dari dua fungsi eksekusi di bawah ini. Untuk input "Hello, World!", keduanya mencetak Workflow output: !DLROW ,OLLEH.

Eksekusi Alur Kerja

Alur kerja mendukung mode eksekusi streaming dan non-streaming:

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

Gunakan RunStreaming saat Anda menginginkan event saat event tersebut terjadi. Teruskan konteks pemanggil sehingga pembatalan mencapai workflow, periksa kesalahannya setelah iterasi berakhir, dan tutup run saat Anda selesai:

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

Menunggu sementara permintaan eksternal masih tertunda. Untuk alur kerja yang memerlukan input manusia, tangani peristiwa permintaan atau gunakan WatchUntilHalt untuk kembali saat dijeda. Lihat Event alur kerja.

Gunakan Run untuk melanjutkan ke halt berikutnya lalu periksa peristiwa yang dikumpulkan. Eksekusi dapat berhenti saat menunggu input eksternal, tidak hanya ketika pekerjaannya selesai:

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() memajukan bookmark-nya setiap kali Anda memanggilnya. Perulangan kedua menggunakan OutgoingEvents() untuk memutar ulang akumulasi peristiwa eksekutor alih-alih mencoba mengonsumsinya lagi dengan NewEvents().

Fungsi-fungsi ini mengakhiri proses eksekusinya setelah mengamati hasilnya. Biarkan run tetap terbuka jika aplikasi Anda perlu mengirim lebih banyak pesan atau melanjutkannya setelah jeda.

Lihat Event alur kerja untuk konsumen lengkap, klasifikasi output, dan semantik jeda.

Tip

Lihat contoh alur kerja untuk sampel lengkap yang dapat dijalankan.

Validasi Alur Kerja

Penyusun alur kerja memvalidasi graf sebelum eksekusi. Pemeriksaan yang tepat tergantung pada SDK dan metadata jenis yang tersedia:

  • Kompatibilitas jenis: Memeriksa protokol pesan yang dideklarasikan antara eksekutor yang terhubung jika metadata tersedia.
  • Konektivitas graf: Memeriksa apakah executor terdaftar dapat dijangkau dari executor awal.
  • Pengikatan executor: Pemeriksaan apakah registrasi executor yang diperlukan ada.
  • Validasi edge: Memeriksa koneksi yang tidak valid dan edge duplikat sesuai dengan aturan builder.

Go Build() menolak pengikatan yang tidak terselesaikan, eksekutor terdaftar yang tidak dapat dijangkau, tepi tanpa syarat duplikat, dan jenis pesan yang dinyatakan tidak kompatibel. Rute bersyarat dan protokol dinamis masih memerlukan pengujian runtime.

Validasi jenis dapat membuat instans eksekutor sementara untuk memeriksa protokol mereka. Jika factory atau protokol tidak dapat diperiksa, pemeriksaan kompatibilitas tersebut akan dilewati. Build yang berhasil tidak menjamin bahwa pembuatan executor atau setiap pengiriman pesan berhasil saat runtime. Menangani kesalahan dari peristiwa pembuatan run, iterasi event, dan kegagalan alur kerja.

BuildWithOptions(workflow.BuildOptions{DisableOrphanValidation: true}) hanya menonaktifkan pemeriksaan orphan-executor. Validasi lain masih berjalan.

Model Eksekusi: Supersteps

Kerangka kerja menggunakan model eksekusi Pregel yang dimodifikasi - pendekatan Paralel Sinkron Massal (BSP) dengan pemrosesan berbasis superstep.

Cara Kerja Supersteps

Eksekusi alur kerja diatur menjadi "supersteps" diskrit. Setiap langkah super:

  1. Mengumpulkan semua pesan yang masih tertunda dari superstep sebelumnya
  2. Mengalihkan pesan ke pelaksana target berdasarkan definisi batas
  3. Menjalankan semua eksekutor target secara bersamaan dalam superstep
  4. Menunggu semua pelaksana selesai sebelum maju (hambatan sinkronisasi)
  5. Mengantre pesan baru apa pun yang dihasilkan oleh eksekutor untuk superstep berikutnya
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   │
└─────────────────┘    └─────────────────┘

Hambatan Sinkronisasi

Karakteristik yang paling penting adalah hambatan sinkronisasi antara supersteps. Pelaksana yang dipicu dapat berjalan bersamaan dalam superstep, tetapi alur kerja tidak maju ke superstep berikutnya sampai pelaksana yang dipicu tersebut selesai.

Ini memengaruhi pola fan-out: jika Anda membagi ke beberapa jalur — satu dengan rangkaian eksekutor dan satu lagi dengan eksekutor tunggal yang berjalan lama — jalur dengan rantai tidak dapat maju sampai eksekutor jangka panjang selesai.

Mengapa Supersteps?

Model BSP memberikan jaminan penting:

  • Batas langkah yang ditentukan: Pesan yang dipancarkan selama superstep diantrekan untuk diproses nanti.
  • Batas checkpoint: Alur kerja dengan checkpoint aktif menyimpan status eksekusi pada batas superstep.
  • Pembaruan status yang di-buffer: Perubahan status alur kerja yang diantrekan dilakukan di antara supersteps.

Jaminan ini tidak menjamin eksekusi atau pemesanan peristiwa di antara pelaksana bersamaan dalam satu superstep. Sinkronkan akses ke state bersama milik aplikasi daripada mengasumsikan penghalang superstep mencegah semua kondisi balapan.

Bekerja dengan Model Superstep

Jika Anda memerlukan jalur paralel yang benar-benar independen yang tidak saling memblokir, konsolidasikan langkah berurutan ke dalam satu pelaksana. Alih-alih menautkan, gabungkan logika tersebut menjadi satu eksekutor. Kedua jalur paralel kemudian dijalankan dalam satu superstep.

Langkah berikutnya

Agen dalam Alur Kerja

Topik terkait:

  • Terakhir diperbarui pada