Flujos de trabajo de Microsoft Agent Framework: aislamiento de estado

En las aplicaciones del mundo real, administrar correctamente el estado es fundamental al controlar varias tareas o solicitudes. Sin el aislamiento adecuado, el estado compartido entre distintas ejecuciones de flujo de trabajo puede provocar un comportamiento inesperado, daños en los datos y condiciones de carrera. En este artículo se explica cómo garantizar el aislamiento de estado en los flujos de trabajo de Microsoft Agent Framework, lo que proporciona información sobre los procedimientos recomendados y los problemas comunes.

Generadores de flujos de trabajo mutables frente a flujos de trabajo inmutables

Los generadores de flujos de trabajo crean flujos de trabajo. Por lo general, los generadores de flujos de trabajo se consideran mutables, donde se puede agregar, modificar el ejecutor de inicio u otras configuraciones después de crear el generador o incluso después de compilar un flujo de trabajo. Por otro lado, los flujos de trabajo son inmutables en que una vez compilado un flujo de trabajo, no se puede modificar (ninguna API pública para modificar un flujo de trabajo).

Esta distinción es importante porque afecta a cómo se administra el estado en distintas ejecuciones de flujo de trabajo. No se recomienda reutilizar una sola instancia de flujo de trabajo para varias tareas o solicitudes, ya que esto puede provocar un uso compartido de estado no deseado. En su lugar, se recomienda crear una nueva instancia de flujo de trabajo desde el generador para cada tarea o solicitud para garantizar el aislamiento de estado adecuado y la seguridad de subprocesos.

Garantizar el aislamiento de estado en los generadores de flujos de trabajo

Cuando una instancia del ejecutor se pasa directamente a un generador de flujos de trabajo, esa instancia del ejecutor se comparte entre todas las instancias de flujo de trabajo creadas a partir del generador. Esto puede provocar problemas si la instancia del ejecutor contiene el estado que no se debe compartir entre varias ejecuciones de flujo de trabajo. Para garantizar el aislamiento de estado adecuado y la seguridad de subprocesos, se recomienda usar funciones de fábrica que creen una nueva instancia del ejecutor para cada instancia de flujo de trabajo.

executor_a = CustomExecutorA()
executor_b = CustomExecutorB()

workflow_builder = WorkflowBuilder()
# executor_a and executor_b are passed directly to the workflow builder
workflow_builder.add_edge(executor_a, executor_b)
workflow_builder.set_start_executor(executor_b)

# All workflow instances created from the builder will share the same executor instances
workflow_a = workflow_builder.build()
workflow_b = workflow_builder.build()

workflow_builder = WorkflowBuilder()
# Register executor factory functions with the workflow builder
workflow_builder.register_executor(factory_func=CustomExecutorA, name="executor_a")
workflow_builder.register_executor(factory_func=CustomExecutorB, name="executor_b")
# Add edges using registered factory function names
workflow_builder.add_edge("executor_a", "executor_b")
workflow_builder.set_start_executor("executor_b")

# Each workflow instance created from the builder will have its own executor instances
workflow_a = workflow_builder.build()
workflow_b = workflow_builder.build()

Administración del estado del agente

El contexto del agente se administra a través de subprocesos del agente. De forma predeterminada, cada agente de un flujo de trabajo obtendrá su propio subproceso a menos que un ejecutor personalizado administre el agente. Para obtener más información, consulte Trabajar con agentes.

Los subprocesos del agente se conservan en las ejecuciones de flujo de trabajo. Esto significa que si se invoca un agente en la primera ejecución de un flujo de trabajo, el contenido generado por el agente estará disponible en ejecuciones posteriores de la misma instancia de flujo de trabajo. Aunque esto puede ser útil para mantener la continuidad dentro de una sola tarea, también puede dar lugar a un uso compartido de estado no deseado si se reutiliza la misma instancia de flujo de trabajo para diferentes tareas o solicitudes. Para asegurarse de que cada tarea tiene el estado del agente aislado, use las funciones de generador de agentes en el generador de flujos de trabajo para crear una nueva instancia de flujo de trabajo para cada tarea o solicitud.

writer_agent = AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
    instructions=(
        "You are an excellent content writer. You create new content and edit contents based on the feedback."
    ),
    name="writer_agent",
)
reviewer_agent = AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
    instructions=(
        "You are an excellent content reviewer."
        "Provide actionable feedback to the writer about the provided content."
        "Provide the feedback in the most concise manner possible."
    ),
    name="reviewer_agent",
)

builder = WorkflowBuilder()
# writer_agent and reviewer_agent are passed directly to the workflow builder
builder.add_edge(writer_agent, reviewer_agent)
builder.set_start_executor(writer_agent)

# All workflow instances created from the builder will share the same agent
# instances and agent threads
workflow = builder.build()

def create_writer_agent() -> ChatAgent:
    """Factory function to create a Writer agent."""
    return AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
        instructions=(
            "You are an excellent content writer. You create new content and edit contents based on the feedback."
        ),
        name="writer_agent",
    )

def create_reviewer_agent() -> ChatAgent:
    """Factory function to create a Reviewer agent."""
    return AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
        instructions=(
            "You are an excellent content reviewer."
            "Provide actionable feedback to the writer about the provided content."
            "Provide the feedback in the most concise manner possible."
        ),
        name="reviewer_agent",
    )

builder = WorkflowBuilder()
# Register agent factory functions with the workflow builder
builder.register_agent(factory_func=create_writer_agent, name="writer_agent")
builder.register_agent(factory_func=create_reviewer_agent, name="reviewer_agent")
# Add edges using registered factory function names
builder.add_edge("writer_agent", "reviewer_agent")
builder.set_start_executor("writer_agent")

# Each workflow instance created from the builder will have its own agent
# instances and agent threads
workflow = builder.build()

Conclusión

El aislamiento de estado en los flujos de trabajo de Microsoft Agent Framework se puede administrar de forma eficaz mediante funciones de fábrica con generadores de flujos de trabajo para crear instancias de agente y ejecutor nuevas. Al crear nuevas instancias de flujo de trabajo para cada tarea o solicitud, puede mantener el aislamiento de estado adecuado y evitar el uso compartido de estado no deseado entre distintas ejecuciones de flujo de trabajo.

Pasos siguientes

• Aprenda a visualizar flujos de trabajo.