Microsoft Agent Framework ワークフロー - チェックポイント

このページでは、Microsoft Agent Framework ワークフローシステムの チェックポイント の概要について説明します。

Overview

チェックポイントを使用すると、実行中の特定のポイントでワークフローの状態を保存し、後でそれらのポイントから再開することができます。この機能は、次のシナリオで特に役立ちます。

失敗した場合の進行状況の損失を回避する、実行時間の長いワークフロー。
実行を一時停止して後で再開する、実行時間の長いワークフロー。
監査またはコンプライアンスの目的で定期的な状態の保存を必要とするワークフロー。
異なる環境またはインスタンス間で移行する必要があるワークフロー。

チェックポイントはいつ作成されますか?

ワークフロー実行モデルに記載されているように、ワークフローは スーパーステップで実行されます。チェックポイントは、そのスーパーステップ内のすべての Executor が実行を完了した後、各スーパーステップの最後に作成されます。チェックポイントは、次のようなワークフローの状態全体をキャプチャします。

すべてのエグゼキューターの現在の状態
次のスーパーステップのワークフロー内のすべての保留中のメッセージ
保留中の要求と応答
共有状態

Note

Python バージョン 1.13.0 以降では、ワークフローでは、ワークフロー入力を記録する最初のスーパーステップの前にエントリチェックポイントも作成され、要求イベントへの応答が配信されるときに別のエントリチェックポイントも作成されます。これらのチェックポイントにより、完全なワークフロー実行が再生可能になります。このリリースには、反復回数、メッセージソース ID、またはチェックポイントの順序に依存するアプリケーションに対する軽微な破壊的変更が含まれています。既存のチェックポイントは引き続きサポートされます。移行の詳細については、「Python ワークフローチェックポイントを 1.13.0 にアップグレードする」を参照してください。

チェックポイントのキャプチャ

チェックポイント処理を有効にするには、ワークフローの実行時に 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 には、次の 3 つの組み込み実装が付属しています。持続性とデプロイのニーズに合った実装を選択します。

Provider	パッケージ	Durability	最適な用途
`InMemoryCheckpointStorage`	`agent-framework`	処理中のみ	テスト、デモ、有効期間の短いワークフロー
`FileCheckpointStorage`	`agent-framework`	ローカルディスク	単一マシンワークフロー、ローカル開発
`CosmosCheckpointStorage`	`agent-framework-azure-cosmos`	Azure Cosmos DB	運用ワークフロー、分散ワークフロー、クロスプロセスワークフロー

3 つすべてが同じ CheckpointStorage プロトコルを実装するため、ワークフローまたは Executor コードを変更せずにプロバイダーをスワップできます。

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)

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)

パラメーターを使用して逆シリアル化できるPython型を制限する方法については、「allowed_checkpoint_types」セクションを参照してください。

CosmosCheckpointStorage はチェックポイントをAzure Cosmos DB NoSQLに保持します。永続的でプロセス間のチェックポイント処理を必要とする運用および分散ワークフローに最適です。オプションのプロバイダーパッケージをインストールします。

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

データベースとコンテナーは最初の使用時に自動的に作成され、/workflow_name がワークフローごとの効率的なクエリのためのパーティションキーとして使用されます。推奨される認証モードは、TokenCredential などの Azure DefaultAzureCredential を介したマネージド ID/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",
)

接続の詳細は、環境変数を使用して完全に指定することもできます。

Variable	説明
`AZURE_COSMOS_ENDPOINT`	Cosmos DB アカウントエンドポイント
`AZURE_COSMOS_DATABASE_NAME`	データベース名
`AZURE_COSMOS_CONTAINER_NAME`	コンテナー名
`AZURE_COSMOS_KEY`	アカウントキー (Azure資格情報を使用する場合は省略可能)

CosmosCheckpointStorage アプリケーションが既に Cosmos クライアントのライフサイクルを管理している場合は、事前に作成された CosmosClient ( cosmos_client= 経由) または ContainerProxy ( container_client= 経由) も受け入れます。

チェックポイント処理を有効にするには、チェックポイントマネージャーを使用して実行環境を構成します。その後、 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)
    }
}

チェックポイントからのリハイドレート

リハイドレートされたワークフローでは、チェックポイントを作成したワークフローのトポロジ ID と Executor ID を保持する必要があります。 Executor ID の解決方法は、SDK と Executor の種類によって異なります。

