Microsoft 代理框架工作流程 - 檢查點

本頁提供 Microsoft 代理框架工作流程系統中 檢查點的概述。

概觀

檢查點可讓您在工作流程執行期間的特定點儲存工作流程的狀態，並在稍後從這些點繼續。 此功能對於下列案例特別有用：

• 長時間執行的工作流程，您希望於失敗時避免遺失進度。
• 長時間執行的工作流程，您希望在之後的時間點暫停並恢復執行。
• 需要定期儲存狀態以進行稽核或合規目的的工作流程。
• 需要移轉跨不同環境或執行個體的工作流程。

檢查點何時建立？

請記得，工作流程是以 超步驟執行方式執行，這在 工作流程執行模型中有記載。 在每個超級步驟中的所有執行程式完成其執行之後，都會在每個超級步驟的結尾建立檢查點。 檢查點會擷取工作流程的整個狀態，包括：

• 所有執行程式的目前狀態
• 工作流程中下一個超級步驟的所有待處理訊息
• 擱置中的請求和回應
• 共用狀態

Note

從 Python 1.13.0 版本開始，工作流程還會在第一個超步驟前建立一個入口檢查點來記錄工作流程輸入，並在對請求事件的回應交付時建立另一個入口檢查點。 這些檢查點讓整個工作流程的執行變得可重複遊玩。 此版本包含對依賴迭代次數、訊息來源 ID 或檢查點排序的應用程式進行小幅破壞性變更。 現有的檢查站仍然受到支援。 有關遷移細節，請參見「升級 Python 工作流程檢查點至 1.13.0」。

佔領檢查點

要啟用檢查點，執行工作流程時必須提供 a CheckpointManager 。 檢查點可以透過 SuperStepCompletedEvent 存取，或是透過執行中的 Checkpoints 屬性進入。

using Microsoft.Agents.AI.Workflows;

// Create a checkpoint manager to manage checkpoints
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();

// Run the workflow with checkpointing enabled
StreamingRun run = await InProcessExecution
    .RunStreamingAsync(workflow, input, checkpointManager)
    .ConfigureAwait(false);
await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false))
{
    if (evt is SuperStepCompletedEvent superStepCompletedEvt)
    {
        // Access the checkpoint
        CheckpointInfo? checkpoint = superStepCompletedEvt.CompletionInfo?.Checkpoint;
    }
}

// Checkpoints can also be accessed from the run directly
IReadOnlyList<CheckpointInfo> checkpoints = run.Checkpoints;

要啟用檢查點，必須在建立工作流程時提供一個 CheckpointStorage。 然後可以通過儲存裝置存取檢查點。 Agent Framework 內建三種實作——請選擇符合你耐用性與部署需求的那一種：

提供者	Package	Durability	最適合用於
InMemoryCheckpointStorage	agent-framework	僅進行中	測試、示範、短暫的工作流程
FileCheckpointStorage	agent-framework	本地磁碟	單機器工作流程與本地開發
CosmosCheckpointStorage	agent-framework-azure-cosmos	Azure Cosmos DB	生產、分散式、跨流程工作流程

這三者都實作相同的 CheckpointStorage 協定，因此你可以在不改變工作流程或執行程式的情況下交換提供者。

內存中

InMemoryCheckpointStorage 將檢查點保留在程序記憶體中。 最適合用於測試、演示及不需跨重啟時保持持久性的短期工作流程。

from agent_framework import (
    InMemoryCheckpointStorage,
    WorkflowBuilder,
)

# Create a checkpoint storage to manage checkpoints
checkpoint_storage = InMemoryCheckpointStorage()

# Build a workflow with checkpointing enabled
builder = WorkflowBuilder(start_executor=start_executor, checkpoint_storage=checkpoint_storage)
builder.add_edge(start_executor, executor_b)
builder.add_edge(executor_b, executor_c)
builder.add_edge(executor_b, end_executor)
workflow = builder.build()

# Run the workflow
async for event in workflow.run(input, stream=True):
    ...

# Access checkpoints from the storage
checkpoints = await checkpoint_storage.list_checkpoints(workflow_name=workflow.name)

File

FileCheckpointStorage 將檢查點儲存到磁碟上的本地目錄。 最適合需要在流程重啟中存活的單機工作流程，以及本地開發。

