Eksekutor yang Dapat Diatur Ulang

Overview

Pelaksana dalam alur kerja sering kali bersifat stateful; misalnya, mereka dapat mengakumulasi pesan, melacak jumlah giliran, atau men-cache hasil perantara. Saat alur kerja digunakan kembali di beberapa eksekusi dengan instans pelaksana bersama, status sisa dari eksekusi sebelumnya dapat bocor ke eksekusi berikutnya, menyebabkan perilaku yang tidak terduga atau kerusakan data.

Antarmuka IResettableExecutor menyelesaikan ini dengan menyediakan kontrak bagi pelaksana untuk menghapus status internal mereka di antara eksekusi. Runtime alur kerja secara otomatis memanggil instans pelaksana bersama ResetAsync() saat proses selesai, memastikan kondisi awal yang bersih untuk proses berikutnya.

Masalah

Pertimbangkan pelaksana yang mengumpulkan pesan selama alur kerja berjalan:

internal sealed partial class AggregationExecutor() : Executor("AggregationExecutor")
{
    private readonly List<string> _messages = [];

    [MessageHandler]
    private async ValueTask HandleAsync(string message, IWorkflowContext context)
    {
        this._messages.Add(message);
        // Process aggregated messages...
    }
}

Jika pengelola ini dibagi dalam berbagai alur kerja, _messages menyimpan data dari jalankan sebelumnya. Pada eksekusi kedua, akan terdapat pesan usang yang tidak relevan.

Antarmuka IResettableExecutor

IResettableExecutor menentukan metode tunggal yang dipanggil runtime alur kerja di antara eksekusi:

public interface IResettableExecutor
{
    ValueTask ResetAsync();
}

Ketika pelaksana mengimplementasikan antarmuka ini, runtime dapat mengatur ulang dengan aman setelah setiap eksekusi, memungkinkan alur kerja digunakan kembali tanpa status basi.

Menerapkan IResettableExecutor

Untuk membuat eksekutor stateful agar dapat diatur ulang, terapkan antarmuka dan hapus semua state yang dapat diubah di ResetAsync():

internal sealed partial class AggregationExecutor()
    : Executor("AggregationExecutor"), IResettableExecutor
{
    private readonly List<string> _messages = [];

    [MessageHandler]
    private async ValueTask HandleAsync(string message, IWorkflowContext context)
    {
        this._messages.Add(message);
        // Process aggregated messages...
    }

    public ValueTask ResetAsync()
    {
        this._messages.Clear();
        return default;
    }
}

Untuk contoh kerja lengkap alur kerja yang menggunakan eksekutor yang dapat diatur ulang, lihat sampel WorkflowAsAnAgent.

Kapan harus Menerapkan

Tidak semua pelaksana perlu menerapkan IResettableExecutor. Gunakan panduan keputusan ini:

Skenario Menerapkan? Alasan
Eksekutor memiliki status yang dapat diubah (daftar, penghitung, cache) dan digunakan bersama dalam setiap eksekusi. Yes Status dari satu eksekusi akan bocor ke eksekusi berikutnya
Pelaksana tidak memiliki status No Tidak ada yang perlu direset
Eksekutor dibuat baru per alur kerja (melalui metode pembuat) No Setiap eksekusi mendapatkan instans baru dengan status bersih
Eksekutor dinyatakan sebagai dapat dibagikan antar sesi (declareCrossRunShareable: true) No Eksekutor yang dapat dibagikan antar-jalankan mendukung penggunaan secara bersamaan tanpa perlu reset ulang

Warning

Jika pelaksana stateful bersama tidak menerapkan IResettableExecutor, maka menggunakan kembali alur kerja akan menghasilkan error berupa InvalidOperationException.

"Cannot reuse Workflow with shared Executor instances that do not implement IResettableExecutor."

Cara Runtime Menggunakannya

Runtime alur kerja mengelola siklus hidup reset secara otomatis. Anda tidak perlu memanggil ResetAsync() sendiri. Urutannya adalah:

  1. Kepemilikan diperoleh — saat alur kerja berjalan dimulai, runtime mengambil kepemilikan instans alur kerja dan mencatat pelaksana mana yang perlu direset.
  2. Jalankan eksekusi — eksekutor memproses pesan dan dapat mengakumulasi status.
  3. Kepemilikan dirilis — ketika eksekusi selesai (atau dibuang), runtime melepaskan kepemilikan dan panggilan ResetAsync() pada semua instans pelaksana bersama yang mengimplementasikan IResettableExecutor.
  4. Siap digunakan kembali — setelah reset berhasil, alur kerja dapat digunakan untuk eksekusi baru.

Jika ada pelaksana bersama yang gagal mengatur ulang (karena tidak mengimplementasikan antarmuka), alur kerja ditandai sebagai tidak dapat digunakan kembali dan eksekusi berikutnya akan dilemparkan.

Hubungan dengan Isolasi Status

IResettableExecutor melengkapi pola metode pembantu yang dijelaskan dalam Manajemen Status. Dua pendekatan tersebut melayani kebutuhan yang berbeda:

  • Metode pembantu (membuat instans baru per eksekusi) memberikan jaminan isolasi terkuat dan direkomendasikan sebagai pendekatan default.
  • IResettableExecutor berguna ketika Anda perlu berbagi instans pelaksana di seluruh eksekusi — misalnya, ketika konstruksi pelaksana mahal atau ketika alur kerja diekspos sebagai agen dan digunakan kembali di beberapa pemanggilan.

Pilih pendekatan yang paling sesuai dengan skenario Anda. Untuk sebagian besar alur kerja, metode pembantu sudah cukup. Gunakan IResettableExecutor saat berbagi instance adalah pilihan desain yang sengaja.

Konsep ini tidak berlaku untuk Python. Untuk isolasi status penuh, bangun alur kerja baru dan instans pelaksana untuk setiap eksekusi independen. Lihat Manajemen Status untuk pola dan contoh.

Eksekutor Go dapat mengatur ulang state lokal bersama dengan menyertakan ResetFunc pada workflow.Executor. Binding yang dibuat menggunakan workflow.BindNewExecutorFunc membuat eksekutor baru untuk setiap sesi alur kerja dan biasanya tidak memerlukan hook reset.

var count int

counter := workflow.NewExecutor("Counter", func(input string) int {
    count++
    return count
}).Extend(&workflow.Executor{
    ResetFunc: func() error {
        count = 0
        return nil
    },
}).Bind()

Langkah berikutnya

Manajemen Negara

  • Last updated on