Ejecutores restablecibles

Visión general

Los ejecutores de flujos de trabajo suelen tener estado; por ejemplo, pueden acumular mensajes, realizar un seguimiento de los recuentos de turnos o almacenar en caché resultados intermedios. Cuando se reutiliza un flujo de trabajo en varias ejecuciones con instancias compartidas del ejecutor, el estado restante de una ejecución anterior puede filtrarse en ejecuciones posteriores, lo que provoca un comportamiento inesperado o daños en los datos.

La IResettableExecutor interfaz resuelve esto proporcionando un contrato para que los ejecutores borren su estado interno entre ejecuciones. El tiempo de ejecución del flujo de trabajo llama ResetAsync() automáticamente a las instancias del ejecutor compartido cuando se completa una ejecución, lo que garantiza una pizarra limpia para la siguiente ejecución.

El problema

Considere un ejecutor que recopile mensajes durante una ejecución de flujo de trabajo:

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

Si este ejecutor se comparte entre ejecuciones de flujo de trabajo, _messages conserva los datos de la ejecución anterior. La segunda ejecución vería mensajes obsoletos que no pertenecen a ella.

Interfaz IResettableExecutor

IResettableExecutor define un único método al que llama el tiempo de ejecución del flujo de trabajo entre ejecuciones:

public interface IResettableExecutor
{
    ValueTask ResetAsync();
}

Cuando un ejecutor implementa esta interfaz, el tiempo de ejecución puede restablecerlo de forma segura después de cada ejecución, lo que permite reutilizar el flujo de trabajo sin estado obsoleto.

Implementación de IResettableExecutor

Para hacer que un ejecutor con estado sea restablecible, implemente la interfaz y borre todo el estado mutable en 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;
    }
}

Para obtener un ejemplo de trabajo completo de un flujo de trabajo que usa ejecutores resettables, consulte el ejemplo WorkflowAsAnAgent.

Cuándo implementar

No todos los ejecutores deben implementar IResettableExecutor. Use esta guía de decisión:

Escenario	¿Implementar?	Motivo
El ejecutor tiene un estado mutable (listas, contadores, cachés) y se comparte entre ejecuciones.	Sí	El estado de una ejecución se filtraría al siguiente
El ejecutor es sin estado	No	No hay nada que restablecer
El ejecutor se crea de nuevo por flujo de trabajo (mediante un método de fábrica)	No	Cada ejecución obtiene una nueva instancia con estado limpio
Executor se declara como compartido entre ejecución (declareCrossRunShareable: true)	No	Los ejecutores compartibles entre ejecuciones admiten el uso concurrente sin restablecer

Advertencia

Si un ejecutor con estado compartido no implementa IResettableExecutor, volver a usar el flujo de trabajo arroja una excepción InvalidOperationException

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

Cómo lo usa el runtime

El tiempo de ejecución del flujo de trabajo administra automáticamente el ciclo de vida del reinicio. No necesitas llamar a ResetAsync() tú mismo. La secuencia es:

1. Adquisición de propiedad — cuando se inicia una ejecución de flujo de trabajo, el entorno de ejecución toma posesión de la instancia de flujo de trabajo y determina qué ejecutores necesitan ser restablecidos.
2. Run se ejecuta : los ejecutores procesan mensajes y pueden acumular el estado.
3. Propiedad liberada — cuando la ejecución se completa (o se elimina), el runtime libera la propiedad y llama a ResetAsync() en todas las instancias de ejecutor compartidas que implementan IResettableExecutor.
4. Listo para su reutilización : después de un restablecimiento correcto, el flujo de trabajo se puede usar para una nueva ejecución.

Si un ejecutor compartido no se puede restablecer (porque no implementa la interfaz), el flujo de trabajo se marca como no reutilizable y se producirán ejecuciones posteriores.

Relación con el aislamiento de estado

IResettableExecutor complementa el patrón de método auxiliar descrito en Gestión de estados. Los dos enfoques satisfacen necesidades diferentes:

• Los métodos auxiliares (creación de instancias nuevas por ejecución) proporcionan las garantías de aislamiento más seguras y se recomiendan como enfoque predeterminado.
• IResettableExecutor resulta útil cuando necesita compartir instancias del ejecutor entre ejecuciones; por ejemplo, cuando la construcción del ejecutor es costosa o cuando un flujo de trabajo se expone como agente y se reutiliza en varias invocaciones.

Elija el enfoque que mejor se adapte a su escenario. Para la mayoría de los flujos de trabajo, los métodos auxiliares son suficientes. Utiliza IResettableExecutor cuando compartir instancias es una elección de diseño deliberada.

Este concepto no se aplica a Python. Para el aislamiento de estado completo, cree nuevas instancias de flujo de trabajo y ejecutor para cada ejecución independiente. Consulte Administración de estado para obtener patrones y ejemplos.

Pasos siguientes

Administración de estados