from agent_framework import (
    FileCheckpointStorage,
    WorkflowBuilder,
)

# Create a checkpoint storage backed by a directory on disk.
# storage_path is required — there is no default directory.
checkpoint_storage = FileCheckpointStorage("/var/lib/agent-framework/checkpoints")

# Build a workflow with checkpointing enabled
builder = WorkflowBuilder(start_executor=start_executor, checkpoint_storage=checkpoint_storage)
builder.add_edge(start_executor, executor_b)
builder.add_edge(executor_b, executor_c)
builder.add_edge(executor_b, end_executor)
workflow = builder.build()

# Run the workflow
async for event in workflow.run(input, stream=True):
    ...

# Access checkpoints from the storage
checkpoints = await checkpoint_storage.list_checkpoints(workflow_name=workflow.name)

請參閱 Security 考量章節，了解如何限制哪些Python類型可透過 allowed_checkpoint_types 參數被反序列化。

Azure Cosmos DB

CosmosCheckpointStorage 會持續儲存檢查點到 Azure Cosmos DB NoSQL。 最適合需要穩定耐用的跨流程檢查點的生產和分散式工作流程。 安裝可選的提供者套件：

pip install agent-framework-azure-cosmos --pre

資料庫與容器在首次使用時自動建立，使用 /workflow_name 作為分區鍵，以便高效進行每個工作流程的查詢。 推薦的驗證模式是透過 Azure TokenCredential（如 DefaultAzureCredential）進行管理身份驗證與 RBAC：

from azure.identity.aio import DefaultAzureCredential
from agent_framework import WorkflowBuilder
from agent_framework_azure_cosmos import CosmosCheckpointStorage

# CosmosCheckpointStorage is an async context manager — it closes the underlying
# Cosmos client on exit when it created the client itself.
async with (
    DefaultAzureCredential() as credential,
    CosmosCheckpointStorage(
        endpoint="https://<account>.documents.azure.com:443/",
        credential=credential,
        database_name="agent-framework",
        container_name="workflow-checkpoints",
    ) as checkpoint_storage,
):
    # Build a workflow with checkpointing enabled
    builder = WorkflowBuilder(start_executor=start_executor, checkpoint_storage=checkpoint_storage)
    builder.add_edge(start_executor, executor_b)
    builder.add_edge(executor_b, executor_c)
    builder.add_edge(executor_b, end_executor)
    workflow = builder.build()

    # Run the workflow
    async for event in workflow.run(input, stream=True):
        ...

    # Access checkpoints from the storage
    checkpoints = await checkpoint_storage.list_checkpoints(workflow_name=workflow.name)

帳號金鑰認證也支援，透過直接傳遞金鑰作為 credential 參數：

from agent_framework_azure_cosmos import CosmosCheckpointStorage

checkpoint_storage = CosmosCheckpointStorage(
    endpoint="https://<account>.documents.azure.com:443/",
    credential="<your-account-key>",
    database_name="agent-framework",
    container_name="workflow-checkpoints",
)

連接細節也可完全透過環境變數提供：

變數	說明
AZURE_COSMOS_ENDPOINT	Cosmos DB 帳戶端點
AZURE_COSMOS_DATABASE_NAME	資料庫名稱
AZURE_COSMOS_CONTAINER_NAME	容器名稱
AZURE_COSMOS_KEY	帳戶金鑰（使用 Azure 憑證則可選）

CosmosCheckpointStorage 也接受透過 CosmosClient 預先建立的 cosmos_client= 或透過 ContainerProxy 預先建立的 container_client=，前提是你的應用程式已經在管理 Cosmos 用戶端的生命週期。

要啟用檢查點，請用檢查點管理器配置執行環境。 檢查點可從 workflow.SuperStepCompletedEvent或透過該行程的檢查點清單存取。

checkpointManager := checkpoint.NewInMemoryManager()

run, err := inproc.Default.
    WithCheckpointing(checkpointManager).
    RunStreaming(ctx, wf, input)
if err != nil {
    return err
}
defer run.Close(ctx)

