Zurücksetzbare Ausführungsdienste

Übersicht

Ausführungen in Workflows sind häufig zustandsbehaftet – z. B. können sie Nachrichten ansammeln, die Anzahl der Durchläufe nachverfolgen oder Zwischenergebnisse zwischenspeichern. Wenn ein Workflow für mehrere Ausführungen mit freigegebenen Ausführungsinstanzen wiederverwendet wird, kann der Überlaufzustand einer vorherigen Ausführung in nachfolgende Ausführungsläufe fließen, was zu unerwartetem Verhalten oder Datenbeschädigung führt.

Die IResettableExecutor Schnittstelle löst dies, indem sie einen Vertrag für die Ausführungsinstanzen bereitstellt, um ihren internen Zustand zwischen den Läufen zu löschen. Die Workflowlaufzeit ruft ResetAsync() automatisch freigegebene Executorinstanzen auf, wenn eine Ausführung abgeschlossen ist, und stellt sicher, dass die nächste Ausführung sauber ausgeführt wird.

Das Problem

Erwägen Sie einen Executor, der Nachrichten während einer Workflowausführung sammelt:

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...
    }
}

Wenn dieser Executor für alle Workflowausführungen freigegeben wird, _messages werden Daten aus der vorherigen Ausführung beibehalten. Die zweite Ausführung würde veraltete Nachrichten sehen, die nicht zu ihr gehören.

Die IResettableExecutor-Schnittstelle

IResettableExecutor definiert eine einzelne Methode, die von der Workflow-Laufzeit zwischen den Ausführungen aufgerufen wird.

public interface IResettableExecutor
{
    ValueTask ResetAsync();
}

Wenn ein Executor diese Schnittstelle implementiert, kann die Laufzeit sie nach jeder Ausführung sicher zurücksetzen, sodass der Workflow ohne veralteten Zustand wiederverwendet werden kann.

Implementieren von IResettableExecutor

Um einen zustandsbehafteten Executor zurücksetzbar zu machen, implementieren Sie die Schnittstelle und löschen Sie den gesamten veränderlichen Zustand in 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;
    }
}

Ein vollständiges Arbeitsbeispiel für einen Workflow, der resettable executors verwendet, finden Sie im WorkflowAsAnAgent-Beispiel.

Wann muss ich implementieren?

Nicht alle Ausführungsmechanismen müssen IResettableExecutor implementieren. Verwenden Sie diesen Entscheidungsleitfaden:

Wie die Runtime sie verwendet

Die Workflow-Laufzeit managt den Zurücksetzungslebenszyklus automatisch. Sie müssen ResetAsync() nicht selbst anrufen. Die Sequenz lautet:

1. Eigentum übernommen – wenn ein Workflow ausgeführt wird, übernimmt die Laufzeit die Kontrolle über die Workflow-Instanz und vermerkt, welche Executor zurückgesetzt werden müssen.
2. Ausführung gestartet – Ausführende verarbeiten Nachrichten und können den Zustand ansammeln.
3. Besitz freigegeben – wenn die Ausführung endet (oder verworfen wird), gibt die Laufzeit den Besitz frei und ruft ResetAsync() auf allen freigegebenen Executor-Instanzen auf, die IResettableExecutor implementieren.
4. Bereit für die Wiederverwendung – nach einem erfolgreichen Zurücksetzen kann der Workflow für eine neue Ausführung verwendet werden.

Wenn ein freigegebener Executor nicht zurückgesetzt werden kann (da er die Schnittstelle nicht implementiert), wird der Workflow als nicht wiederverwendbar gekennzeichnet und nachfolgende Ausführung wird ausgelöst.

Beziehung zur Status-Isolation

IResettableExecutor ergänzt das in der Zustandsverwaltung beschriebene Hilfsmethodenmuster. Die beiden Ansätze dienen unterschiedlichen Bedürfnissen:

• Hilfsmethoden (Erstellen neuer Instanzen pro Ausführung) bieten die stärksten Isolationsgarantien und werden als Standardansatz empfohlen.
• IResettableExecutor ist nützlich, wenn Sie ausführungsübergreifende Ausführungsinstanzen freigeben müssen , z. B. wenn die Ausführung des Ausführungsvorgangs teuer ist oder wenn ein Workflow als Agent verfügbar gemacht und über mehrere Aufrufe hinweg wiederverwendet wird.

Wählen Sie den Ansatz aus, der am besten zu Ihrem Szenario passt. Für die meisten Workflows sind Hilfsmethoden ausreichend. Verwenden Sie IResettableExecutor, wenn die Instanzenfreigabe eine absichtliche Entwurfsentscheidung ist.

Dieses Konzept gilt nicht für Python. Erstellen Sie für die vollständige Statusisolation neue Workflow- und Executorinstanzen für jede unabhängige Ausführung. Muster und Beispiele finden Sie unter "Zustandsverwaltung ".