Microsoft 代理框架工作流程 - 狀態

本文件概述了 Microsoft 代理框架工作流程系統中的狀態。

概觀

State 允許工作流程中的多個執行者存取並修改共用資料。此功能對於工作流程的不同部分需要共享資訊且直接訊息傳遞不可行或不高效的場景至關重要。

致州政府的書信

using Microsoft.Agents.AI.Workflows;

internal sealed partial class FileReadExecutor(): Executor("FileReadExecutor")
{
    /// <summary>
    /// Reads a file and stores its content in a shared state.
    /// </summary>
    /// <param name="message">The path to the embedded resource file.</param>
    /// <param name="context">The workflow context for accessing shared states.</param>
    /// <returns>The ID of the shared state where the file content is stored.</returns>
    [MessageHandler]
    private async ValueTask<string> HandleAsync(string message, IWorkflowContext context)
    {
        // Read file content from embedded resource
        string fileContent = File.ReadAllText(message);
        // Store file content in a shared state for access by other executors
        string fileID = Guid.NewGuid().ToString();
        await context.QueueStateUpdateAsync<string>(fileID, fileContent, scopeName: "FileContent");

        return fileID;
    }
}

from agent_framework import (
    Executor,
    WorkflowContext,
    handler,
)

class FileReadExecutor(Executor):

    @handler
    async def handle(self, file_path: str, ctx: WorkflowContext[str]):
        # Read file content from embedded resource
        with open(file_path, 'r') as file:
            file_content = file.read()
        # Store file content in state for access by other executors
        file_id = str(uuid.uuid4())
        ctx.set_state(file_id, file_content)

        await ctx.send_message(file_id)

存取狀態

using Microsoft.Agents.AI.Workflows;

internal sealed partial class WordCountingExecutor() : Executor("WordCountingExecutor")
{
    /// <summary>
    /// Counts the number of words in the file content stored in a shared state.
    /// </summary>
    /// <param name="message">The ID of the shared state containing the file content.</param>
    /// <param name="context">The workflow context for accessing shared states.</param>
    /// <returns>The number of words in the file content.</returns>
    [MessageHandler]
    private async ValueTask<int> HandleAsync(string message, IWorkflowContext context)
    {
        // Retrieve the file content from the shared state
        var fileContent = await context.ReadStateAsync<string>(message, scopeName: "FileContent")
            ?? throw new InvalidOperationException("File content state not found");

        return fileContent.Split([' ', '\n', '\r'], StringSplitOptions.RemoveEmptyEntries).Length;
    }
}

from agent_framework import (
    Executor,
    WorkflowContext,
    handler,
)

class WordCountingExecutor(Executor):

    @handler
    async def handle(self, file_id: str, ctx: WorkflowContext[int]):
        # Retrieve the file content from state
        file_content = ctx.get_state(file_id)
        if file_content is None:
            raise ValueError("File content state not found")

        await ctx.send_message(len(file_content.split()))

州隔離

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

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

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

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

使用輔助方法確保狀態隔離

當執行器實例只建立一次並在多個工作流程建置間共享時，其內部狀態會被所有工作流程執行共享。如果執行器包含應在工作流程中隔離的可變狀態，可能會產生問題。為確保狀態隔離與執行緒安全，將執行執行器實例化與工作流程建置封裝於輔助方法中，使每次呼叫都能產生全新且獨立的實例。

即將推出...

非孤立範例（共享狀態）：

executor_a = CustomExecutorA()
executor_b = CustomExecutorB()

# executor_a and executor_b are shared across all workflows built from this builder
workflow_builder = WorkflowBuilder(start_executor=executor_a).add_edge(executor_a, executor_b)

workflow_a = workflow_builder.build()
workflow_b = workflow_builder.build()
# workflow_a and workflow_b share the same executor instances and their mutable state

孤立的例子（輔助方法）：

def create_workflow() -> Workflow:
    """Create a fresh workflow with isolated state.

    Each call produces independent executor instances, ensuring no state
    leaks between workflow runs.
    """
    executor_a = CustomExecutorA()
    executor_b = CustomExecutorB()

    return WorkflowBuilder(start_executor=executor_a).add_edge(executor_a, executor_b).build()

# Each workflow has its own executor instances with independent state
workflow_a = create_workflow()
workflow_b = create_workflow()

小提示

為確保適當的狀態隔離與執行緒安全，也請確保在輔助方法內建立的執行者實例不會共享外部可變狀態。

代理狀態管理

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

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

即將推出...

非孤立範例（共享代理人狀態）：

writer_agent = AzureOpenAIChatClient(credential=AzureCliCredential()).as_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()).as_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",
)

# writer_agent and reviewer_agent are shared across all workflows
workflow = WorkflowBuilder(start_executor=writer_agent).add_edge(writer_agent, reviewer_agent).build()

孤立的例子（輔助方法）：

def create_workflow() -> Workflow:
    """Create a fresh workflow with isolated agent state.

    Each call produces new agent instances with their own threads,
    ensuring no conversation history leaks between workflow runs.
    """
    writer_agent = AzureOpenAIChatClient(credential=AzureCliCredential()).as_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()).as_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",
    )

    return WorkflowBuilder(start_executor=writer_agent).add_edge(writer_agent, reviewer_agent).build()

# Each workflow has its own agent instances and threads
workflow_a = create_workflow()
workflow_b = create_workflow()

總結

Microsoft 代理框架工作流程中的狀態隔離可以透過包裝執行器與代理實例化，以及在輔助方法中建置工作流程來有效管理。每次需要新工作流程時呼叫輔助方法，確保每個實例都有新鮮且獨立的狀態，避免不同工作流程執行間的非預期狀態共享。

後續步驟

意見反應

此頁面對您有幫助嗎？

Last updated on 2026-02-13