var checkpoints []workflow.CheckpointInfo
for evt, err := range run.WatchStream(ctx) {
    if err != nil {
        return err
    }
    if completed, ok := evt.(workflow.SuperStepCompletedEvent); ok && completed.CompletionInfo != nil {
        if completed.CompletionInfo.CheckpointInfo != nil {
            checkpoints = append(checkpoints, *completed.CompletionInfo.CheckpointInfo)
        }
    }
}

// Checkpoints can also be accessed from the run directly.
checkpoints = run.Checkpoints()

從檢查點繼續

您可以直接在同一次執行中從特定檢查點繼續工作程序。

// Assume we want to resume from the 6th checkpoint
CheckpointInfo savedCheckpoint = run.Checkpoints[5];
// Restore the state directly on the same run instance.
await run.RestoreCheckpointAsync(savedCheckpoint).ConfigureAwait(false);
await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false))
{
    if (evt is WorkflowOutputEvent workflowOutputEvt)
    {
        Console.WriteLine($"Workflow completed with result: {workflowOutputEvt.Data}");
    }
}

您可以在相同的工作流程執行個體上從特定檢查點接續執行該工作流程。

# Assume we want to resume from the 6th checkpoint
saved_checkpoint = checkpoints[5]
async for event in workflow.run(checkpoint_id=saved_checkpoint.checkpoint_id, stream=True):
    ...

你可以直接在同一個執行作業中，將串流執行還原到特定檢查點。

// Assume we want to resume from the 6th checkpoint.
savedCheckpoint := checkpoints[5]
if err := run.RestoreCheckpoint(ctx, savedCheckpoint); err != nil {
    return err
}

for evt, err := range run.WatchStream(ctx) {
    if err != nil {
        return err
    }
    if outputEvent, ok := evt.(workflow.OutputEvent); ok {
        fmt.Printf("Workflow completed with result: %v\n", outputEvent.Output)
    }
}

從檢查站補水

重新水化的工作流程必須保留創建檢查點的工作流程的拓撲與執行者身份。 執行者身份的解析方式取決於 SDK 和執行者類型。

或者，您可以將工作流程從檢查點重新激活到新的運行實例。

// A rehydrated workflow must preserve the topology and executor identities of the workflow that
// created the checkpoint. This executor-only workflow rebuilds identically because its executors
// use fixed ids. Agent-based workflows must recreate each local agent with the same
// ChatClientAgentOptions.Id (and, if set, the same Name), otherwise the executor ids no longer
// match the checkpoint and resume fails.
var newWorkflow = WorkflowFactory.BuildWorkflow();
const int CheckpointIndex = 5;
Console.WriteLine($"\n\nHydrating a new workflow instance from the {CheckpointIndex + 1}th checkpoint.");
CheckpointInfo savedCheckpoint = checkpoints[CheckpointIndex];

await using StreamingRun newCheckpointedRun =
    await InProcessExecution.ResumeStreamingAsync(newWorkflow, savedCheckpoint, checkpointManager);

這很重要

傳入 ResumeStreamingAsync 的工作流程必須與建立檢查點的工作流程擁有相同的結構與執行者身份。 如果工作流程包含跨請求、相依注入範圍、程序或部署重建的本地 ChatClientAgent 實例，請為每個代理指派一個穩定 ChatClientAgentOptions.Id的 。 如果代理人也設定 ， Name也保持 Name 不變。

例如，指派一個代表代理邏輯角色的 ID：

// Give each agent a stable, unique Id so its workflow executor identity stays the same when the
// workflow is reconstructed (for example per request or dependency-injection scope), which keeps
// checkpoints resumable. If an agent also has a Name, keep that stable too, since the executor
// identity includes it. Use a fixed logical role here, not a conversation, request, or user id.
internal const string IntakeAgentName = "Assistant";
public AIAgent IntakeAgent { get; } = chatClient.AsAIAgent(new ChatClientAgentOptions
{
    Id = "intake-agent",
    Name = IntakeAgentName,
    ChatOptions = new()
    {
        Instructions =
            """
            You receive a user request and are responsible for routing to the correct initial expert agent.
            """,
    },
});

將此模式套用於所有參與工作流程的代理人。 代理 ID 必須在工作流程中唯一，且在重建同一邏輯代理時必須重複使用。 不要將對話 ID、請求 ID、使用者 ID、個人可識別資訊或秘密資料當作代理 ID。

