Fluxos de trabalho do Microsoft Agent Framework – Isolamento de Estado

Em aplicativos reais, o gerenciamento adequado do estado é fundamental ao lidar com várias tarefas ou solicitações. Sem isolamento adequado, o estado compartilhado entre diferentes execuções de fluxo de trabalho pode levar a comportamentos inesperados, corrupção de dados e condições de concorrência. Este artigo explica como garantir o isolamento de estado nos fluxos de trabalho do Microsoft Agent Framework, fornecendo insights sobre práticas recomendadas e armadilhas comuns.

Construtores de Fluxo de Trabalho Mutáveis versus Fluxos de Trabalho Imutáveis

Os fluxos de trabalho são criados por construtores de fluxo de trabalho. Os construtores de fluxo de trabalho geralmente são considerados mutáveis, em que é possível adicionar, modificar o executor inicial ou outras configurações após a criação do construtor ou até mesmo após a criação de um fluxo de trabalho. Por outro lado, os fluxos de trabalho são imutáveis porque, depois que um fluxo de trabalho é criado, ele não pode ser modificado (nenhuma API pública para modificar um fluxo de trabalho).

Essa distinção é importante porque afeta a forma como o estado é gerenciado em diferentes execuções de fluxo de trabalho. Não é recomendável reutilizar uma única instância de fluxo de trabalho para várias tarefas ou solicitações, pois isso pode levar ao compartilhamento de estado não intencional. Em vez disso, é recomendável criar uma nova instância de fluxo de trabalho do construtor para cada tarefa ou solicitação para garantir o isolamento de estado adequado e a segurança de thread.

Garantindo o isolamento de estado em construtores de fluxo de trabalho

Quando uma instância de executor é passada diretamente para um construtor de fluxo de trabalho, essa instância do executor é compartilhada entre todas as instâncias de fluxo de trabalho criadas do construtor. Isso pode levar a problemas se a instância do executor contiver um estado que não deve ser compartilhado entre várias execuções de fluxo de trabalho. Para garantir o isolamento de estado adequado e a segurança de thread, é recomendável usar funções de fábrica que criam uma nova instância de executor para cada instância de fluxo de trabalho.

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

Gerenciamento de Estado do Agente

O contexto do agente é gerenciado por meio de threads de agente. Por padrão, cada agente em um fluxo de trabalho obterá seu próprio thread, a menos que o agente seja gerenciado por um executor personalizado. Para obter mais informações, consulte Como trabalhar com agentes.

Os threads do agente são persistidos em execuções de fluxo de trabalho. Isso significa que, se um agente for invocado na primeira execução de um fluxo de trabalho, o conteúdo gerado pelo agente estará disponível em execuções subsequentes da mesma instância de fluxo de trabalho. Embora isso possa ser útil para manter a continuidade em uma única tarefa, ele também pode levar ao compartilhamento de estado não intencional se a mesma instância de fluxo de trabalho for reutilizado para diferentes tarefas ou solicitações. Para garantir que cada tarefa tenha estado de agente isolado, use as funções de fábrica do agente no construtor de fluxo de trabalho para criar uma nova instância de fluxo de trabalho para cada tarefa ou solicitação.

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

Conclusion

O isolamento de estado nos fluxos de trabalho do Microsoft Agent Framework pode ser efetivamente gerenciado usando funções de fábrica com construtores de fluxo de trabalho para criar novas instâncias de executor e agente. Ao criar novas instâncias de fluxo de trabalho para cada tarefa ou solicitação, você pode manter o isolamento de estado adequado e evitar o compartilhamento de estado não intencional entre diferentes execuções de fluxo de trabalho.

Próximas etapas

• Saiba como visualizar fluxos de trabalho.