共用方式為

Microsoft Agent Framework Workflows - 狀態隔離

在實際應用中，妥善管理狀態在處理多項任務或請求時至關重要。 若沒有適當的隔離，不同工作流程執行間的共享狀態可能導致意外行為、資料損壞及競賽狀態。 本文說明如何在 Microsoft Agent Framework 工作流程中確保狀態隔離，並提供最佳實務與常見陷阱的見解。

可變工作流程建構器與不可變工作流程

工作流程是由工作流程建構者所建立。 工作流程建構器通常被視為可變的，使用者可以在建構器建立後，甚至在工作流程建置完成後，新增、修改啟動執行器或其他設定。 另一方面，工作流程是不可變的，因為一旦建立起工作流程，就無法被修改（沒有公開 API 可以修改工作流程）。

這個區分很重要，因為它會影響不同工作流程執行間狀態的管理方式。 不建議重複使用單一工作流程實例處理多個任務或請求，因為這可能導致非預期的狀態共享。 相反地，建議從建構器為每個任務或請求建立新的工作流程實例，以確保適當的狀態隔離與執行緒安全。

確保工作流程建構器中的狀態隔離

當執行器實例直接傳入工作流程建構器時，該執行器實例會被所有由建構者建立的工作流實例共享。 如果執行器實例包含不應該在多個工作流程執行間共享的狀態，可能會產生問題。 為確保適當的狀態隔離與執行緒安全，建議使用工廠函式為每個工作流程實例建立新的執行器實例。

即將推出...

非執行緒安全的範例：

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

小提示

為確保適當的狀態隔離與執行緒安全，也請確保由工廠函式建立的執行器實例不共享可變狀態。

代理狀態管理

代理上下文是透過代理執行緒來管理的。 預設情況下，工作流程中的每個代理都會獲得自己的執行緒，除非代理由客製執行者管理。 欲了解更多資訊，請參閱 「與代理人合作」。

代理執行緒會在工作流程執行中持續存在。 這表示如果在工作流程的第一次執行中呼叫代理，該代理產生的內容會在同一工作流程實例的後續執行中可用。 雖然這有助於維持單一任務的連續性，但若同一工作流程實例被用於不同任務或請求，也可能造成非預期的狀態共享。 為了確保每個任務都有獨立的代理狀態，可以在工作流程建構器中使用代理工廠函式，為每個任務或請求建立新的工作流程實例。

即將推出...

非執行緒安全範例：

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

Microsoft 代理框架工作流程中的狀態隔離可透過使用工廠函式與工作流程建構器來有效管理，建立全新的執行器與代理實例。 透過為每個任務或請求建立新的工作流程實例，你可以維持適當的狀態隔離，避免不同工作流程執行間的非預期狀態共享。

後續步驟

  • Last updated on