當Name代理被設定時，目前的 .NET 工作流程執行者身份會從其 Name 和 Id兩元推導而來，因此改變任一值都會使重建的工作流程與檢查點不相容。 指派穩定值並不會修復由不同或隨機產生的 ID 所產生的檢查點;改為重新開始一場遊戲並設定檢查點血統。

相關情境請參見「 作為代理的工作流程 」與 「交接協調」。

或者，您可以從檢查點重新活化新的工作流程執行個體。

from agent_framework import WorkflowBuilder

builder = WorkflowBuilder(start_executor=start_executor)
builder.add_edge(start_executor, executor_b)
builder.add_edge(executor_b, executor_c)
builder.add_edge(executor_b, end_executor)
# This workflow instance doesn't require checkpointing enabled.
workflow = builder.build()

# Assume we want to resume from the 6th checkpoint
saved_checkpoint = checkpoints[5]
async for event in workflow.run(
    checkpoint_id=saved_checkpoint.checkpoint_id,
    checkpoint_storage=checkpoint_storage,
    stream=True,
):
    ...

或者，您可以從檢查點重新活化新的工作流程執行個體。

// Assume we want to resume from the 6th checkpoint
savedCheckpoint := checkpoints[5]
newWorkflow := buildWorkflow()

newRun, err := inproc.Default.
    WithCheckpointing(checkpointManager).
    ResumeStreaming(ctx, newWorkflow, savedCheckpoint)
if err != nil {
    return err
}
defer newRun.Close(ctx)

for evt, err := range newRun.WatchStream(ctx) {
    if err != nil {
        return err
    }
    if outputEvent, ok := evt.(workflow.OutputEvent); ok {
        fmt.Printf("Workflow completed with result: %v\n", outputEvent.Output)
    }
}

儲存執行程式狀態

若要確保在檢查點中擷取執行程式的狀態，執行程式必須覆寫 OnCheckpointingAsync 方法，並將其狀態儲存至工作流程內容。

using Microsoft.Agents.AI.Workflows;

internal sealed partial class CustomExecutor() : Executor("CustomExecutor")
{
    private const string StateKey = "CustomExecutorState";

    private List<string> messages = new();

    [MessageHandler]
    private async ValueTask HandleAsync(string message, IWorkflowContext context)
    {
        this.messages.Add(message);
        // Executor logic...
    }

    protected override ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellation = default)
    {
        return context.QueueStateUpdateAsync(StateKey, this.messages);
    }
}

此外，若要確保從檢查點繼續時正確還原狀態，執行程式必須覆寫 OnCheckpointRestoredAsync 方法，並從工作流程內容載入其狀態。

protected override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellation = default)
{
    this.messages = await context.ReadStateAsync<List<string>>(StateKey).ConfigureAwait(false);
}

為了確保執行者的狀態在檢查點中被儲存，執行者必需覆寫該 on_checkpoint_save 方法，並以字典形式回傳其狀態。

class CustomExecutor(Executor):
    def __init__(self, id: str) -> None:
        super().__init__(id=id)
        self._messages: list[str] = []

    @handler
    async def handle(self, message: str, ctx: WorkflowContext):
        self._messages.append(message)
        # Executor logic...

    async def on_checkpoint_save(self) -> dict[str, Any]:
        return {"messages": self._messages}

此外，為確保從檢查點恢復時狀態正確還原，執行者必須覆蓋該 on_checkpoint_restore 方法，並從提供的狀態字典中恢復其狀態。

async def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
    self._messages = state.get("messages", [])

為確保執行器狀態被檢查點捕捉，請將檢查點掛鉤附在執行器上，並透過工作流程上下文儲存狀態。

type customExecutor struct {
    messages []string
}

func (e *customExecutor) Handle(message string) {
    e.messages = append(e.messages, message)
}

func (e *customExecutor) OnCheckpoint(ctx *workflow.Context) error {
    return ctx.QueueStateUpdate("CustomExecutorState", "", slices.Clone(e.messages))
}

在 OnCheckpointRestoredFunc 中還原狀態：