または、チェックポイントから新しい実行インスタンスにワークフローをリハイドレートすることもできます。

// 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);

Important

ResumeStreamingAsyncに渡されるワークフローには、チェックポイントを作成したワークフローと同じ構造と Executor ID が必要です。要求、依存関係の挿入スコープ、プロセス、またはデプロイの間で再構築されるローカル 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 ワークフロー実行プログラム ID はNameとIdの両方から派生するため、いずれかの値を変更すると、再構築されたワークフローはチェックポイントと互換性がありません。安定した値を割り当てると、異なる 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)
    }
}

Executorのステートを保存する

Executor の状態がチェックポイントに確実にキャプチャされるようにするには、Executor が 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);
    }
}

また、チェックポイントからの再開時に状態が正しく復元されるようにするには、Executor が OnCheckpointRestoredAsync メソッドをオーバーライドし、ワークフローコンテキストからその状態を読み込む必要があります。

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

Executor の状態がチェックポイントに確実にキャプチャされるようにするには、Executor が 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}

また、チェックポイントから再開するときに状態が正しく復元されるようにするには、executor は on_checkpoint_restore メソッドをオーバーライドし、指定された状態ディクショナリからその状態を復元する必要があります。

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

Executor 状態がチェックポイントでキャプチャされるようにするには、チェックポイントフックを Executor にアタッチし、ワークフローコンテキストを介して状態を格納します。

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

セキュリティに関する考慮事項

Important

チェックポイントストレージは信頼境界です。組み込みのストレージ実装を使用する場合でも、カスタム実装を使用する場合でも、ストレージバックエンドは信頼できるプライベートインフラストラクチャとして扱う必要があります。 信頼されていないソースまたは改ざんされた可能性のあるソースからチェックポイントを読み込むことはありません。

チェックポイントに使用されるストレージの場所が適切にセキュリティで保護されていることを確認します。チェックポイントデータへの読み取りまたは書き込みアクセス権を持つのは、承認されたサービスとユーザーだけです。

Pickle シリアル化

FileCheckpointStorage と CosmosCheckpointStorage では、Pythonの pickle モジュールを使用して、データクラス、datetime、カスタムオブジェクトなどの非 JSON ネイティブ状態をシリアル化します。逆シリアル化中の任意のコード実行のリスクを軽減するために、両方のプロバイダーは既定で 制限付きアンピッカー を使用します。逆シリアル化中は、組み込みの安全なPython型 (プリミティブ、datetime、uuid、Decimal、共通コレクションなど) とサポートされている 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",
    ],
)

脅威モデルで pickle ベースのシリアル化がまったく許可されていない場合は、 InMemoryCheckpointStorage を使用するか、代替のシリアル化戦略でカスタム CheckpointStorage を実装します。

ストレージの場所の責任

FileCheckpointStorage には明示的な storage_path パラメーターが必要です。既定のディレクトリはありません。フレームワークはパストラバーサル攻撃に対して検証しますが、ストレージディレクトリ自体 (ファイルのアクセス許可、保存時の暗号化、アクセス制御) をセキュリティで保護することは開発者の責任です。チェックポイントディレクトリへの読み取りまたは書き込みアクセス権を持つのは、承認されたプロセスだけです。

CosmosCheckpointStorage は、ストレージのAzure Cosmos DBに依存します。可能な場合はマネージド ID/RBAC を使用し、データベースとコンテナーのスコープをワークフローサービスに設定し、キーベースの認証を使用する場合はアカウントキーをローテーションします。ファイルストレージと同様に、チェックポイントドキュメントを保持する Cosmos DB コンテナーへの読み取りまたは書き込みアクセス権を持つのは、承認されたプリンシパルのみです。

Go チェックポイントマネージャーはチェックポイントの状態を JSON としてシリアル化しますが、チェックポイントストレージは引き続き信頼されたアプリケーションの状態です。 checkpoint.NewFileSystemJSONStoreを使用する場合は、チェックポイントファイルを保護されたディレクトリに格納し、読み取り/書き込みアクセスを承認されたプロセスのみに制限します。カスタムストアは、独自のアクセス制御、整合性、持続性の保証を担当します。

次のステップ

フィードバック

このページはお役に立ちましたか?

Last updated on 2026-08-27