Microsoft 에이전트 프레임워크 워크플로 - 검사점

검사점을 활성화하려면 워크플로를 실행할 때 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는 내구성 및 배포 요구 사항과 일치하는 구현을 선택하는 세 가지 기본 제공 구현을 제공합니다.

세 가지 모두 동일한 CheckpointStorage 프로토콜을 구현하므로 워크플로 또는 실행기 코드를 변경하지 않고도 공급자를 교환할 수 있습니다.

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)

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)

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

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)

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",
)

동일한 실행에서 특정 검사점에서 직접 워크플로를 다시 시작할 수 있습니다.

// 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
CheckpointInfo savedCheckpoint = run.Checkpoints[5];
StreamingRun newRun = await InProcessExecution
    .ResumeStreamingAsync(newWorkflow, savedCheckpoint, checkpointManager)
    .ConfigureAwait(false);
await foreach (WorkflowEvent evt in newRun.WatchStreamAsync().ConfigureAwait(false))
{
    if (evt is WorkflowOutputEvent workflowOutputEvt)
    {
        Console.WriteLine($"Workflow completed with result: {workflowOutputEvt.Data}");
    }
}

새 워크플로 인스턴스를 검사점을 통해 리하이드레이트할 수 있습니다.

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,
):
    ...

실행기의 상태가 검사점에서 캡처되도록 하려면 실행기가 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", [])

검사점에서 사용되는 스토리지 위치가 적절하게 보호되었는지 확인합니다. 권한 있는 서비스 및 사용자만 검사점 데이터에 대한 읽기 또는 쓰기 액세스 권한이 있어야 합니다.

피클 시리얼라이제이션

FileCheckpointStorage 및 CosmosCheckpointStorage 모두 Python pickle 모듈을 사용하여 데이터 클래스, datetimes 및 사용자 지정 개체와 같은 비 JSON 네이티브 상태를 직렬화합니다. 역직렬화하는 동안 임의 코드 실행의 위험을 완화하기 위해 두 공급자는 기본적으로 제한적 역직렬화 프로그램을 사용합니다. 안전한 Python 형식(기본 형식, datetime, uuid, Decimal, 공통 컬렉션 등)의 기본 제공 집합과 모든 agent_framework 내부 형식은 역직렬화 중에 허용됩니다. 다른 형식이 검사점에서 발견되면, 이로 인해 WorkflowCheckpointException와 함께 역직렬화가 실패합니다.

추가 애플리케이션 관련 형식을 허용하려면 "module:qualname" 형식을 사용하여 allowed_checkpoint_types 매개 변수를 전달합니다.

from agent_framework import FileCheckpointStorage

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

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 사용합니다. 가능한 경우 관리 ID/RBAC를 사용하고, 데이터베이스 및 컨테이너의 범위를 워크플로 서비스로 지정하고, 키 기반 인증을 사용하는 경우 계정 키를 회전합니다. 파일 스토리지와 마찬가지로 권한 있는 보안 주체만 검사점 문서를 보유하는 Cosmos DB 컨테이너에 대한 읽기 또는 쓰기 액세스 권한이 있어야 합니다.