func (e *customExecutor) OnCheckpointRestored(ctx *workflow.Context) error {
    value, err := ctx.ReadState("CustomExecutorState", "")
    if err != nil {
        return err
    }
    if value == nil {
        e.messages = nil
        return nil
    }

    messages, ok := value.([]string)
    if !ok {
        return fmt.Errorf("unexpected custom executor state type %T", value)
    }
    e.messages = slices.Clone(messages)
    return nil
}

executorState := &customExecutor{}
custom := workflow.NewExecutor("CustomExecutor", executorState).Extend(&workflow.Executor{
    OnCheckpointFunc:         executorState.OnCheckpoint,
    OnCheckpointRestoredFunc: executorState.OnCheckpointRestored,
}).Bind()

安全性考慮

這很重要

檢查點儲存是信任邊界。 無論你使用內建的儲存實作還是自訂的，儲存後端都必須被視為可信的私有基礎設施。 切勿載入來自不受信任或可能被竄改來源的檢查點。

確保檢查站使用的儲存位置得到適當的保護。 只有授權服務與使用者應有讀取或寫入檢查點資料的權限。

醃黃瓜序列化

FileCheckpointStorage 和 CosmosCheckpointStorage 都使用 Python 的 pickle 模組來序列化非 JSON 原生狀態，如資料類別、日期時間和自訂物件。 為降低反序列化過程中任意執行程式碼的風險，兩個提供者預設使用 受限的解選器 。 在反序列化過程中，僅允許內建一組安全的 Python 類型（如原語、datetime、Decimaluuid、、共用集合等）以及支援的代理框架（Agent Framework）或 OpenAI SDK 類型。 模組前綴允許清單僅為類型：輔助函數及其他非型別全域被拒絕。 任何不支援的類型都會使反序列化失敗，且 WorkflowCheckpointException。

若要允許額外的應用程式專用型別，請使用 allowed_checkpoint_types 格式透過 "module:qualname" 參數傳送：

from agent_framework import FileCheckpointStorage

storage = FileCheckpointStorage(
    "/tmp/checkpoints",
    allowed_checkpoint_types=[
        "my_app.models:SafeState",
        "my_app.models:UserProfile",
    ],
)

每個 allowed_checkpoint_types 條目必須解析為一種型別。 新增模組層級函式或其他非型別全域函式，並不會讓該全域可反序列化。

CosmosCheckpointStorage 接受相同的參數：

from azure.identity.aio import DefaultAzureCredential
from agent_framework_azure_cosmos import CosmosCheckpointStorage

storage = CosmosCheckpointStorage(
    endpoint="https://my-account.documents.azure.com:443/",
    credential=DefaultAzureCredential(),
    database_name="agent-db",
    container_name="checkpoints",
    allowed_checkpoint_types=[
        "my_app.models:SafeState",
        "my_app.models:UserProfile",
    ],
)

如果你的威脅模型完全不允許基於醃黃瓜的序列化，請使用 InMemoryCheckpointStorage 或實作帶有替代序列化策略的自訂 CheckpointStorage 工具。

儲存地點責任

FileCheckpointStorage 需要明確的 storage_path 參數——沒有預設目錄。 雖然框架能驗證路徑穿越攻擊，但儲存目錄本身的安全（檔案權限、靜態加密、存取控制）則由開發者負責。 只有授權程序才應有讀取或寫入檢查點目錄的權限。

CosmosCheckpointStorage 依賴 Azure Cosmos DB 來儲存。 盡可能使用管理身份/RBAC，將資料庫和容器範圍設定到工作流程服務，若使用基於金鑰的認證，則輪換帳戶金鑰。與檔案儲存相同，只有授權的主體應有對存放檢查點文件的 Cosmos DB 容器的讀寫權限。

Go 檢查點管理器會將檢查點狀態序列化為 JSON，但檢查點儲存仍然是受信任的應用程式狀態。 如果你使用 checkpoint.NewFileSystemJSONStore，請將檢查點檔案存放在受保護的目錄中，並限制讀寫權限僅限於授權的程序。 客製化門市需自行負責門禁控制、完整性及耐用性保證。

後續步驟

• 學習如何監控工作流程。
• 了解工作流程中的狀態隔離。
• 學習如何視覺化工作流程。